37 lines
1.3 KiB
Python
37 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from device.manager import DeviceManager
|
|
from host_agent.execution import create_execution_factories
|
|
from runtime.task import TaskRunner
|
|
from workflow.runner import WorkflowRunner
|
|
from workflow.store import WorkflowStore
|
|
|
|
|
|
def test_execution_factories_compose_existing_runtime_and_workflow(tmp_path) -> None:
|
|
manager = DeviceManager()
|
|
workflow_store = WorkflowStore(tmp_path / "workflows.sqlite3")
|
|
|
|
factories = create_execution_factories(
|
|
manager,
|
|
workflow_store=workflow_store,
|
|
)
|
|
task_runner = factories.task_runner_factory()
|
|
workflow_runner = factories.workflow_runner_factory()
|
|
|
|
assert isinstance(task_runner, TaskRunner)
|
|
assert isinstance(workflow_runner, WorkflowRunner)
|
|
assert workflow_runner.store is workflow_store
|
|
assert isinstance(workflow_runner.task_runner_factory(), TaskRunner)
|
|
|
|
|
|
def test_runtime_owned_packages_do_not_import_host_or_cloud_concerns() -> None:
|
|
root = Path(__file__).resolve().parents[3]
|
|
forbidden = ("import cloud", "from cloud", "import host_agent", "from host_agent")
|
|
|
|
for package in ("core", "device", "driver", "runtime", "tools"):
|
|
for path in (root / package).rglob("*.py"):
|
|
source = path.read_text(encoding="utf-8")
|
|
assert not any(token in source for token in forbidden), path
|