From 97a81608218f734234ab29a41e789fba7dc14323 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Thu, 16 Apr 2026 08:13:43 +0000 Subject: [PATCH] feat: add hexagonal architecture tests with pytestarch (#177) - Create tests/test_architecture.py with 20 automated rules: inter-layer dependency checks (domain, services, api, infra, persistence), external dependency constraints (fastapi, sqlalchemy, httpx, opensearchpy), and port convention enforcement (Protocol only in domain.ports) - Add requirements-test.txt with pytestarch dependency - Fix persistence.database importing infra.settings (read DB_PATH from env directly) --- document-parser/persistence/database.py | 4 +- document-parser/requirements-test.txt | 4 + document-parser/tests/test_architecture.py | 208 +++++++++++++++++++++ 3 files changed, 213 insertions(+), 3 deletions(-) create mode 100644 document-parser/requirements-test.txt create mode 100644 document-parser/tests/test_architecture.py diff --git a/document-parser/persistence/database.py b/document-parser/persistence/database.py index b32e694..c56620e 100644 --- a/document-parser/persistence/database.py +++ b/document-parser/persistence/database.py @@ -9,11 +9,9 @@ from contextlib import asynccontextmanager import aiosqlite -from infra.settings import settings - logger = logging.getLogger(__name__) -DB_PATH = settings.db_path +DB_PATH = os.environ.get("DB_PATH", "./data/docling_studio.db") _SCHEMA = """ CREATE TABLE IF NOT EXISTS documents ( diff --git a/document-parser/requirements-test.txt b/document-parser/requirements-test.txt new file mode 100644 index 0000000..255ac0b --- /dev/null +++ b/document-parser/requirements-test.txt @@ -0,0 +1,4 @@ +-r requirements.txt +pytest>=8.0.0,<9.0.0 +pytest-asyncio>=0.23.0,<1.0.0 +pytestarch>=2.0.0,<3.0.0 diff --git a/document-parser/tests/test_architecture.py b/document-parser/tests/test_architecture.py new file mode 100644 index 0000000..c74f68d --- /dev/null +++ b/document-parser/tests/test_architecture.py @@ -0,0 +1,208 @@ +"""Hexagonal architecture tests — enforce layer dependency rules. + +Uses pytestarch for inter-layer dependency rules and ast-based import +scanning for external (third-party) dependency constraints. + +Rules enforced: +- domain -> no import from api, services, infra, persistence +- services -> no import from api, infra, persistence +- api -> no import from infra, persistence +- infra -> no import from api, services +- persistence -> no import from api, services, infra +- domain -> no import of fastapi, sqlalchemy, httpx, opensearchpy +- services -> no import of fastapi +""" + +from __future__ import annotations + +import ast +from pathlib import Path + +import pytest +from pytestarch import Rule, get_evaluable_architecture + +# --------------------------------------------------------------------------- +# pytestarch evaluable (project root = document-parser/) +# --------------------------------------------------------------------------- + +_PROJECT_ROOT = Path(__file__).resolve().parent.parent + +# pytestarch uses the directory name as module prefix when given absolute paths. +# We use the directory name to build qualified module references. +_PREFIX = _PROJECT_ROOT.name # "document-parser" + +_evaluable = get_evaluable_architecture(str(_PROJECT_ROOT), str(_PROJECT_ROOT)) + + +def _mod(layer: str) -> str: + """Return the fully-qualified pytestarch module name for a layer.""" + return f"{_PREFIX}.{layer}" + + +# --------------------------------------------------------------------------- +# Helper: collect top-level imports from all .py files in a package +# --------------------------------------------------------------------------- + + +def _collect_imports(package: str) -> set[str]: + """Return the set of top-level module names imported by *package*.""" + pkg_path = Path(_PROJECT_ROOT) / package + imports: set[str] = set() + for py_file in pkg_path.rglob("*.py"): + tree = ast.parse(py_file.read_text(), filename=str(py_file)) + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + imports.add(alias.name.split(".")[0]) + elif isinstance(node, ast.ImportFrom) and node.module: + imports.add(node.module.split(".")[0]) + return imports + + +# --------------------------------------------------------------------------- +# Inter-layer dependency rules (pytestarch) +# --------------------------------------------------------------------------- + + +class TestDomainLayerIsolation: + """domain must not depend on any other layer.""" + + @pytest.mark.parametrize("forbidden", ["api", "services", "infra", "persistence"]) + def test_domain_does_not_import(self, forbidden: str): + rule = ( + Rule() + .modules_that() + .are_sub_modules_of(_mod("domain")) + .should_not() + .import_modules_that() + .are_sub_modules_of(_mod(forbidden)) + ) + rule.assert_applies(_evaluable) + + +class TestServicesLayerIsolation: + """services may import domain only.""" + + @pytest.mark.parametrize("forbidden", ["api", "infra", "persistence"]) + def test_services_does_not_import(self, forbidden: str): + rule = ( + Rule() + .modules_that() + .are_sub_modules_of(_mod("services")) + .should_not() + .import_modules_that() + .are_sub_modules_of(_mod(forbidden)) + ) + rule.assert_applies(_evaluable) + + +class TestApiLayerIsolation: + """api may import services and domain, but not infra or persistence.""" + + @pytest.mark.parametrize("forbidden", ["infra", "persistence"]) + def test_api_does_not_import(self, forbidden: str): + rule = ( + Rule() + .modules_that() + .are_sub_modules_of(_mod("api")) + .should_not() + .import_modules_that() + .are_sub_modules_of(_mod(forbidden)) + ) + rule.assert_applies(_evaluable) + + +class TestInfraLayerIsolation: + """infra may import domain (ports), but not api or services.""" + + @pytest.mark.parametrize("forbidden", ["api", "services"]) + def test_infra_does_not_import(self, forbidden: str): + rule = ( + Rule() + .modules_that() + .are_sub_modules_of(_mod("infra")) + .should_not() + .import_modules_that() + .are_sub_modules_of(_mod(forbidden)) + ) + rule.assert_applies(_evaluable) + + +class TestPersistenceLayerIsolation: + """persistence may import domain, but not api, services, or infra.""" + + @pytest.mark.parametrize("forbidden", ["api", "services", "infra"]) + def test_persistence_does_not_import(self, forbidden: str): + rule = ( + Rule() + .modules_that() + .are_sub_modules_of(_mod("persistence")) + .should_not() + .import_modules_that() + .are_sub_modules_of(_mod(forbidden)) + ) + rule.assert_applies(_evaluable) + + +# --------------------------------------------------------------------------- +# External dependency rules (ast-based) +# --------------------------------------------------------------------------- + +_DOMAIN_FORBIDDEN_EXTERNALS = {"fastapi", "sqlalchemy", "httpx", "opensearchpy"} +_SERVICES_FORBIDDEN_EXTERNALS = {"fastapi"} + + +class TestDomainExternalDependencies: + """domain must not import infrastructure-specific third-party libraries.""" + + @pytest.mark.parametrize("lib", sorted(_DOMAIN_FORBIDDEN_EXTERNALS)) + def test_domain_does_not_import_external(self, lib: str): + imports = _collect_imports("domain") + assert lib not in imports, f"domain imports forbidden external library '{lib}'" + + +class TestServicesExternalDependencies: + """services must not import web-framework libraries.""" + + @pytest.mark.parametrize("lib", sorted(_SERVICES_FORBIDDEN_EXTERNALS)) + def test_services_does_not_import_external(self, lib: str): + imports = _collect_imports("services") + assert lib not in imports, f"services imports forbidden external library '{lib}'" + + +# --------------------------------------------------------------------------- +# Convention: ports live exclusively in domain.ports +# --------------------------------------------------------------------------- + + +class TestPortConvention: + """Protocol definitions (ports) must live in domain.ports only.""" + + def test_no_protocol_outside_domain_ports(self): + """No Protocol subclass should be defined outside domain/ports.py.""" + ports_file = Path(_PROJECT_ROOT) / "domain" / "ports.py" + for py_file in Path(_PROJECT_ROOT).rglob("*.py"): + if py_file == ports_file: + continue + # Skip test files and __pycache__ + if "tests" in py_file.parts or "__pycache__" in py_file.parts: + continue + tree = ast.parse(py_file.read_text(), filename=str(py_file)) + for node in ast.walk(tree): + if isinstance(node, ast.ClassDef): + for base in node.bases: + base_name = _get_name(base) + if base_name == "Protocol": + pytest.fail( + f"Protocol '{node.name}' defined in {py_file.relative_to(_PROJECT_ROOT)}" + f" — ports must live in domain/ports.py" + ) + + +def _get_name(node: ast.expr) -> str: + """Extract a simple name from an AST expression node.""" + if isinstance(node, ast.Name): + return node.id + if isinstance(node, ast.Attribute): + return node.attr + return ""