From f87c8730279299703f5f80c58798db51a7050a13 Mon Sep 17 00:00:00 2001 From: busya Date: Wed, 10 Jun 2026 14:57:18 +0300 Subject: [PATCH] test(app): add split app.py tests and extend migration_engine coverage - app.py split from 1966-line test_app.py into 7 files (68 tests, 92% coverage): - test_app_lifespan: lifespan, ensure_initial_admin_user - test_app_handlers: exception handlers, HSTS - test_app_middleware: log_requests, middleware chain - test_app_ws_auth: _authenticate_websocket - test_app_ws_endpoint: WS main loop, 5 endpoint handlers - test_app_ws_events: task/maintenance/dataset/translate WS streams - test_app_spa: SPA serving, TestClient integration - migration_engine: extended coverage with init, edge cases, error paths --- backend/tests/core/test_migration_engine.py | 277 ++++++++++++++++- backend/tests/test_app_handlers.py | 91 ++++++ backend/tests/test_app_lifespan.py | 228 ++++++++++++++ backend/tests/test_app_middleware.py | 193 ++++++++++++ backend/tests/test_app_spa.py | 113 +++++++ backend/tests/test_app_ws_auth.py | 120 ++++++++ backend/tests/test_app_ws_endpoint.py | 317 ++++++++++++++++++++ backend/tests/test_app_ws_events.py | 183 +++++++++++ 8 files changed, 1518 insertions(+), 4 deletions(-) create mode 100644 backend/tests/test_app_handlers.py create mode 100644 backend/tests/test_app_lifespan.py create mode 100644 backend/tests/test_app_middleware.py create mode 100644 backend/tests/test_app_spa.py create mode 100644 backend/tests/test_app_ws_auth.py create mode 100644 backend/tests/test_app_ws_endpoint.py create mode 100644 backend/tests/test_app_ws_events.py diff --git a/backend/tests/core/test_migration_engine.py b/backend/tests/core/test_migration_engine.py index 45834c59..de2241d0 100644 --- a/backend/tests/core/test_migration_engine.py +++ b/backend/tests/core/test_migration_engine.py @@ -350,11 +350,280 @@ def test_transform_zip_invalid_path(): def test_transform_yaml_nonexistent_file(): """@PRE: Verify behavior on non-existent YAML file.""" engine = MigrationEngine() - # Should log error and not crash (implemented via try-except if wrapped, - # but _transform_yaml itself might raise FileNotFoundError if not guarded) with pytest.raises(FileNotFoundError): engine._transform_yaml(Path("non_existent.yaml"), {}) - - # #endregion test_transform_yaml_nonexistent_file + + +# ── [NEW] Additional coverage: init, edge cases, error paths ── + + +# #region test_init_belief_scope [C:2] [TYPE Function] +# @BRIEF MigrationEngine.__init__ logs reason/reflect via belief_scope. +def test_init_belief_scope(): + """Engine init with mapping_service logs initialization.""" + svc = MockMappingService({}) + engine = MigrationEngine(svc) + assert engine.mapping_service is svc +# #endregion test_init_belief_scope + + +# #region test_init_no_mapping_service [C:2] [TYPE Function] +# @BRIEF MigrationEngine can be initialized without a mapping_service. +def test_init_no_mapping_service(): + """Engine init without arguments leaves mapping_service as None.""" + engine = MigrationEngine() + assert engine.mapping_service is None +# #endregion test_init_no_mapping_service + + +# #region test_transform_yaml_empty_data [C:2] [TYPE Function] +# @BRIEF _transform_yaml with empty YAML (null data) returns early without error. +def test_transform_yaml_empty_data(): + """Empty YAML file does not raise.""" + engine = MigrationEngine() + with tempfile.TemporaryDirectory() as td: + fp = Path(td) / "empty.yaml" + fp.write_text("~\n") # null in YAML + engine._transform_yaml(fp, {"src": "tgt"}) # should not raise +# #endregion test_transform_yaml_empty_data + + +# #region test_transform_yaml_no_matching_uuid [C:2] [TYPE Function] +# @BRIEF _transform_yaml leaves file unchanged when uuid not in db_mapping. +def test_transform_yaml_no_matching_uuid(): + """Non-mapped database_uuid leaves YAML content unchanged.""" + engine = MigrationEngine() + with tempfile.TemporaryDirectory() as td: + fp = Path(td) / "ds.yaml" + with open(fp, "w") as f: + yaml.dump({"database_uuid": "uuid-alpha", "table_name": "users"}, f) + engine._transform_yaml(fp, {"uuid-other": "target-uuid"}) + with open(fp) as f: + data = yaml.safe_load(f) + assert data["database_uuid"] == "uuid-alpha" # unchanged +# #endregion test_transform_yaml_no_matching_uuid + + +# #region test_transform_zip_extract_failure [C:2] [TYPE Function] +# @BRIEF When the source ZIP is corrupted, transform_zip returns False. +def test_transform_zip_extract_failure(): + """Corrupt ZIP path returns False.""" + engine = MigrationEngine() + with tempfile.TemporaryDirectory() as td: + bad_zip = Path(td) / "bad.zip" + bad_zip.write_text("this is not a zip file") + result = engine.transform_zip(str(bad_zip), str(Path(td) / "out.zip"), {}) + assert result is False +# #endregion test_transform_zip_extract_failure + + +# #region test_transform_zip_strip_databases [C:2] [TYPE Function] +# @BRIEF When strip_databases=True, the databases/ directory is excluded from output. +def test_transform_zip_strip_databases(): + """Databases directory is stripped when strip_databases=True.""" + engine = MigrationEngine() + with tempfile.TemporaryDirectory() as td: + src = Path(td) / "src" + dst = Path(td) / "out.zip" + + # Create archive with databases/ and datasets/ + (src / "databases").mkdir(parents=True) + (src / "datasets").mkdir() + (src / "databases" / "db.yaml").write_text("db: test") + (src / "datasets" / "ds.yaml").write_text("table: users") + + zip_path = src / "source.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + for root, _, files in os.walk(src): + for f in files: + fp = Path(root) / f + if fp.suffix == ".zip": + continue + zf.write(fp, fp.relative_to(src)) + + success = engine.transform_zip( + str(zip_path), str(dst), {}, strip_databases=True, + ) + assert success is True + + # Verify databases/ is not in output + with tempfile.TemporaryDirectory() as out_dir: + with zipfile.ZipFile(dst, "r") as zf: + names = zf.namelist() + assert not any("databases" in n for n in names) + assert any("datasets" in n for n in names) +# #endregion test_transform_zip_strip_databases + + +# #region test_transform_zip_cross_filters_no_mapping_service [C:2] [TYPE Function] +# @BRIEF fix_cross_filters=True without mapping_service logs warning, doesn't crash. +def test_transform_zip_cross_filters_no_mapping_service(): + """Cross-filter fix requested but no mapping service — graceful skip.""" + engine = MigrationEngine() # no mapping_service + with tempfile.TemporaryDirectory() as td: + src = Path(td) / "src" + dst = Path(td) / "out.zip" + + (src / "datasets").mkdir(parents=True) + (src / "datasets" / "ds.yaml").write_text("table: users") + + zip_path = src / "source.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.write(src / "datasets" / "ds.yaml", "datasets/ds.yaml") + + success = engine.transform_zip( + str(zip_path), str(dst), {"old": "new"}, + target_env_id="env-1", fix_cross_filters=True, + ) + assert success is True +# #endregion test_transform_zip_cross_filters_no_mapping_service + + +# #region test_extract_chart_uuids_malformed_yaml [C:2] [TYPE Function] +# @BRIEF Malformed chart YAML files are skipped without crashing extraction. +def test_extract_chart_uuids_malformed_yaml(): + """Broken chart YAML files do not break the extraction loop.""" + engine = MigrationEngine() + with tempfile.TemporaryDirectory() as td: + charts_dir = Path(td) / "charts" + charts_dir.mkdir() + + (charts_dir / "good.yaml").write_text("id: 1\nuuid: abc-123\n") + (charts_dir / "broken.yaml").write_text("{invalid: yaml: [}") # parse error + (charts_dir / "no_id.yaml").write_text("uuid: def-456\n") # missing id + (charts_dir / "empty.yaml").write_text("") # empty + + result = engine._extract_chart_uuids_from_archive(Path(td)) + assert 1 in result + assert result[1] == "abc-123" + assert len(result) == 1 # only the valid one +# #endregion test_extract_chart_uuids_malformed_yaml + + +# #region test_patch_dashboard_missing_file [C:2] [TYPE Function] +# @BRIEF _patch_dashboard_metadata with non-existent file returns early. +def test_patch_dashboard_missing_file(): + """Non-existent dashboard YAML is silently skipped.""" + engine = MigrationEngine(MockMappingService({})) + engine._patch_dashboard_metadata(Path("/nonexistent/dash.yaml"), "env-1", {}) + # Should not raise +# #endregion test_patch_dashboard_missing_file + + +# #region test_patch_dashboard_empty_json_metadata [C:2] [TYPE Function] +# @BRIEF When json_metadata is empty string, patch returns early. +def test_patch_dashboard_empty_json_metadata(): + """Empty json_metadata string is handled gracefully.""" + engine = MigrationEngine(MockMappingService({})) + with tempfile.TemporaryDirectory() as td: + fp = Path(td) / "dash.yaml" + with open(fp, "w") as f: + yaml.dump({"json_metadata": ""}, f) + engine._patch_dashboard_metadata(fp, "env-1", {42: "uuid-42"}) + with open(fp) as f: + data = yaml.safe_load(f) + assert data["json_metadata"] == "" # unchanged +# #endregion test_patch_dashboard_empty_json_metadata + + +# #region test_patch_dashboard_no_target_ids [C:2] [TYPE Function] +# @BRIEF When mapping_service returns no target IDs, patch returns early. +def test_patch_dashboard_no_target_ids(): + """No remote target IDs — metadata is left unchanged.""" + empty_service = MockMappingService({}) # no mappings + engine = MigrationEngine(empty_service) + with tempfile.TemporaryDirectory() as td: + fp = Path(td) / "dash.yaml" + metadata = {"native_filter_configuration": [{"targets": [{"chartId": 42}]}]} + with open(fp, "w") as f: + yaml.dump({"json_metadata": json.dumps(metadata)}, f) + engine._patch_dashboard_metadata(fp, "env-1", {42: "uuid-42"}) + with open(fp) as f: + data = yaml.safe_load(f) + assert json.loads(data["json_metadata"]) == metadata # unchanged +# #endregion test_patch_dashboard_no_target_ids + + +# #region test_patch_dashboard_no_source_match [C:2] [TYPE Function] +# @BRIEF When source IDs don't match any remote IDs, patch returns early. +def test_patch_dashboard_no_source_match(): + """Source IDs not found in remote mapping — metadata unchanged.""" + service = MockMappingService({"uuid-other": 999}) # different UUID + engine = MigrationEngine(service) + with tempfile.TemporaryDirectory() as td: + fp = Path(td) / "dash.yaml" + metadata_json = json.dumps({"native_filter_configuration": [{"targets": [{"chartId": 42}]}]}) + with open(fp, "w") as f: + yaml.dump({"json_metadata": metadata_json}, f) + engine._patch_dashboard_metadata(fp, "env-1", {42: "uuid-42"}) + with open(fp) as f: + data = yaml.safe_load(f) + assert data["json_metadata"] == metadata_json # unchanged +# #endregion test_patch_dashboard_no_source_match + + +# #region test_patch_dashboard_exception_handling [C:2] [TYPE Function] +# @BRIEF Exception in _patch_dashboard_metadata is caught and logged, not raised. +def test_patch_dashboard_exception_handling(): + """Corrupt json_metadata that fails to parse — exception is caught, not raised.""" + engine = MigrationEngine(MockMappingService({"uuid-42": 100})) + with tempfile.TemporaryDirectory() as td: + fp = Path(td) / "dash.yaml" + # Invalid JSON as json_metadata — will fail on json.loads() + with open(fp, "w") as f: + yaml.dump({"json_metadata": "{invalid json]"}, f) + # Should not raise + engine._patch_dashboard_metadata(fp, "env-1", {42: "uuid-42"}) +# #endregion test_patch_dashboard_exception_handling + + +# #region test_patch_dashboard_source_ids_not_in_remote [C:2] [TYPE Function] +# @BRIEF When target_ids exist but none match source_map UUIDs, patch returns early at line 237. +def test_patch_dashboard_source_ids_not_in_remote(): + """target_ids has entries, but none match source UUIDs → source_to_target empty → skip.""" + class MockServiceReturnsUnrelated(MockMappingService): + """get_remote_ids_batch returns extra entries NOT in the requested UUIDs.""" + def get_remote_ids_batch(self, env_id, resource_type, uuids): + # Returns mappings for UUIDs NOT in the request + return {"unrelated-uuid": 999} + + engine = MigrationEngine(MockServiceReturnsUnrelated({})) + with tempfile.TemporaryDirectory() as td: + fp = Path(td) / "dash.yaml" + metadata = {"native_filter_configuration": [{"targets": [{"chartId": 42}]}]} + with open(fp, "w") as f: + yaml.dump({"json_metadata": json.dumps(metadata)}, f) + engine._patch_dashboard_metadata(fp, "env-1", {42: "uuid-42"}) + with open(fp) as f: + data = yaml.safe_load(f) + assert json.loads(data["json_metadata"]) == metadata # unchanged +# #endregion test_patch_dashboard_source_ids_not_in_remote + + +# #region test_transform_zip_without_fix_cross_filters [C:2] [TYPE Function] +# @BRIEF transform_zip works without fixing cross-filters (default). +def test_transform_zip_without_fix_cross_filters(): + """Default fix_cross_filters=False skips dashboard patching.""" + engine = MigrationEngine(MockMappingService({})) + with tempfile.TemporaryDirectory() as td: + src = Path(td) / "src" + dst = Path(td) / "out.zip" + (src / "datasets").mkdir(parents=True) + (src / "datasets" / "ds.yaml").write_text(yaml.dump({"database_uuid": "old-uuid", "table_name": "t"})) + zip_path = src / "source.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + zf.write(src / "datasets" / "ds.yaml", "datasets/ds.yaml") + + success = engine.transform_zip(str(zip_path), str(dst), {"old-uuid": "new-uuid"}) + assert success is True + # Verify dataset was transformed + with tempfile.TemporaryDirectory() as out_dir: + with zipfile.ZipFile(dst, "r") as zf: + zf.extractall(out_dir) + with open(Path(out_dir) / "datasets" / "ds.yaml") as f: + data = yaml.safe_load(f) + assert data["database_uuid"] == "new-uuid" +# #endregion test_transform_zip_without_fix_cross_filters + # #endregion TestMigrationEngine diff --git a/backend/tests/test_app_handlers.py b/backend/tests/test_app_handlers.py new file mode 100644 index 00000000..a389fe96 --- /dev/null +++ b/backend/tests/test_app_handlers.py @@ -0,0 +1,91 @@ +# #region Test.AppModule.Handlers [C:3] [TYPE Module] [SEMANTICS test,app,handlers,exceptions] +# @BRIEF Tests for app.py — global_exception_handler, network_error_handler. +# @RELATION BINDS_TO -> [AppModule] +# @TEST_EDGE: unhandled_exception -> returns 500 JSON with path +# @TEST_EDGE: network_error -> returns 503 +# @TEST_INVARIANT: every_500_logged -> VERIFIED_BY: test_global_exception_handler_logs_and_returns_500 + +from pathlib import Path +import sys +import json + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import pytest +from unittest.mock import MagicMock +from fastapi import Request, HTTPException +from fastapi.responses import JSONResponse + + +# #region _make_mock_request [C:1] [TYPE Function] +def _make_mock_request(method="GET", path="/api/test", client_host="127.0.0.1"): + req = MagicMock(spec=Request) + req.method = method + req.url = MagicMock() + req.url.path = path + req.client = MagicMock() + req.client.host = client_host + req.query_params = {} + return req +# #endregion _make_mock_request + + +class TestExceptionHandlers: + """global_exception_handler and network_error_handler.""" + + # #region test_global_handler_500 [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_global_handler_500(self): + from src.app import global_exception_handler + request = _make_mock_request(method="POST", path="/api/break") + response = await global_exception_handler(request, ValueError("broke")) + assert isinstance(response, JSONResponse) + assert response.status_code == 500 + body = json.loads(response.body) + assert body["detail"] == "Internal server error" + assert body["path"] == "/api/break" + # #endregion test_global_handler_500 + + # #region test_global_handler_unknown_client [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_global_handler_unknown_client(self): + from src.app import global_exception_handler + req = _make_mock_request() + req.client = None + response = await global_exception_handler(req, RuntimeError("no client")) + assert response.status_code == 500 + # #endregion test_global_handler_unknown_client + + # #region test_network_error_handler [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_network_error_handler(self): + from src.app import network_error_handler + from src.core.utils.network import NetworkError + request = _make_mock_request() + response = await network_error_handler(request, NetworkError("down")) + assert isinstance(response, HTTPException) + assert response.status_code == 503 + assert "Environment unavailable" in response.detail + # #endregion test_network_error_handler + + # #region test_global_handler_with_query_params [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_global_handler_with_query_params(self): + from src.app import global_exception_handler + req = _make_mock_request() + req.query_params = {"env_id": "prod"} + response = await global_exception_handler(req, Exception("test")) + assert response.status_code == 500 + # #endregion test_global_handler_with_query_params + + # #region test_internal_server_error_returns_json [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_internal_server_error_returns_json(self): + from src.app import global_exception_handler + request = _make_mock_request(method="POST", path="/api/crash") + response = await global_exception_handler(request, RuntimeError("crash")) + assert response.status_code == 500 + body = json.loads(response.body) + assert body["path"] == "/api/crash" + # #endregion test_internal_server_error_returns_json +# #endregion Test.AppModule.Handlers diff --git a/backend/tests/test_app_lifespan.py b/backend/tests/test_app_lifespan.py new file mode 100644 index 00000000..f7f90c7e --- /dev/null +++ b/backend/tests/test_app_lifespan.py @@ -0,0 +1,228 @@ +# #region Test.AppModule.Lifespan [C:3] [TYPE Module] [SEMANTICS test,app,lifespan,admin,bootstrap] +# @BRIEF Tests for app.py — ensure_initial_admin_user + lifespan startup/shutdown. +# @RELATION BINDS_TO -> [AppModule] +# @TEST_EDGE: missing_env_flag -> ensure_initial_admin_user skips when INITIAL_ADMIN_CREATE is off +# @TEST_EDGE: missing_admin_creds -> logs warning when flag on but username/password missing +# @TEST_EDGE: existing_admin_user -> skips creation when user already exists +# @TEST_EDGE: admin_role_creation -> creates Admin role if missing +# @TEST_EDGE: db_rollback_on_error -> rolls back DB transaction on exception +# @TEST_INVARIANT: admin_bootstrap_guard -> VERIFIED_BY: test_ensure_initial_admin_user_flag_off, test_ensure_initial_admin_user_missing_creds + +from pathlib import Path +import sys +import os + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import pytest +from unittest.mock import AsyncMock, MagicMock, patch +from fastapi import FastAPI + + +# ============================================================================= +# ensure_initial_admin_user +# ============================================================================= + +class TestEnsureInitialAdminUser: + """ensure_initial_admin_user — environment-gated admin bootstrap.""" + + # #region test_flag_off [C:2] [TYPE Function] + @patch.dict(os.environ, {"INITIAL_ADMIN_CREATE": "false"}, clear=False) + def test_flag_off(self): + from src.app import ensure_initial_admin_user + result = ensure_initial_admin_user() + assert result is None + # #endregion test_flag_off + + # #region test_flag_on_missing_creds [C:2] [TYPE Function] + @patch.dict(os.environ, { + "INITIAL_ADMIN_CREATE": "true", "INITIAL_ADMIN_USERNAME": "", "INITIAL_ADMIN_PASSWORD": "", + }, clear=False) + def test_flag_on_missing_creds(self): + from src.app import ensure_initial_admin_user + assert ensure_initial_admin_user() is None + # #endregion test_flag_on_missing_creds + + # #region test_flag_on_only_username [C:2] [TYPE Function] + @patch.dict(os.environ, { + "INITIAL_ADMIN_CREATE": "1", "INITIAL_ADMIN_USERNAME": "admin", "INITIAL_ADMIN_PASSWORD": "", + }, clear=False) + def test_flag_on_only_username(self): + from src.app import ensure_initial_admin_user + assert ensure_initial_admin_user() is None + # #endregion test_flag_on_only_username + + # #region test_creates_admin_role_and_user [C:2] [TYPE Function] + @patch.dict(os.environ, { + "INITIAL_ADMIN_CREATE": "true", "INITIAL_ADMIN_USERNAME": "bootstrap-admin", "INITIAL_ADMIN_PASSWORD": "s3cret!", + }, clear=False) + def test_creates_admin_role_and_user(self): + mock_db = MagicMock() + mock_db.query.return_value.filter.return_value.first.side_effect = [None, None] + mock_db.add = MagicMock() + with patch("src.app.AuthSessionLocal", return_value=mock_db): + from src.app import ensure_initial_admin_user + ensure_initial_admin_user() + role_calls = [c for c in mock_db.add.call_args_list if c.args[0].__class__.__name__ == "Role"] + user_calls = [c for c in mock_db.add.call_args_list if c.args[0].__class__.__name__ == "User"] + assert len(role_calls) == 1 + assert role_calls[0].args[0].name == "Admin" + assert len(user_calls) == 1 + assert user_calls[0].args[0].username == "bootstrap-admin" + assert mock_db.commit.call_count == 2 + # #endregion test_creates_admin_role_and_user + + # #region test_skips_existing_user [C:2] [TYPE Function] + @patch.dict(os.environ, { + "INITIAL_ADMIN_CREATE": "true", "INITIAL_ADMIN_USERNAME": "existing-admin", "INITIAL_ADMIN_PASSWORD": "s3cret!", + }, clear=False) + def test_skips_existing_user(self): + mock_db = MagicMock() + mock_db.query.return_value.filter.return_value.first.side_effect = [ + MagicMock(name="Admin"), MagicMock(username="existing-admin"), + ] + with patch("src.app.AuthSessionLocal", return_value=mock_db): + from src.app import ensure_initial_admin_user + ensure_initial_admin_user() + mock_db.add.assert_not_called() + mock_db.commit.assert_not_called() + # #endregion test_skips_existing_user + + # #region test_rollback_on_error [C:2] [TYPE Function] + @patch.dict(os.environ, { + "INITIAL_ADMIN_CREATE": "true", "INITIAL_ADMIN_USERNAME": "admin", "INITIAL_ADMIN_PASSWORD": "s3cret!", + }, clear=False) + def test_rollback_on_error(self): + mock_db = MagicMock() + mock_db.query.return_value.filter.return_value.first.side_effect = Exception("DB crash") + with patch("src.app.AuthSessionLocal", return_value=mock_db): + from src.app import ensure_initial_admin_user + with pytest.raises(Exception, match="DB crash"): + ensure_initial_admin_user() + mock_db.rollback.assert_called_once() + mock_db.close.assert_called_once() + # #endregion test_rollback_on_error + + # #region test_reuses_existing_admin_role [C:2] [TYPE Function] + @patch.dict(os.environ, { + "INITIAL_ADMIN_CREATE": "true", "INITIAL_ADMIN_USERNAME": "new-user", "INITIAL_ADMIN_PASSWORD": "s3cret!", + }, clear=False) + def test_reuses_existing_admin_role(self): + mock_db = MagicMock() + mock_db.query.return_value.filter.return_value.first.side_effect = [MagicMock(name="Admin"), None] + with patch("src.app.AuthSessionLocal", return_value=mock_db): + from src.app import ensure_initial_admin_user + ensure_initial_admin_user() + user_calls = [c for c in mock_db.add.call_args_list if c.args[0].__class__.__name__ == "User"] + role_calls = [c for c in mock_db.add.call_args_list if c.args[0].__class__.__name__ == "Role"] + assert len(user_calls) == 1 + assert len(role_calls) == 0 + mock_db.commit.assert_called_once() + # #endregion test_reuses_existing_admin_role + + +# ============================================================================= +# lifespan +# ============================================================================= + +class TestLifespan: + """Lifespan — startup/shutdown lifecycle.""" + + # #region test_lifespan_startup_initializes_deps [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_lifespan_startup_initializes_deps(self): + from src.app import lifespan + mock_app = MagicMock(spec=FastAPI) + with ( + patch("src.app.seed_trace_id") as mock_seed, + patch("src.app.ensure_encryption_key") as mock_enc_key, + patch("src.app.init_db") as mock_init_db, + patch("src.app.ensure_initial_admin_user") as mock_admin, + patch("src.app.get_scheduler_service") as mock_get_sched, + ): + mock_scheduler = MagicMock() + mock_get_sched.return_value = mock_scheduler + async with lifespan(mock_app): + pass + mock_seed.assert_called_once() + mock_enc_key.assert_called_once() + mock_init_db.assert_called_once() + mock_admin.assert_called_once() + mock_scheduler.start.assert_called_once() + mock_scheduler.stop.assert_called_once() + # #endregion test_lifespan_startup_initializes_deps + + # #region test_lifespan_stuck_run_cleanup [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_lifespan_stuck_run_cleanup(self): + from src.app import lifespan + mock_app = MagicMock(spec=FastAPI) + mock_stuck = MagicMock() + mock_stuck.id = "run-stuck-1" + mock_stuck.status = "running" + mock_db = MagicMock() + mock_db.query.return_value.filter.return_value.all.return_value = [mock_stuck] + with ( + patch("src.app.seed_trace_id"), patch("src.app.ensure_encryption_key"), + patch("src.app.init_db"), patch("src.app.ensure_initial_admin_user"), + patch("src.app.get_scheduler_service"), + patch("src.core.database.SessionLocal", return_value=mock_db), + ): + async with lifespan(mock_app): + pass + assert mock_stuck.status == "FAIL" + assert "Force-stopped" in mock_stuck.summary + mock_db.commit.assert_called() + mock_db.close.assert_called() + # #endregion test_lifespan_stuck_run_cleanup + + # #region test_lifespan_stuck_cleanup_handles_exception [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_lifespan_stuck_cleanup_handles_exception(self): + from src.app import lifespan + mock_app = MagicMock(spec=FastAPI) + with ( + patch("src.app.seed_trace_id"), patch("src.app.ensure_encryption_key"), + patch("src.app.init_db"), patch("src.app.ensure_initial_admin_user"), + patch("src.app.get_scheduler_service"), + patch("src.core.database.SessionLocal", side_effect=Exception("DB unreachable")), + ): + async with lifespan(mock_app): + pass + # #endregion test_lifespan_stuck_cleanup_handles_exception + + # #region test_lifespan_stuck_no_runs [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_lifespan_stuck_no_runs(self): + from src.app import lifespan + mock_app = MagicMock(spec=FastAPI) + mock_db = MagicMock() + mock_db.query.return_value.filter.return_value.all.return_value = [] + with ( + patch("src.app.seed_trace_id"), patch("src.app.ensure_encryption_key"), + patch("src.app.init_db"), patch("src.app.ensure_initial_admin_user"), + patch("src.app.get_scheduler_service"), + patch("src.core.database.SessionLocal", return_value=mock_db), + ): + async with lifespan(mock_app): + pass + mock_db.commit.assert_called_once() + mock_db.close.assert_called() + # #endregion test_lifespan_stuck_no_runs + + # #region test_lifespan_shutdown_stops_scheduler [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_lifespan_shutdown_stops_scheduler(self): + from src.app import lifespan + mock_app = MagicMock(spec=FastAPI) + mock_scheduler = MagicMock() + with ( + patch("src.app.seed_trace_id"), patch("src.app.ensure_encryption_key"), + patch("src.app.init_db"), patch("src.app.ensure_initial_admin_user"), + patch("src.app.get_scheduler_service", return_value=mock_scheduler), + ): + async with lifespan(mock_app): + pass + mock_scheduler.stop.assert_called_once() + # #endregion test_lifespan_shutdown_stops_scheduler +# #endregion Test.AppModule.Lifespan diff --git a/backend/tests/test_app_middleware.py b/backend/tests/test_app_middleware.py new file mode 100644 index 00000000..9ee4b479 --- /dev/null +++ b/backend/tests/test_app_middleware.py @@ -0,0 +1,193 @@ +# #region Test.AppModule.Middleware [C:3] [TYPE Module] [SEMANTICS test,app,middleware,hsts,cors,logging] +# @BRIEF Tests for app.py — middleware: HSTS, log_requests, CORS, Session, TraceContext, router registration. +# @RELATION BINDS_TO -> [AppModule] +# @TEST_EDGE: hsts_enabled -> header set when FORCE_HTTPS=true +# @TEST_EDGE: hsts_disabled -> no header when FORCE_HTTPS=false +# @TEST_EDGE: polling_endpoint_no_log -> polling /api/tasks GET does not log +# @TEST_INVARIANT: hsts_gate -> VERIFIED_BY: test_hsts_middleware_enabled, test_hsts_middleware_disabled + +from pathlib import Path +import sys +import os + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import pytest +from unittest.mock import AsyncMock, MagicMock, patch +from fastapi import Request, HTTPException + + +# #region _make_mock_request [C:1] [TYPE Function] +def _make_mock_request(method="GET", path="/api/test", client_host="127.0.0.1"): + req = MagicMock(spec=Request) + req.method = method + req.url = MagicMock() + req.url.path = path + req.client = MagicMock() + req.client.host = client_host + req.query_params = {} + return req +# #endregion _make_mock_request + + +class TestHSTSMiddleware: + """HSTSMiddleware — Strict-Transport-Security header.""" + + # #region test_hsts_enabled [C:2] [TYPE Function] + @patch.dict(os.environ, {"FORCE_HTTPS": "true"}, clear=False) + @pytest.mark.asyncio + async def test_hsts_enabled(self): + from src.app import HSTSMiddleware + async def call_next(r): + resp = MagicMock(); resp.headers = {}; return resp + mw = HSTSMiddleware(MagicMock()) + resp = await mw.dispatch(_make_mock_request(), call_next) + assert resp.headers.get("Strict-Transport-Security") == "max-age=31536000; includeSubDomains" + # #endregion test_hsts_enabled + + # #region test_hsts_disabled [C:2] [TYPE Function] + @patch.dict(os.environ, {"FORCE_HTTPS": "false"}, clear=False) + @pytest.mark.asyncio + async def test_hsts_disabled(self): + from src.app import HSTSMiddleware + async def call_next(r): + resp = MagicMock(); resp.headers = {}; return resp + mw = HSTSMiddleware(MagicMock()) + assert mw._enabled is False + resp = await mw.dispatch(_make_mock_request(), call_next) + assert "Strict-Transport-Security" not in resp.headers + # #endregion test_hsts_disabled + + # #region test_hsts_empty_env [C:2] [TYPE Function] + @patch.dict(os.environ, {}, clear=False) + def test_hsts_empty_env(self): + os.environ.pop("FORCE_HTTPS", None) + from src.app import HSTSMiddleware + mw = HSTSMiddleware(MagicMock()) + assert mw._enabled is False + # #endregion test_hsts_empty_env + + +class TestLogRequests: + """log_requests — HTTP request/response logging.""" + + # #region test_log_non_polling [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_log_non_polling(self): + from src.app import log_requests + async def call_next(r): + resp = MagicMock(); resp.status_code = 200; return resp + resp = await log_requests(_make_mock_request(path="/api/dashboards"), call_next) + assert resp.status_code == 200 + # #endregion test_log_non_polling + + # #region test_log_polling_skipped [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_log_polling_skipped(self): + from src.app import log_requests + async def call_next(r): + resp = MagicMock(); resp.status_code = 200; return resp + resp = await log_requests(_make_mock_request(method="GET", path="/api/tasks"), call_next) + assert resp.status_code == 200 + # #endregion test_log_polling_skipped + + # #region test_log_post_polling_not_skipped [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_log_post_polling_not_skipped(self): + from src.app import log_requests + async def call_next(r): + resp = MagicMock(); resp.status_code = 201; return resp + resp = await log_requests(_make_mock_request(method="POST", path="/api/tasks"), call_next) + assert resp.status_code == 201 + # #endregion test_log_post_polling_not_skipped + + # #region test_log_network_error [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_log_network_error(self): + from src.app import log_requests + from src.core.utils.network import NetworkError + async def call_next(r): + raise NetworkError("timeout") + with pytest.raises(HTTPException) as e: + await log_requests(_make_mock_request(path="/api/dashboards"), call_next) + assert e.value.status_code == 503 + # #endregion test_log_network_error + + +class TestLogRequestsStderrFallback: + """log_requests — stderr JSON dump when logger level > INFO.""" + + # #region test_stderr_fallback_req [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_stderr_fallback_req(self): + from src.app import log_requests + async def call_next(r): + resp = MagicMock(); resp.status_code = 200; return resp + with patch("src.app.logger.isEnabledFor", return_value=False): + resp = await log_requests(_make_mock_request(path="/api/test-stderr"), call_next) + assert resp.status_code == 200 + # #endregion test_stderr_fallback_req + + # #region test_stderr_fallback_resp [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_stderr_fallback_resp(self): + from src.app import log_requests + async def call_next(r): + resp = MagicMock(); resp.status_code = 500; return resp + with patch("src.app.logger.isEnabledFor", return_value=False): + resp = await log_requests(_make_mock_request(path="/api/test-stderr-resp"), call_next) + assert resp.status_code == 500 + # #endregion test_stderr_fallback_resp + + +class TestMiddlewareConfig: + """Middleware registration — Session, CORS, HSTS, TraceContext.""" + + def test_session_middleware(self): + from src.app import app + names = [m.cls.__name__ for m in app.user_middleware] + assert "SessionMiddleware" in names + + def test_cors_middleware(self): + from src.app import app + names = [m.cls.__name__ for m in app.user_middleware] + assert "CORSMiddleware" in names + + def test_hsts_middleware(self): + from src.app import app + names = [m.cls.__name__ for m in app.user_middleware] + assert "HSTSMiddleware" in names + + def test_trace_context_middleware(self): + from src.app import app + names = [m.cls.__name__ for m in app.user_middleware] + assert "TraceContextMiddleware" in names + + +class TestCorsConfig: + """CORS and Session configuration edge cases.""" + + def test_cors_empty_allowed_origins(self): + from src.app import app + names = [m.cls.__name__ for m in app.user_middleware] + assert "CORSMiddleware" in names + + def test_session_secret_fallback(self): + from src.app import app + names = [m.cls.__name__ for m in app.user_middleware] + assert "SessionMiddleware" in names + + +class TestRouterRegistration: + """Verify all expected route groups are registered.""" + + def test_expected_routers_registered(self): + from src.app import app + routes = "\n".join(r.path for r in app.routes) + assert "/api/dashboards" in routes or "/api/plugins" in routes + assert "/api/tasks" in routes + assert "/api/settings" in routes + assert "/api/git" in routes or "/api/mappings" in routes + assert "/ws/logs/" in routes + assert "/ws/task-events" in routes +# #endregion Test.AppModule.Middleware diff --git a/backend/tests/test_app_spa.py b/backend/tests/test_app_spa.py new file mode 100644 index 00000000..9d21d153 --- /dev/null +++ b/backend/tests/test_app_spa.py @@ -0,0 +1,113 @@ +# #region Test.AppModule.Spa [C:3] [TYPE Module] [SEMANTICS test,app,spa,serving,static] +# @BRIEF Tests for app.py — SPA serving, read_root, catch-all route, TestClient integration. +# @RELATION BINDS_TO -> [AppModule] +# @TEST_EDGE: no_frontend_build -> read_root returns API status JSON + +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import pytest + + +class TestSpaServing: + """serve_spa and read_root — SPA fallback and API status.""" + + # #region test_read_root_response [C:2] [TYPE Function] + def test_read_root_response(self): + from fastapi.testclient import TestClient + from src.app import app, frontend_path + client = TestClient(app) + response = client.get("/") + assert response.status_code == 200 + if frontend_path.exists(): + assert "text/html" in response.headers.get("content-type", "") + else: + data = response.json() + assert "message" in data + assert "API is running" in data["message"] + # #endregion test_read_root_response + + # #region test_spa_api_path_rejected [C:2] [TYPE Function] + def test_spa_api_path_rejected(self): + from fastapi.testclient import TestClient + from src.app import app + client = TestClient(app) + response = client.get("/api/unknown-route-test-xyz") + assert response.status_code == 404 + assert response.headers.get("content-type", "").startswith("application/json") + # #endregion test_spa_api_path_rejected + + +class TestAppWithTestClient: + """Integration tests using TestClient against the real app.""" + + # #region test_unknown_api_returns_404 [C:2] [TYPE Function] + def test_unknown_api_returns_404(self): + from fastapi.testclient import TestClient + from src.app import app + client = TestClient(app) + response = client.get("/api/nonexistent-endpoint-xyz") + assert response.status_code == 404 + assert response.headers.get("content-type", "").startswith("application/json") + # #endregion test_unknown_api_returns_404 + + +class TestServeSpaEdgeCases: + """serve_spa — static file serving edge cases.""" + + # #region test_serve_spa_known_file [C:2] [TYPE Function] + def test_serve_spa_known_file(self): + from pathlib import Path as P + root = P(__file__).resolve().parent.parent.parent + fp = root / "frontend" / "build" + if not fp.exists(): + pytest.skip("Frontend build not found") + from fastapi.testclient import TestClient + from src.app import app + client = TestClient(app) + app_dir = fp / "_app" + if app_dir.exists(): + assets = list(app_dir.rglob("*")) + if assets: + rel = assets[0].relative_to(fp) + resp = client.get(f"/{rel}") + assert resp.status_code == 200 + # #endregion test_serve_spa_known_file + + # #region test_spa_unknown_file_returns_index [C:2] [TYPE Function] + def test_spa_unknown_file_returns_index(self): + from pathlib import Path as P + root = P(__file__).resolve().parent.parent.parent + fp = root / "frontend" / "build" + if not fp.exists(): + pytest.skip("Frontend build not found") + from fastapi.testclient import TestClient + from src.app import app + client = TestClient(app) + resp = client.get("/some-random-non-api-path") + assert resp.status_code == 200 + assert "text/html" in resp.headers.get("content-type", "") + # #endregion test_spa_unknown_file_returns_index + + # #region test_spa_api_path_404 [C:2] [TYPE Function] + def test_spa_api_path_404(self): + from fastapi.testclient import TestClient + from src.app import app + client = TestClient(app) + resp = client.get("/api/does-not-exist-98765") + assert resp.status_code == 404 + assert "application/json" in resp.headers.get("content-type", "") + # #endregion test_spa_api_path_404 + + # #region test_spa_api_path_no_leading_slash [C:2] [TYPE Function] + def test_spa_api_path_no_leading_slash(self): + from fastapi.testclient import TestClient + from src.app import app + client = TestClient(app) + resp = client.get("/api/another-missing-endpoint") + assert resp.status_code == 404 + assert "application/json" in resp.headers.get("content-type", "") + # #endregion test_spa_api_path_no_leading_slash +# #endregion Test.AppModule.Spa diff --git a/backend/tests/test_app_ws_auth.py b/backend/tests/test_app_ws_auth.py new file mode 100644 index 00000000..5b0fefa5 --- /dev/null +++ b/backend/tests/test_app_ws_auth.py @@ -0,0 +1,120 @@ +# #region Test.AppModule.WsAuth [C:3] [TYPE Module] [SEMANTICS test,app,ws,auth,jwt,apikey] +# @BRIEF Tests for app.py — _authenticate_websocket: JWT/API key auth for WebSocket connections. +# @RELATION BINDS_TO -> [AppModule] +# @TEST_EDGE: ws_missing_token -> returns False +# @TEST_EDGE: ws_jwt_auth -> accepts valid JWT +# @TEST_EDGE: ws_jwt_fallback_apikey -> accepts valid API key after JWT fail +# @TEST_EDGE: ws_both_fail -> returns False when both auth methods fail +# @TEST_INVARIANT: ws_auth_gate -> VERIFIED_BY: test_websocket_auth_missing_token, test_websocket_auth_jwt_valid, test_websocket_auth_api_key + +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + + +class TestAuthenticateWebsocket: + """_authenticate_websocket — JWT/API key auth.""" + + # #region test_missing_token [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_missing_token(self): + from src.app import _authenticate_websocket + ws = MagicMock(); ws.query_params = {} + assert await _authenticate_websocket(ws, "ws/logs") is False + # #endregion test_missing_token + + # #region test_jwt_valid [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_jwt_valid(self): + from src.app import _authenticate_websocket + ws = MagicMock(); ws.query_params = {"token": "valid.jwt"} + with patch("src.core.auth.jwt.decode_token", return_value={"sub": "testuser"}) as md: + assert await _authenticate_websocket(ws, "ws/logs") is True + md.assert_called_once_with("valid.jwt") + # #endregion test_jwt_valid + + # #region test_jwt_no_sub [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_jwt_no_sub(self): + from src.app import _authenticate_websocket + ws = MagicMock(); ws.query_params = {"token": "no-sub"} + with ( + patch("src.core.auth.jwt.decode_token", return_value={"role": "admin"}), + patch("src.core.auth.api_key.hash_api_key", return_value="hk"), + patch("src.core.database.SessionLocal") as msl, + ): + msl.return_value = MagicMock() + msl.return_value.query.return_value.filter.return_value.first.return_value = None + assert await _authenticate_websocket(ws, "ws/logs") is False + # #endregion test_jwt_no_sub + + # #region test_jwt_fallback_apikey [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_jwt_fallback_apikey(self): + from src.app import _authenticate_websocket + ws = MagicMock(); ws.query_params = {"token": "valid-key"} + key = MagicMock(); key.active = True; key.name = "Test" + with ( + patch("src.core.auth.jwt.decode_token", side_effect=Exception("expired")), + patch("src.core.auth.api_key.hash_api_key", return_value="hk"), + patch("src.core.database.SessionLocal") as msl, + ): + msl.return_value = MagicMock() + msl.return_value.query.return_value.filter.return_value.first.return_value = key + assert await _authenticate_websocket(ws, "ws/logs") is True + # #endregion test_jwt_fallback_apikey + + # #region test_apikey_inactive [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_apikey_inactive(self): + from src.app import _authenticate_websocket + ws = MagicMock(); ws.query_params = {"token": "inactive"} + key = MagicMock(); key.active = False + with ( + patch("src.core.auth.jwt.decode_token", side_effect=Exception("fail")), + patch("src.core.auth.api_key.hash_api_key", return_value="hk"), + patch("src.core.database.SessionLocal") as msl, + ): + msl.return_value = MagicMock() + msl.return_value.query.return_value.filter.return_value.first.return_value = key + assert await _authenticate_websocket(ws, "ws/logs") is False + # #endregion test_apikey_inactive + + # #region test_both_fail [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_both_fail(self): + from src.app import _authenticate_websocket + ws = MagicMock(); ws.query_params = {"token": "bad"} + with ( + patch("src.core.auth.jwt.decode_token", side_effect=Exception("invalid")), + patch("src.core.auth.api_key.hash_api_key", return_value="hk"), + patch("src.core.database.SessionLocal") as msl, + ): + msl.return_value = MagicMock() + msl.return_value.query.return_value.filter.return_value.first.return_value = None + assert await _authenticate_websocket(ws, "ws/logs") is False + # #endregion test_both_fail + + +class TestAuthenticateWebsocketApiKeyException: + """_authenticate_websocket — API key DB exception is caught.""" + + # #region test_apikey_db_exception [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_apikey_db_exception(self): + from src.app import _authenticate_websocket + ws = MagicMock(); ws.query_params = {"token": "crash"} + with ( + patch("src.core.auth.jwt.decode_token", side_effect=Exception("JWT fail")), + patch("src.core.auth.api_key.hash_api_key", return_value="hk"), + patch("src.core.database.SessionLocal") as msl, + ): + db = MagicMock(); db.query.side_effect = Exception("DB lost") + msl.return_value = db + assert await _authenticate_websocket(ws, "ws/logs") is False + # #endregion test_apikey_db_exception +# #endregion Test.AppModule.WsAuth diff --git a/backend/tests/test_app_ws_endpoint.py b/backend/tests/test_app_ws_endpoint.py new file mode 100644 index 00000000..4f2731cf --- /dev/null +++ b/backend/tests/test_app_ws_endpoint.py @@ -0,0 +1,317 @@ +# #region Test.AppModule.WsEndpoint [C:3] [TYPE Module] [SEMANTICS test,app,ws,endpoint,logs] +# @BRIEF Tests for app.py — websocket_endpoint: full flow, filters, disconnect, AWAITING_INPUT, terminal log, exceptions. +# @RELATION BINDS_TO -> [AppModule] + +import asyncio +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import pytest +from unittest.mock import AsyncMock, MagicMock, patch +from starlette.websockets import WebSocketDisconnect + + +# ── Helpers ────────────────────────────────────────────────────── + +# #region _make_mock_task [C:1] [TYPE Function] +def _make_mock_task(task_id="task-1", status="RUNNING", plugin_id="test"): + from datetime import datetime, timezone + task = MagicMock() + task.id = task_id; task.plugin_id = plugin_id; task.status = status + task.started_at = datetime(2025, 1, 15, 10, 0, 0, tzinfo=timezone.utc) + task.finished_at = None; task.user_id = "user-1" + task.result = None; task.input_required = False; task.input_request = None + task.logs = [] + return task +# #endregion _make_mock_task + +# #region _make_mock_log_entry [C:1] [TYPE Function] +def _make_mock_log_entry(msg="Test log", source="plugin", level="INFO"): + from datetime import datetime, timezone + ts = datetime(2025, 1, 15, 10, 0, 0, tzinfo=timezone.utc) + entry = MagicMock() + entry.source = source; entry.level = level; entry.message = msg; entry.timestamp = ts + entry.model_dump = MagicMock(return_value={ + "timestamp": ts, "level": level, "message": msg, "source": source, + }) + return entry +# #endregion _make_mock_log_entry + +# #region _make_task_manager_mock [C:1] [TYPE Function] +def _make_task_manager_mock(task=None, logs=None): + tm = MagicMock() + tm.get_task = MagicMock(return_value=task) + tm.get_task_logs = MagicMock(return_value=logs or []) + tm.unsubscribe_logs = MagicMock() + tm.unsubscribe_status = MagicMock() + tm.unsubscribe_task_events = MagicMock() + tm.unsubscribe_maintenance_events = MagicMock() + tm.unsubscribe_dataset_events = MagicMock() + return tm +# #endregion _make_task_manager_mock + + +class TestWebSocketEndpointFull: + """websocket_endpoint — full flow.""" + + # #region test_auth_rejected_closes [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_auth_rejected_closes(self): + from src.app import websocket_endpoint + ws = MagicMock(); ws.query_params = {}; ws.close = AsyncMock() + with patch("src.app._authenticate_websocket", return_value=False): + await websocket_endpoint(ws, "task-1") + ws.close.assert_called_once_with(code=4001, reason="Authentication required") + ws.accept.assert_not_called() + # #endregion test_auth_rejected_closes + + # #region test_accepts_and_sends_initial_status [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_accepts_and_sends_initial_status(self): + from src.app import websocket_endpoint + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(); ws.close = AsyncMock(); ws.accept = AsyncMock() + task = _make_mock_task("task-1", "RUNNING") + log = _make_mock_log_entry(msg="Starting") + lq, sq = asyncio.Queue(), asyncio.Queue() + await sq.put({"type": "task_status", "task_id": "task-1", "task": {"status": "SUCCESS"}}) + async def sl(t): return lq + async def ss(t): return sq + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.app.get_task_manager") as mg, + ): + tm = _make_task_manager_mock(task=task, logs=[log]) + tm.subscribe_logs = sl; tm.subscribe_status = ss; mg.return_value = tm + await websocket_endpoint(ws, "task-1", source=None, level=None) + ws.accept.assert_called_once() + assert ws.send_json.call_count >= 2 + # #endregion test_accepts_and_sends_initial_status + + # #region test_source_filter [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_source_filter(self): + from src.app import websocket_endpoint + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(); ws.accept = AsyncMock() + task = _make_mock_task("task-1", "RUNNING") + plug = _make_mock_log_entry(msg="Plugin", source="plugin") + super = _make_mock_log_entry(msg="Superset", source="superset_api") + lq, sq = asyncio.Queue(), asyncio.Queue() + await sq.put({"type": "task_status", "task_id": "task-1", "task": {"status": "SUCCESS"}}) + async def sl(t): return lq + async def ss(t): return sq + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.app.get_task_manager") as mg, + ): + tm = _make_task_manager_mock(task=task, logs=[plug, super]) + tm.subscribe_logs = sl; tm.subscribe_status = ss; mg.return_value = tm + await websocket_endpoint(ws, "task-1", source="plugin") + msgs = [c[0][0].get("message") for c in ws.send_json.call_args_list if isinstance(c[0][0], dict)] + assert "Superset" not in msgs + # #endregion test_source_filter + + # #region test_level_filter [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_level_filter(self): + from src.app import websocket_endpoint + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(); ws.accept = AsyncMock() + task = _make_mock_task("task-1", "RUNNING") + dbg = _make_mock_log_entry(msg="Debug", level="DEBUG") + inf = _make_mock_log_entry(msg="Info", level="INFO") + lq, sq = asyncio.Queue(), asyncio.Queue() + await sq.put({"type": "task_status", "task_id": "task-1", "task": {"status": "SUCCESS"}}) + async def sl(t): return lq + async def ss(t): return sq + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.app.get_task_manager") as mg, + ): + tm = _make_task_manager_mock(task=task, logs=[dbg, inf]) + tm.subscribe_logs = sl; tm.subscribe_status = ss; mg.return_value = tm + await websocket_endpoint(ws, "task-1", source=None, level="INFO") + msgs = [c[0][0].get("message") for c in ws.send_json.call_args_list if isinstance(c[0][0], dict)] + assert "Debug" not in msgs + assert "Info" in msgs + # #endregion test_level_filter + + # #region test_disconnect_in_main_loop [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_disconnect_in_main_loop(self): + from src.app import websocket_endpoint + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(side_effect=[None, WebSocketDisconnect()]) + ws.close = AsyncMock(); ws.accept = AsyncMock() + task = _make_mock_task("task-1", "RUNNING") + lq, sq = asyncio.Queue(), asyncio.Queue() + await sq.put({"type": "task_status", "task_id": "task-1", "task": {"status": "SUCCESS"}}) + async def sl(t): return lq + async def ss(t): return sq + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.app.get_task_manager") as mg, + ): + tm = _make_task_manager_mock(task=task, logs=[]) + tm.subscribe_logs = sl; tm.subscribe_status = ss; mg.return_value = tm + await websocket_endpoint(ws, "task-1") + ws.accept.assert_called_once() + # #endregion test_disconnect_in_main_loop + + # #region test_awaiting_input_prompt [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_awaiting_input_prompt(self): + from src.app import websocket_endpoint + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(); ws.accept = AsyncMock() + task = _make_mock_task("task-1", "AWAITING_INPUT") + task.input_request = {"type": "database_password", "databases": ["db1"]} + lq, sq = asyncio.Queue(), asyncio.Queue() + await sq.put({"type": "task_status", "task_id": "task-1", "task": {"status": "SUCCESS"}}) + async def sl(t): return lq + async def ss(t): return sq + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.app.get_task_manager") as mg, + ): + tm = _make_task_manager_mock(task=task, logs=[]) + tm.subscribe_logs = sl; tm.subscribe_status = ss; mg.return_value = tm + await websocket_endpoint(ws, "task-1") + msgs = [c[0][0].get("message") for c in ws.send_json.call_args_list if isinstance(c[0][0], dict)] + assert any("Task paused for user input" in (m or "") for m in msgs) + # #endregion test_awaiting_input_prompt + + # #region test_terminal_log_triggers_close [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_terminal_log_triggers_close(self): + from src.app import websocket_endpoint + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(); ws.accept = AsyncMock() + task = _make_mock_task("task-1", "RUNNING") + term = _make_mock_log_entry(msg="Task completed successfully") + lq, sq = asyncio.Queue(), asyncio.Queue() + await sq.put({"type": "task_status", "task_id": "task-1", "task": {"status": "SUCCESS"}}) + async def sl(t): return lq + async def ss(t): return sq + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.app.get_task_manager") as mg, + ): + tm = _make_task_manager_mock(task=task, logs=[term]) + tm.subscribe_logs = sl; tm.subscribe_status = ss; mg.return_value = tm + await websocket_endpoint(ws, "task-1") + ws.accept.assert_called_once() + # #endregion test_terminal_log_triggers_close + + +class TestMatchesFilters: + """matches_filters — extracted filter logic.""" + + # #region test_matches_filters [C:2] [TYPE Function] + def test_matches_filters(self): + def mf(entry, source_filter=None, level_filter=None): + lh = {"DEBUG": 0, "INFO": 1, "WARNING": 2, "ERROR": 3} + ml = lh.get(level_filter, 0) if level_filter else 0 + src = getattr(entry, "source", None) + if source_filter and str(src or "").lower() != source_filter: + return False + if level_filter: + el = lh.get(str(entry.level).upper(), 0) + if el < ml: + return False + return True + class E: + def __init__(self, s, l): self.source = s; self.level = l + assert mf(E("plugin", "INFO")) is True + assert mf(E("plugin", "INFO"), source_filter="other") is False + assert mf(E("other", "DEBUG"), level_filter="INFO") is False + assert mf(E("superset_api", "WARNING"), source_filter="superset_api", level_filter="WARNING") is True + assert mf(E(None, "INFO"), source_filter="plugin") is False + # #endregion test_matches_filters + + +class TestWebSocketMainLoopExceptions: + """Generic Exception re-raise in WS main loops.""" + + # #region test_ws_endpoint_generic_exception [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_ws_endpoint_generic_exception(self): + from src.app import websocket_endpoint + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(); ws.accept = AsyncMock() + task = _make_mock_task("task-1", "RUNNING") + lq, sq = asyncio.Queue(), asyncio.Queue() + await sq.put("not-a-dict") # will cause AttributeError + async def sl(t): return lq + async def ss(t): return sq + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.app.get_task_manager") as mg, + ): + tm = _make_task_manager_mock(task=task, logs=[]) + tm.subscribe_logs = sl; tm.subscribe_status = ss; mg.return_value = tm + with pytest.raises(Exception): + await websocket_endpoint(ws, "task-1") + # #endregion test_ws_endpoint_generic_exception + + # #region test_task_events_generic_exception [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_task_events_generic_exception(self): + from src.app import task_events_websocket + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(side_effect=RuntimeError("crash")) + ws.accept = AsyncMock() + q = asyncio.Queue() + await q.put({"type": "task_status"}) + async def sub(): return q + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.app.get_task_manager") as mg, + ): + tm = MagicMock(); tm.subscribe_task_events = sub; tm.unsubscribe_task_events = MagicMock() + mg.return_value = tm + with pytest.raises(RuntimeError, match="crash"): + await task_events_websocket(ws) + # #endregion test_task_events_generic_exception + + # #region test_maintenance_events_generic_exception [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_maintenance_events_generic_exception(self): + from src.app import maintenance_events_websocket + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(side_effect=ValueError("bad")) + ws.accept = AsyncMock() + q = asyncio.Queue(); await q.put({"type": "maintenance.event_created"}) + async def sub(): return q + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.app.get_task_manager") as mg, + ): + tm = MagicMock(); tm.subscribe_maintenance_events = sub; tm.unsubscribe_maintenance_events = MagicMock() + mg.return_value = tm + with pytest.raises(ValueError, match="bad"): + await maintenance_events_websocket(ws) + # #endregion test_maintenance_events_generic_exception + + # #region test_dataset_ws_generic_exception [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_dataset_ws_generic_exception(self): + from src.app import dataset_websocket_endpoint + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(side_effect=RuntimeError("fail")) + ws.accept = AsyncMock() + q = asyncio.Queue(); await q.put({"type": "dataset.updated"}) + async def sub(e): return q + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.app.get_task_manager") as mg, + ): + tm = MagicMock(); tm.subscribe_dataset_events = sub; tm.unsubscribe_dataset_events = MagicMock() + mg.return_value = tm + await dataset_websocket_endpoint(ws, "env-1") + ws.accept.assert_called_once() + # #endregion test_dataset_ws_generic_exception +# #endregion Test.AppModule.WsEndpoint diff --git a/backend/tests/test_app_ws_events.py b/backend/tests/test_app_ws_events.py new file mode 100644 index 00000000..904f369e --- /dev/null +++ b/backend/tests/test_app_ws_events.py @@ -0,0 +1,183 @@ +# #region Test.AppModule.WsEvents [C:3] [TYPE Module] [SEMANTICS test,app,ws,events,maintenance,translate,dataset] +# @BRIEF Tests for app.py — task_events_websocket, maintenance_events_websocket, dataset_websocket_endpoint, translate_run_websocket. +# @RELATION BINDS_TO -> [AppModule] + +import asyncio +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +import pytest +from unittest.mock import AsyncMock, MagicMock, patch +from starlette.websockets import WebSocketDisconnect + + +class TestTaskEventsWebSocket: + """task_events_websocket — global task events stream.""" + + # #region test_auth_rejected [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_auth_rejected(self): + from src.app import task_events_websocket + ws = MagicMock(); ws.query_params = {}; ws.close = AsyncMock() + with patch("src.app._authenticate_websocket", return_value=False): + await task_events_websocket(ws) + ws.close.assert_called_once_with(code=4001, reason="Authentication required") + ws.accept.assert_not_called() + # #endregion test_auth_rejected + + # #region test_accepts_and_streams [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_accepts_and_streams(self): + from src.app import task_events_websocket + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(side_effect=[None, WebSocketDisconnect()]) + ws.accept = AsyncMock() + q = asyncio.Queue() + await q.put({"type": "task_status", "task_id": "t1", "task": {"status": "RUNNING"}}) + await q.put({"type": "task_status", "task_id": "t1", "task": {"status": "COMPLETED"}}) + async def sub(): return q + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.app.get_task_manager") as mg, + ): + tm = MagicMock(); tm.subscribe_task_events = sub; tm.unsubscribe_task_events = MagicMock() + mg.return_value = tm + await task_events_websocket(ws) + ws.accept.assert_called_once() + ws.send_json.assert_any_call({"type": "task_status", "task_id": "t1", "task": {"status": "RUNNING"}}) + tm.unsubscribe_task_events.assert_called_once() + # #endregion test_accepts_and_streams + + +class TestMaintenanceEventsWebSocket: + """maintenance_events_websocket — maintenance event stream.""" + + # #region test_auth_rejected [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_auth_rejected(self): + from src.app import maintenance_events_websocket + ws = MagicMock(); ws.query_params = {}; ws.close = AsyncMock() + with patch("src.app._authenticate_websocket", return_value=False): + await maintenance_events_websocket(ws) + ws.close.assert_called_once_with(code=4001, reason="Authentication required") + # #endregion test_auth_rejected + + # #region test_accepts [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_accepts(self): + from src.app import maintenance_events_websocket + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(side_effect=[None, WebSocketDisconnect()]) + ws.accept = AsyncMock() + q = asyncio.Queue() + await q.put({"type": "maintenance.event_created", "maintenance_id": "m-1"}) + await q.put({"type": "maintenance.event_ended", "maintenance_id": "m-1"}) + async def sub(): return q + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.app.get_task_manager") as mg, + ): + tm = MagicMock(); tm.subscribe_maintenance_events = sub; tm.unsubscribe_maintenance_events = MagicMock() + mg.return_value = tm + await maintenance_events_websocket(ws) + ws.accept.assert_called_once() + evt = ws.send_json.call_args_list[0][0][0] + assert evt.get("type") == "maintenance.event_created" + tm.unsubscribe_maintenance_events.assert_called_once() + # #endregion test_accepts + + +class TestDatasetWebSocket: + """dataset_websocket_endpoint — dataset.updated event stream.""" + + # #region test_auth_rejected [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_auth_rejected(self): + from src.app import dataset_websocket_endpoint + ws = MagicMock(); ws.query_params = {}; ws.close = AsyncMock() + with patch("src.app._authenticate_websocket", return_value=False): + await dataset_websocket_endpoint(ws, "env-1") + ws.close.assert_called_once_with(code=4001, reason="Authentication required") + # #endregion test_auth_rejected + + # #region test_accepts [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_accepts(self): + from src.app import dataset_websocket_endpoint + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(side_effect=[None, WebSocketDisconnect()]) + ws.accept = AsyncMock() + q = asyncio.Queue() + await q.put({"type": "dataset.updated", "dataset_id": "ds-1"}) + await q.put({"type": "dataset.updated", "dataset_id": "ds-2"}) + async def sub(e): return q + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.app.get_task_manager") as mg, + ): + tm = MagicMock(); tm.subscribe_dataset_events = sub; tm.unsubscribe_dataset_events = MagicMock() + mg.return_value = tm + await dataset_websocket_endpoint(ws, "env-1") + ws.accept.assert_called_once() + tm.unsubscribe_dataset_events.assert_called_once() + # #endregion test_accepts + + +class TestTranslateRunWebSocket: + """translate_run_websocket — translation run progress stream.""" + + # #region test_auth_rejected [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_auth_rejected(self): + from src.app import translate_run_websocket + ws = MagicMock(); ws.query_params = {}; ws.close = AsyncMock() + with patch("src.app._authenticate_websocket", return_value=False): + await translate_run_websocket(ws, "run-1") + ws.close.assert_called_once_with(code=4001, reason="Authentication required") + # #endregion test_auth_rejected + + # #region test_accepts_and_streams [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_accepts_and_streams(self): + from src.app import translate_run_websocket + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(); ws.accept = AsyncMock() + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.core.database.SessionLocal") as msl, + patch("src.plugins.translate.orchestrator_aggregator.TranslationResultAggregator") as ma, + patch("src.plugins.translate.events.TranslationEventLog"), + ): + msl.return_value = MagicMock() + agg = MagicMock() + agg.get_run_status.return_value = { + "status": "COMPLETED", "run_id": "run-1", + "total_records": 100, "successful_records": 100, + "failed_records": 0, "skipped_records": 0, + } + ma.return_value = agg + await translate_run_websocket(ws, "run-1") + ws.accept.assert_called_once() + sent = ws.send_json.call_args[0][0] + assert sent.get("status") == "COMPLETED" + assert sent.get("progressPct") == 100 + # #endregion test_accepts_and_streams + + # #region test_error_tick [C:2] [TYPE Function] + @pytest.mark.asyncio + async def test_error_tick(self): + from src.app import translate_run_websocket + ws = MagicMock(); ws.query_params = {"token": "valid"} + ws.send_json = AsyncMock(); ws.accept = AsyncMock() + with ( + patch("src.app._authenticate_websocket", return_value=True), + patch("src.core.database.SessionLocal", side_effect=Exception("DB failed")), + ): + await translate_run_websocket(ws, "run-1") + ws.accept.assert_called_once() + sent = ws.send_json.call_args[0][0] + assert "error" in sent + # #endregion test_error_tick +# #endregion Test.AppModule.WsEvents