"""Automated Architecture Boundary Enforcement & DDD Conformance Suite. Enforces Hexagonal Architecture (Ports & Adapters) and Domain-Driven Design boundaries. """ from __future__ import annotations import ast from pathlib import Path import pytest def get_all_python_files(directory: Path) -> list[Path]: return [p for p in directory.rglob("*.py") if "__pycache__" not in p.parts] def test_zero_direct_llm_api_calls_in_modules(): """Verify no module calls external LLM REST endpoints directly, bypassing AI Gateway.""" modules_dir = Path("app/modules") py_files = get_all_python_files(modules_dir) forbidden_patterns = [ "https://openrouter.ai/api/v1/chat/completions", "https://generativelanguage.googleapis.com", ] violations = [] for file_path in py_files: # Exclude legacy llm service implementation in agents module if "app/modules/agents/application/llm" in file_path.as_posix(): continue content = file_path.read_text(encoding="utf-8") for pattern in forbidden_patterns: if pattern in content: violations.append(f"{file_path}: Found direct LLM URL '{pattern}'") assert not violations, f"Architectural violations detected:\n" + "\n".join(violations) def test_zero_raw_ddl_in_presentation_layer(): """Verify presentation layer contains zero raw DDL (CREATE/ALTER/DROP TABLE) queries.""" presentation_files = [ p for p in Path("app/modules").rglob("*.py") if "presentation" in p.parts and "__pycache__" not in p.parts ] ddl_keywords = ["CREATE TABLE", "ALTER TABLE", "DROP TABLE"] violations = [] for file_path in presentation_files: content = file_path.read_text(encoding="utf-8").upper() for kw in ddl_keywords: if kw in content: violations.append(f"{file_path}: Found raw DDL keyword '{kw}' in presentation layer") assert not violations, f"Raw DDL found in presentation layer:\n" + "\n".join(violations) def test_zero_raw_sql_in_presentation_layer(): """Verify presentation layer contains zero raw SQL executions (cursor.execute / cur.execute).""" presentation_files = [ p for p in Path("app/modules").rglob("*.py") if "presentation" in p.parts and "__pycache__" not in p.parts ] violations = [] sql_exec_patterns = ["cursor.execute(", "cur.execute("] for file_path in presentation_files: content = file_path.read_text(encoding="utf-8") for pattern in sql_exec_patterns: if pattern in content: violations.append(f"{file_path}: Found raw SQL execution '{pattern}' in presentation layer") assert not violations, f"Raw SQL executions found in presentation layer (Rule 6 violation):\n" + "\n".join(violations) def test_application_and_domain_network_isolation(): """Verify application & domain layers in operations and zalo_crm do not import direct HTTP clients.""" target_dirs = [ Path("app/modules/operations/application"), Path("app/modules/operations/domain"), Path("app/modules/zalo_crm/application"), Path("app/modules/zalo_crm/domain"), Path("app/modules/invoices/application"), Path("app/modules/invoices/domain"), Path("app/modules/core/domain"), ] forbidden_modules = {"httpx", "urllib.request", "requests"} violations = [] for base_dir in target_dirs: if not base_dir.exists(): continue for file_path in get_all_python_files(base_dir): tree = ast.parse(file_path.read_text(encoding="utf-8"), filename=str(file_path)) for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: if alias.name in forbidden_modules: violations.append(f"{file_path}: Illegal direct import '{alias.name}'") elif isinstance(node, ast.ImportFrom): if node.module and any(node.module == mod or node.module.startswith(f"{mod}.") for mod in forbidden_modules): violations.append(f"{file_path}: Illegal from-import '{node.module}'") assert not violations, f"Network isolation violations in domain/application layers:\n" + "\n".join(violations) def test_ports_and_adapters_contracts(): """Verify all domain ports and infrastructure adapters are defined and adhere to protocol conventions.""" # 1. Core Email Port & Adapter from app.modules.core.domain.ports.notification_port import EmailNotificationPort from app.modules.core.infrastructure.adapters.brevo_adapter import BrevoEmailAdapter assert issubclass(BrevoEmailAdapter, EmailNotificationPort) or hasattr(BrevoEmailAdapter, "send_email") # 2. Integrations Telegram Port & Adapter from app.modules.integrations.domain.ports.telegram_port import TelegramPort from app.modules.integrations.infrastructure.adapters.telegram_adapter import TelegramHttpAdapter assert issubclass(TelegramHttpAdapter, TelegramPort) or hasattr(TelegramHttpAdapter, "send_message") # 3. Operations Equipment Port & Adapter from app.modules.operations.domain.ports.equipment_repository_port import EquipmentRepositoryPort from app.modules.operations.infrastructure.repositories.equipment_repository import PostgresEquipmentRepository assert issubclass(PostgresEquipmentRepository, EquipmentRepositoryPort) or hasattr(PostgresEquipmentRepository, "record_fuel_log") # 4. Zalo CRM Gateway & Repository Ports & Adapters from app.modules.zalo_crm.domain.ports.zalo_gateway_port import ZaloGatewayPort from app.modules.zalo_crm.infrastructure.adapters.zalo_http_adapter import ZaloHttpAdapter assert issubclass(ZaloHttpAdapter, ZaloGatewayPort) or hasattr(ZaloHttpAdapter, "send_message") from app.modules.zalo_crm.domain.ports.zalo_crm_repository_port import ZaloCrmRepositoryPort from app.modules.zalo_crm.infrastructure.repositories.zalo_crm_repository import PostgresZaloCrmRepository assert issubclass(PostgresZaloCrmRepository, ZaloCrmRepositoryPort) or hasattr(PostgresZaloCrmRepository, "upsert_contact") # 5. Inventory SXD Crawler Port & Adapter from app.modules.inventory.domain.ports.sxd_crawler_port import ISxdCrawlerPort from app.modules.inventory.infrastructure.adapters.sxd_stealth_crawler_adapter import SxdStealthCrawlerAdapter assert issubclass(SxdStealthCrawlerAdapter, ISxdCrawlerPort) or hasattr(SxdStealthCrawlerAdapter, "crawl_latest_announcements") # 6. Invoices GDT Gateway Port & Adapter (Rule 6 Hexagonal Architecture) from app.modules.invoices.domain.ports.gdt_gateway_port import GdtGatewayPort from app.modules.invoices.infrastructure.adapters.gdt.gdt_adapter import GdtPortalAdapter assert issubclass(GdtPortalAdapter, GdtGatewayPort) or hasattr(GdtPortalAdapter, "authenticate_portal")