43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from gyxx_flow.adapters.native import DeferredModuleCommandStep
|
|
from gyxx_flow.catalog import WorkflowCatalog
|
|
from gyxx_flow.modules import create_default_registry
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_default_runtime_uses_only_project_owned_module_steps() -> None:
|
|
registry = create_default_registry(
|
|
catalog=WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
)
|
|
|
|
for workflow_id in registry.workflow_ids:
|
|
definition = registry.workflow(workflow_id)
|
|
assert all(
|
|
isinstance(step.action, DeferredModuleCommandStep)
|
|
for step in definition.steps
|
|
)
|
|
|
|
|
|
def test_runtime_composition_has_no_old_root_environment_contract() -> None:
|
|
runtime_files = [
|
|
PROJECT_ROOT / "config" / "workflows.json",
|
|
*(
|
|
path
|
|
for path in (PROJECT_ROOT / "src" / "gyxx_flow" / "modules").glob("*/__init__.py")
|
|
),
|
|
PROJECT_ROOT / "src" / "gyxx_flow" / "adapters" / "native.py",
|
|
]
|
|
forbidden = ("GYXX_LEGACY_", "D:\\yingxiaoyunying", "D:\\shop-data-flow", "D:\\product-collector-analyze-flow", "E:\\auto-flow")
|
|
|
|
violations = [
|
|
f"{path}: {marker}"
|
|
for path in runtime_files
|
|
for marker in forbidden
|
|
if marker in path.read_text(encoding="utf-8")
|
|
]
|
|
assert violations == []
|