"""Tests for cloud.arch_conformance: architecture consistency checks. Every check has a positive and a negative case; the repo-level test pins that the current tree passes the layer bans, so a future violation fails CI (the existing host-verify workflow runs ``discover -s cloud/tests``; a failure here is a CI failure). Only the standard library; no real model, no network, no secrets. """ import os import sys import tempfile import unittest from pathlib import Path sys.path.insert(0, os.path.dirname(os.path.dirname( os.path.dirname(os.path.abspath(__file__))))) from cloud.arch_conformance import ( # noqa: E402 check_contract_version, check_frozen_stack, check_layer_imports, check_skill_project, check_task_writes, is_within, scan_cloud_package, scan_imports) class TestContractVersion(unittest.TestCase): def test_equal_versions_pass(self): self.assertTrue(check_contract_version("v1", "v1")["ok"]) def test_approved_step_bump_passes(self): change = {"approved": True, "backward_compatible": True, "affected_clients": ["web"]} self.assertTrue( check_contract_version("v1", "v2", change=change)["ok"]) def test_breaking_bump_needs_regression_task(self): change = {"approved": True, "backward_compatible": False, "affected_clients": ["web"]} verdict = check_contract_version("v1", "v2", change=change) self.assertFalse(verdict["ok"]) self.assertTrue(any("regression" in r for r in verdict["reasons"])) def test_breaking_bump_with_regression_task_passes(self): change = {"approved": True, "backward_compatible": False, "affected_clients": ["web"], "regression_task": "t-1"} self.assertTrue( check_contract_version("v1", "v2", change=change)["ok"]) def test_unapproved_bump_fails(self): change = {"approved": False, "backward_compatible": True, "affected_clients": ["web"]} self.assertFalse( check_contract_version("v1", "v2", change=change)["ok"]) def test_bump_without_change_record_fails(self): self.assertFalse(check_contract_version("v1", "v2")["ok"]) def test_skip_bump_fails_even_when_approved(self): change = {"approved": True, "backward_compatible": True, "affected_clients": ["web"]} verdict = check_contract_version("v1", "v3", change=change) self.assertFalse(verdict["ok"]) def test_unparseable_version_fails(self): self.assertFalse(check_contract_version("v1", "draft")["ok"]) class TestFrozenStack(unittest.TestCase): def test_unchanged_stack_passes(self): frozen = {"web": "React", "ios": "SwiftUI"} self.assertTrue(check_frozen_stack(frozen, dict(frozen))["ok"]) def test_new_client_is_addition_not_swap(self): frozen = {"web": "React"} current = {"web": "React", "ios": "SwiftUI"} self.assertTrue(check_frozen_stack(frozen, current)["ok"]) def test_swap_without_approval_fails(self): verdict = check_frozen_stack({"web": "React"}, {"web": "Vue"}) self.assertFalse(verdict["ok"]) self.assertEqual(verdict["swapped"], ["web"]) def test_removed_client_without_approval_fails(self): self.assertFalse( check_frozen_stack({"web": "React", "ios": "SwiftUI"}, {"web": "React"})["ok"]) def test_swap_with_approved_change_passes(self): change = {"approved": True, "backward_compatible": True, "affected_clients": ["web"]} self.assertTrue( check_frozen_stack({"web": "React"}, {"web": "Vue"}, change=change)["ok"]) class TestDirectoryGuard(unittest.TestCase): def test_inside_passes_and_outside_fails(self): with tempfile.TemporaryDirectory() as tmp: own = os.path.join(tmp, "work") os.makedirs(own) self.assertTrue(is_within(own, os.path.join(own, "a.txt"))) self.assertTrue(is_within(own, own)) self.assertFalse(is_within(own, os.path.join(tmp, "other"))) def test_dotdot_escape_fails(self): with tempfile.TemporaryDirectory() as tmp: own = os.path.join(tmp, "work") os.makedirs(own) escape = os.path.join(own, "..", "secret.txt") self.assertFalse(is_within(own, escape)) self.assertEqual(check_task_writes(own, [escape]), [escape]) def test_symlink_escape_fails(self): with tempfile.TemporaryDirectory() as tmp: own = os.path.join(tmp, "work") outside = os.path.join(tmp, "outside") os.makedirs(own) os.makedirs(outside) link = os.path.join(own, "link") try: os.symlink(outside, link) except OSError: self.skipTest("symlinks unavailable") self.assertFalse(is_within(own, link)) def test_mixed_writes_report_only_violations(self): with tempfile.TemporaryDirectory() as tmp: own = os.path.join(tmp, "work") os.makedirs(own) good = os.path.join(own, "ok.txt") bad = os.path.join(tmp, "bad.txt") self.assertEqual(check_task_writes(own, [good, bad]), [bad]) class TestLayerImports(unittest.TestCase): def test_foundation_importing_server_fails(self): verdict = check_layer_imports( {"db": "from cloud.server import foo\n"}) self.assertEqual(len(verdict), 1) self.assertEqual(verdict[0]["rule"], "foundation_must_not_import_orchestration") def test_foundation_importing_store_passes(self): # db importing nothing, models importing nothing: the baseline # shape. A foundation module importing another foundation # module is not an upward import. self.assertEqual(check_layer_imports( {"models": "import os\n", "queue": "from cloud.models import x\n"}), []) def test_sandbox_importing_db_fails(self): verdict = check_layer_imports( {"sandbox": "from cloud.db import connect\n"}) self.assertEqual(len(verdict), 1) def test_sandbox_stdlib_only_passes(self): self.assertEqual(check_layer_imports( {"sandbox": "import os\nimport subprocess\n"}), []) def test_service_opening_sqlite_directly_fails(self): verdict = check_layer_imports( {"server": "import sqlite3\nconn = sqlite3.connect('x.db')\n"}) self.assertTrue(any(v["rule"] == "service_must_not_open_driver_" "connection_directly" for v in verdict)) def test_service_sqlite_row_adapter_passes(self): # Deliberate narrow ban: Row/exception adapters on a # cloud.db connection are adaptation, not a bypass. self.assertEqual(check_layer_imports( {"server": "import sqlite3\nconn.row_factory = sqlite3.Row\n" "try:\n pass\nexcept sqlite3.IntegrityError:\n" " pass\n"}), []) def test_service_importing_psycopg_fails(self): verdict = check_layer_imports( {"worker": "import psycopg\n"}) self.assertTrue(any(v["rule"] == "service_must_not_open_driver_" "connection_directly" for v in verdict)) def test_driver_adjacent_modules_exempt(self): self.assertEqual(check_layer_imports( {"backup": "import sqlite3\nsqlite3.connect('x.db')\n", "migrations": "import sqlite3\nsqlite3.connect('y.db')\n"}), []) def test_scan_imports_sees_from_cloud_form(self): scanned = scan_imports("from cloud.worker import Worker\n") self.assertIn("worker", scanned["cloud"]) class TestRepoLevelScan(unittest.TestCase): def test_current_tree_has_zero_layer_violations(self): repo_root = Path(__file__).resolve().parents[2] violations = scan_cloud_package(repo_root) self.assertEqual(violations, []) class TestSkillProjectFixture(unittest.TestCase): def test_frozen_project_passes_all_checks(self): with tempfile.TemporaryDirectory() as tmp: own = os.path.join(tmp, "apps", "web") os.makedirs(own) verdict = check_skill_project( tmp, "v1", "v1", {"web": "React"}, {"web": "React"}, own, [os.path.join(own, "App.tsx")]) self.assertTrue(verdict["ok"]) def test_drifting_project_fails_each_changed_check(self): with tempfile.TemporaryDirectory() as tmp: own = os.path.join(tmp, "apps", "web") os.makedirs(own) verdict = check_skill_project( tmp, "v1", "v2", {"web": "React"}, {"web": "Vue"}, own, [os.path.join(tmp, "apps", "ios", "leak.txt")]) self.assertFalse(verdict["ok"]) self.assertFalse(verdict["contract"]["ok"]) self.assertFalse(verdict["stack"]["ok"]) self.assertFalse(verdict["directory"]["ok"]) if __name__ == "__main__": unittest.main()