Merge pull request #159 from awslabs/refactor/tdd-guard-rails

test: add architecture boundary enforcement tests (TDD guard rails)

Author: fgogolli

tests/unit/architecture/conftest.py

@@ -0,0 +1,40 @@
+"""Shared helpers for architecture boundary tests."""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+# Root of the orb source tree — resolved relative to this file's location
+_TESTS_ARCH_DIR = Path(__file__).parent
+_REPO_ROOT = _TESTS_ARCH_DIR.parent.parent.parent
+SRC_ORB = _REPO_ROOT / "src" / "orb"
+
+
+def collect_python_files(directory: Path) -> list[Path]:
+    """Return all .py files under *directory*, sorted for stable parametrize IDs."""
+    return sorted(directory.rglob("*.py"))
+
+
+def extract_imports(filepath: Path) -> list[str]:
+    """AST-parse *filepath* and return every imported module name."""
+    try:
+        tree = ast.parse(filepath.read_text(encoding="utf-8"))
+    except (SyntaxError, UnicodeDecodeError):
+        return []
+    imports: list[str] = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                imports.append(alias.name)
+        elif isinstance(node, ast.ImportFrom):
+            if node.module:
+                imports.append(node.module)
+    return imports
+
+
+# Paths that are explicitly allowed to cross certain boundaries because they
+# perform DI wiring (registering concrete implementations against ports).
+EXCEPTION_PATHS: frozenset[str] = frozenset(
+    str(p) for p in collect_python_files(SRC_ORB / "infrastructure" / "di")
+)

tests/unit/architecture/test_application_import_boundaries.py

