gyxx-flow/tests/test_business_modules.py

from __future__ import annotations

import ast
from pathlib import Path

import pytest

from gyxx_flow.catalog import WorkflowCatalog
from gyxx_flow.modules import (
    EXPECTED_MODULE_IDS,
    BusinessModule,
    ModuleRegistry,
    create_default_registry,
)
from gyxx_flow.modules.content_marketing import ContentMarketingModule
from gyxx_flow.modules.product_commerce import ProductCommerceModule
from gyxx_flow.modules.shop_intelligence import ShopIntelligenceModule
from gyxx_flow.modules.supply_chain import SupplyChainModule
from gyxx_flow.workflow import WorkflowDefinition


def test_default_registry_contains_exactly_the_four_business_modules() -> None:
    registry = create_default_registry()

    assert registry.module_ids == EXPECTED_MODULE_IDS
    assert isinstance(registry.get("content_marketing"), ContentMarketingModule)
    assert isinstance(registry.get("product_commerce"), ProductCommerceModule)
    assert isinstance(registry.get("shop_intelligence"), ShopIntelligenceModule)
    assert isinstance(registry.get("supply_chain"), SupplyChainModule)


def test_catalog_composition_registers_all_workflows() -> None:
    catalog = WorkflowCatalog.load(Path(__file__).parents[1] / "config")

    registry = create_default_registry(catalog=catalog)

    assert len(registry.workflow_ids) == 40
    assert set(registry.workflow_ids) == {
        "shop.metrics.weekly",
        "shop.competitor.weekly",
        "shop.jd_self_operated.daily",
        "shop.jd_customer_service.daily",
        "shop.jd_negative_reviews.daily",
        "shop.jd_after_sales_issues.daily",
        "shop.douyin_customer_service.daily",
        "shop.tmall_customer_service.daily",
        "shop.tmall_negative_reviews.daily",
        "shop.dy_negative_reviews.daily",
        "shop.douyin_after_sales_quality.daily",
        "shop.douyin_price_appeal",
        "supply.purchase_confirmation.daily",
        "supply.replenishment.weekly",
        "supply.replenishment_alert.daily",
        "supply.replenishment_arrival_sync.daily",
        "content.metrics.daily",
        "content.metrics.self_operated.daily",
        "content.metrics.backfill",
        "content.marketing_report.daily",
        "content.creator_report.monthly",
        "content.summary.monthly",
        "content.notes_master.daily",
        "content.comments.weekly",
        "content.summary.weekly",
        "product.persona.daily",
        "product.daily",
        "product.online_stock_alert.daily",
        "product.points_exchange.weekly",
        "product.hk_anpei_public_inventory.daily",
            "product.alert.daily",
            "product.ecommerce_costs.daily",
            "product.style_analysis.interval",
        "product.main_image.weekly",
        "product.sales_sheet.daily",
        "product.erp_all_shop_daily",
        "product.market_rank",
        "product.video_upload",
        "product.jd_video_upload",
        "product.tmall_baibu_apply",
    }


@pytest.mark.parametrize(
    ("module", "module_id", "expected_workflow_ids"),
    [
        (ContentMarketingModule(), "content_marketing", ()),
        (ProductCommerceModule(), "product_commerce", ()),
        (ShopIntelligenceModule(), "shop_intelligence", ()),
        (SupplyChainModule(), "supply_chain", ()),
    ],
)
def test_every_module_implements_the_uniform_contract(
    module: BusinessModule, module_id: str, expected_workflow_ids: tuple[str, ...]
) -> None:
    assert isinstance(module, BusinessModule)
    assert module.module_id == module_id
    assert tuple(
        definition.workflow_id for definition in module.workflow_definitions()
    ) == expected_workflow_ids


def test_registry_rejects_duplicate_module_ids() -> None:
    with pytest.raises(ValueError, match="duplicate module id"):
        ModuleRegistry([ContentMarketingModule(), ContentMarketingModule()])


def test_registry_indexes_workflows_without_exposing_mutable_state() -> None:
    class ExampleModule:
        module_id = "content_marketing"

        def workflow_definitions(self) -> tuple[WorkflowDefinition, ...]:
            return (WorkflowDefinition("content.example", ()),)

    registry = ModuleRegistry([ExampleModule()])

    assert registry.workflow_ids == ("content.example",)
    assert registry.workflow("content.example").workflow_id == "content.example"
    with pytest.raises(KeyError):
        registry.workflow("content.missing")


def test_business_packages_do_not_import_each_other_or_forbidden_gyxx_layers() -> None:
    source_root = Path(__file__).parents[1] / "src" / "gyxx_flow" / "modules"
    business_packages = set(EXPECTED_MODULE_IDS)
    allowed_shared_roots = {
        "gyxx_flow.core",
        "gyxx_flow.workflow",
        "gyxx_flow.adapters",
        "gyxx_flow.contracts",
        "gyxx_flow.modules.contracts",
    }
    violations: list[str] = []

    for package in sorted(business_packages):
        allowed_roots = allowed_shared_roots | {f"gyxx_flow.modules.{package}"}
        for path in (source_root / package).rglob("*.py"):
            tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
            for imported in _resolved_imports(tree, package):
                other_business = {
                    f"gyxx_flow.modules.{name}" for name in business_packages - {package}
                }
                if any(imported == root or imported.startswith(f"{root}.") for root in other_business):
                    violations.append(f"{path.name}: cross-module import {imported}")
                    continue
                if imported.startswith("gyxx_flow.") and not any(
                    imported == root or imported.startswith(f"{root}.")
                    for root in allowed_roots
                ):
                    violations.append(f"{path.name}: forbidden layer import {imported}")

    assert violations == []


def _resolved_imports(tree: ast.AST, package: str) -> tuple[str, ...]:
    names: list[str] = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            names.extend(alias.name for alias in node.names)
        elif isinstance(node, ast.ImportFrom):
            if node.level == 0 and node.module:
                names.append(node.module)
                continue
            base = ["gyxx_flow", "modules", package]
            parent = base[: len(base) - node.level + 1]
            if node.module:
                names.append(".".join([*parent, node.module]))
            else:
                names.extend(".".join([*parent, alias.name]) for alias in node.names)
    return tuple(names)