diff --git a/tests/test_api_smoke.py b/tests/test_api_smoke.py index 766f96a..317e346 100644 --- a/tests/test_api_smoke.py +++ b/tests/test_api_smoke.py @@ -2463,13 +2463,42 @@ class ApiSmokeTests(unittest.TestCase): self.assertEqual(unchanged["file"]["id"], nextcloud_file["id"]) self.assertEqual(unchanged["current_version_id"], synced["current_version_id"]) - with patch("govoplan_files.backend.storage.connector_browse._smbclient_module") as smb_sdk: - smb_browse = self.client.get("/api/v1/files/connectors/profiles/tenant-smb/browse?path=shared", headers=headers) - self.assertEqual(smb_browse.status_code, 400, smb_browse.text) - self.assertIn("redirects/referrals", smb_browse.json()["detail"]) - smb_sdk.assert_not_called() + from govoplan_files.backend.storage.connector_browse import ( + ConnectorBrowseUnsupported, + ) - with patch("govoplan_files.backend.storage.connector_imports._smbclient_module") as smb_import_sdk: + optional_dependency_error = ConnectorBrowseUnsupported( + "SMB connector browsing requires the optional smbprotocol dependency" + ) + with patch( + "govoplan_files.backend.storage.connector_browse._smbclient_module", + side_effect=optional_dependency_error, + ) as missing_smb_loader: + missing_smb_browse = self.client.get( + "/api/v1/files/connectors/profiles/tenant-smb/browse?path=shared", + headers=headers, + ) + self.assertEqual(missing_smb_browse.status_code, 501, missing_smb_browse.text) + self.assertIn("optional smbprotocol", missing_smb_browse.json()["detail"]) + missing_smb_loader.assert_called_once_with() + + with patch( + "govoplan_files.backend.storage.connector_browse._smbclient_module" + ) as available_smb_loader: + available_smb_loader.return_value.scandir.return_value.__enter__.return_value = iter(()) + available_smb_browse = self.client.get( + "/api/v1/files/connectors/profiles/tenant-smb/browse?path=shared", + headers=headers, + ) + self.assertEqual(available_smb_browse.status_code, 200, available_smb_browse.text) + self.assertEqual([], available_smb_browse.json()["items"]) + self.assertNotIn("super-secret-smb", available_smb_browse.text) + available_smb_loader.assert_called_once_with() + + with patch( + "govoplan_files.backend.storage.connector_imports._smbclient_module", + side_effect=optional_dependency_error, + ) as missing_smb_import_loader: smb_import = self.client.post( "/api/v1/files/connectors/profiles/tenant-smb/import", headers=headers, @@ -2481,9 +2510,9 @@ class ApiSmokeTests(unittest.TestCase): "target_folder": "imports", }, ) - self.assertEqual(smb_import.status_code, 400, smb_import.text) - self.assertIn("redirects/referrals", smb_import.json()["detail"]) - smb_import_sdk.assert_not_called() + self.assertEqual(smb_import.status_code, 501, smb_import.text) + self.assertIn("optional smbprotocol", smb_import.json()["detail"]) + missing_smb_import_loader.assert_called_once_with() filtered = self.client.get("/api/v1/files/connectors/profiles?provider=smb", headers=headers) self.assertEqual(filtered.status_code, 200, filtered.text) @@ -2915,7 +2944,18 @@ class ApiSmokeTests(unittest.TestCase): updated_campaigns = {item["id"]: item for item in after_update.json()["campaigns"]} self.assertEqual(updated_campaigns[campaign_id]["name"], "Delta campaign updated") - deleted = self.client.delete(f"/api/v1/campaigns/{campaign_id}", headers=headers) + lifecycle = self.client.get( + f"/api/v1/campaigns/{campaign_id}/lifecycle-policy", + headers=headers, + ) + self.assertEqual(lifecycle.status_code, 200, lifecycle.text) + self.assertTrue(lifecycle.json()["actions"]["delete_campaign"]["allowed"]) + deleted = self.client.request( + "DELETE", + f"/api/v1/campaigns/{campaign_id}", + headers=headers, + json={"expected_state_token": lifecycle.json()["state_token"]}, + ) self.assertEqual(deleted.status_code, 204, deleted.text) after_delete = self.client.get( diff --git a/tests/test_principal_cache.py b/tests/test_principal_cache.py index ff0dd89..0a0bff2 100644 --- a/tests/test_principal_cache.py +++ b/tests/test_principal_cache.py @@ -14,11 +14,22 @@ from govoplan_core.core.principal_cache import ( auth_principal_revision, invalidate_auth_principals, ) +from govoplan_core.core.runtime import ( + clear_runtime, + configure_runtime, + get_runtime_context, +) from govoplan_core.db.base import Base class AuthPrincipalRevisionTests(unittest.TestCase): def setUp(self) -> None: + # These tests create only the Core change-sequence tables. Keep them + # independent of a registry configured by an earlier discovery test; + # that registry may expose an optional durable Audit outbox whose + # tables intentionally are not part of this fixture. + self._runtime_context = get_runtime_context() + clear_runtime() self.engine = create_engine("sqlite:///:memory:") Base.metadata.create_all( bind=self.engine, @@ -31,6 +42,10 @@ class AuthPrincipalRevisionTests(unittest.TestCase): def tearDown(self) -> None: self.engine.dispose() + if self._runtime_context is None: + clear_runtime() + else: + configure_runtime(self._runtime_context) def test_revisions_are_tenant_scoped_and_include_system_changes(self) -> None: with self.SessionLocal() as session: diff --git a/tests/test_wheel_runtime.py b/tests/test_wheel_runtime.py index ea9f055..777695c 100644 --- a/tests/test_wheel_runtime.py +++ b/tests/test_wheel_runtime.py @@ -40,8 +40,17 @@ class WheelRuntimeTests(unittest.TestCase): "pip", "install", "--no-deps", + "--force-reinstall", str(core_wheels[0]), cwd=temporary_root, + # Full editable discovery exports every sibling src directory. + # Do not let pip mistake that importable Core metadata for an + # installation inside this fresh wheel-only environment. + env={ + key: value + for key, value in os.environ.items() + if key != "PYTHONPATH" + }, ) database_path = temporary_root / "wheel-runtime.db"