@@ -0,0 +1,70 @@
+"""Task 1740: Application layer import boundary tests.
+
+Asserts that no file under src/orb/application/ imports from infrastructure,
+providers, interface, api, or cli layers.
+
+Known violations are tracked in violation_counts.json (ratchet).
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.unit.architecture.conftest import (
+    SRC_ORB,
+    collect_python_files,
+    extract_imports,
+)
+
+_APP_DIR = SRC_ORB / "application"
+
+_FORBIDDEN_PREFIXES = (
+    "orb.infrastructure",
+    "orb.providers",
+    "orb.interface",
+    "orb.api",
+    "orb.cli",
+)
+
+# Known violations that exist in the current codebase — whitelisted so tests
+# pass today while still catching any NEW violations introduced.
+_KNOWN_VIOLATIONS: frozenset[tuple[str, str]] = frozenset(
+    {
+        ("application/queries/template_query_handlers.py", "orb.infrastructure.template.dtos"),
+        (
+            "application/commands/cleanup_handlers.py",
+            "orb.infrastructure.events.infrastructure_events",
+        ),
+        (
+            "application/services/provisioning_orchestration_service.py",
+            "orb.infrastructure.resilience.exceptions",
+        ),
+        (
+            "application/services/provisioning_orchestration_service.py",
+            "orb.infrastructure.resilience.strategy.circuit_breaker",
+        ),
+        ("application/services/provider_registry_service.py", "orb.providers.registry"),
+    }
+)
+
+_APP_FILES = collect_python_files(_APP_DIR)
+
+
+@pytest.mark.parametrize("filepath", _APP_FILES, ids=lambda p: str(p.relative_to(SRC_ORB)))
+@pytest.mark.unit
+@pytest.mark.architecture
+def test_application_file_has_no_new_forbidden_imports(filepath: Path) -> None:
+    """Application file must not introduce NEW imports from outer layers."""
+    rel = str(filepath.relative_to(SRC_ORB))
+    imports = extract_imports(filepath)
+    new_violations = [
+        imp
+        for imp in imports
+        if any(imp == prefix or imp.startswith(prefix + ".") for prefix in _FORBIDDEN_PREFIXES)
+        and (rel, imp) not in _KNOWN_VIOLATIONS
+    ]
+    assert new_violations == [], (
+        f"{rel} has NEW forbidden imports (not in known-violations whitelist): {new_violations}"
+    )

tests/unit/architecture/test_circular_dependencies.py

@@ -0,0 +1,123 @@
+"""Task 1746: Cross-layer circular dependency test.
+
+Builds a module-level dependency graph from all src/orb/ files and asserts
+there are no circular dependencies between top-level packages.
+
+Current known cycles are whitelisted so the test passes today while preventing
+any NEW cycles from being introduced.
+"""
+
+from __future__ import annotations
+
+import collections
+
+from tests.unit.architecture.conftest import (
+    SRC_ORB,
+    collect_python_files,
+    extract_imports,
+)
+
+_TOP_PACKAGES = frozenset(
+    [
+        "domain",
+        "application",
+        "infrastructure",
+        "interface",
+        "api",
+        "cli",
+        "providers",
+        "config",
+        "mcp",
+        "monitoring",
+        "sdk",
+    ]
+)
+
+# Known cycles that exist in the current codebase.  Each cycle is stored as a
+# frozenset of the participating package names (direction-independent, duplicate-free).
+# A detected cycle is "known" if its node-set is a subset of any known cycle node-set.
+_KNOWN_CYCLE_NODESETS: list[frozenset[str]] = [
+    frozenset({"api", "application", "config", "infrastructure"}),
+    frozenset({"application", "config", "infrastructure"}),
+    frozenset({"application", "config", "infrastructure", "cli"}),
+    frozenset({"infrastructure", "cli"}),
+    frozenset({"api", "application", "config", "infrastructure", "cli", "interface"}),
+    frozenset({"application", "config", "infrastructure", "cli", "interface"}),
+    frozenset({"cli", "interface"}),
+    frozenset({"config", "infrastructure", "cli", "interface"}),
+    frozenset({"infrastructure", "cli", "interface"}),
+    frozenset({"application", "config", "infrastructure", "cli", "interface", "mcp", "sdk"}),
+    frozenset({"infrastructure", "cli", "interface", "mcp", "sdk"}),
+    frozenset({"config", "infrastructure", "cli", "interface", "monitoring"}),
+    frozenset({"infrastructure", "cli", "interface", "monitoring"}),
+    frozenset({"application", "config", "infrastructure", "cli", "interface", "providers"}),
+    frozenset({"cli", "interface", "providers"}),
+    frozenset({"config", "infrastructure", "cli", "interface", "providers"}),
+    frozenset({"infrastructure", "cli", "interface", "providers"}),
+    frozenset({"config", "infrastructure"}),
+]
+
+
+def _build_graph() -> dict[str, set[str]]:
+    graph: dict[str, set[str]] = collections.defaultdict(set)
+    for f in collect_python_files(SRC_ORB):
+        parts = f.relative_to(SRC_ORB).parts
+        if not parts:
+            continue
+        src_pkg = parts[0]
+        if src_pkg not in _TOP_PACKAGES:
+            continue
+        for imp in extract_imports(f):
+            if imp.startswith("orb."):
+                rest = imp[4:].split(".")[0]
+                if rest in _TOP_PACKAGES and rest != src_pkg:
+                    graph[src_pkg].add(rest)
+    return dict(graph)
+
+
+def _find_cycles(graph: dict[str, set[str]]) -> list[list[str]]:
+    cycles: list[list[str]] = []
+    visited: set[str] = set()
+    rec_stack: set[str] = set()
+    path: list[str] = []
+    seen_keys: set[tuple[str, ...]] = set()
+
+    def dfs(node: str) -> None:
+        visited.add(node)
+        rec_stack.add(node)
+        path.append(node)
+        for neighbor in sorted(graph.get(node, [])):
+            if neighbor not in visited:
+                dfs(neighbor)
+            elif neighbor in rec_stack:
+                start = path.index(neighbor)
+                cycle = path[start:] + [neighbor]
+                key = tuple(sorted(set(cycle)))
+                if key not in seen_keys:
+                    seen_keys.add(key)
+                    cycles.append(cycle)
+        path.pop()
+        rec_stack.discard(node)
+
+    for node in sorted(graph.keys()):
+        if node not in visited:
+            dfs(node)
+    return cycles
+
+
+def test_no_new_circular_dependencies() -> None:
+    """No new circular dependencies may be introduced between top-level packages."""
+    graph = _build_graph()
+    cycles = _find_cycles(graph)
+
+    new_cycles = []
+    for cycle in cycles:
+        cycle_nodeset = frozenset(cycle)  # includes the repeated last node, frozenset dedupes
+        is_known = any(cycle_nodeset <= known for known in _KNOWN_CYCLE_NODESETS)
+        if not is_known:
+            new_cycles.append(" -> ".join(cycle))
+
+    assert new_cycles == [], (
+        f"NEW circular dependencies detected between top-level packages:\n"
+        + "\n".join(f"  {c}" for c in new_cycles)
+    )

tests/unit/architecture/test_cli_infrastructure_boundary.py

@@ -0,0 +1,57 @@
+"""Task 1747: CLI-to-infrastructure direct import test.
+
+Asserts that no file under src/orb/cli/ imports directly from orb.infrastructure
+except through the DI container (which is an allowed entry point).
+
+Known violations are whitelisted so tests pass today while catching NEW ones.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.unit.architecture.conftest import (
+    SRC_ORB,
+    collect_python_files,
+    extract_imports,
+)
+
+_CLI_DIR = SRC_ORB / "cli"
+_CLI_FILES = collect_python_files(_CLI_DIR)
+
+# Known violations — CLI files that currently import infrastructure directly.
+# Remove entries as they are cleaned up.
+_KNOWN_VIOLATIONS: frozenset[tuple[str, str]] = frozenset(
+    {
+        ("cli/console.py", "orb.infrastructure.constants"),
+        ("cli/registry.py", "orb.infrastructure.di.buses"),
+        ("cli/registry.py", "orb.infrastructure.di.container"),
+        ("cli/router.py", "orb.infrastructure.di.container"),
+        ("cli/response_formatter.py", "orb.infrastructure.logging.logger"),
+        ("cli/main.py", "orb.infrastructure.logging.logger"),
+        ("cli/main.py", "orb.infrastructure.mocking.dry_run_context"),
+        ("cli/main.py", "orb.infrastructure.di.container"),
+        ("cli/factories/provider_command_factory.py", "orb.infrastructure.utilities.json_utils"),
+    }
+)
+
+
+@pytest.mark.parametrize("filepath", _CLI_FILES, ids=lambda p: str(p.relative_to(SRC_ORB)))
+@pytest.mark.unit
+@pytest.mark.architecture
+def test_cli_has_no_new_infrastructure_import(filepath: Path) -> None:
+    """CLI file must not introduce NEW direct imports from orb.infrastructure."""
+    rel = str(filepath.relative_to(SRC_ORB))
+    imports = extract_imports(filepath)
+    new_violations = [
+        imp
+        for imp in imports
+        if (imp == "orb.infrastructure" or imp.startswith("orb.infrastructure."))
+        and (rel, imp) not in _KNOWN_VIOLATIONS
+    ]
+    assert new_violations == [], (
+        f"{rel} has NEW direct infrastructure imports — route through application layer "
+        f"or DI ports instead: {new_violations}"
+    )

tests/unit/architecture/test_domain_import_boundaries.py

@@ -0,0 +1,45 @@
+"""Task 1739: Domain layer import boundary tests.
+
+Asserts that no file under src/orb/domain/ imports from infrastructure,
+providers, interface, api, or cli layers.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.unit.architecture.conftest import (
+    SRC_ORB,
+    collect_python_files,
+    extract_imports,
+)
+
+_DOMAIN_DIR = SRC_ORB / "domain"
+
+_FORBIDDEN_PREFIXES = (
+    "orb.infrastructure",
+    "orb.providers",
+    "orb.interface",
+    "orb.api",
+    "orb.cli",
+)
+
+_DOMAIN_FILES = collect_python_files(_DOMAIN_DIR)
+
+
+@pytest.mark.parametrize("filepath", _DOMAIN_FILES, ids=lambda p: str(p.relative_to(SRC_ORB)))
+@pytest.mark.unit
+@pytest.mark.architecture
+def test_domain_file_has_no_forbidden_imports(filepath: Path) -> None:
+    """Domain file must not import from outer layers."""
+    imports = extract_imports(filepath)
+    violations = [
+        imp
+        for imp in imports
+        if any(imp == prefix or imp.startswith(prefix + ".") for prefix in _FORBIDDEN_PREFIXES)
+    ]
+    assert violations == [], (
+        f"{filepath.relative_to(SRC_ORB)} imports from forbidden layers: {violations}"
+    )