39 lines
1.3 KiB
Python
39 lines
1.3 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.supply_chain import (
|
|
SUPPLY_RESOURCE,
|
|
SUPPLY_TIMEOUT_SECONDS,
|
|
SUPPLY_WORKFLOW_IDS,
|
|
SupplyChainModule,
|
|
)
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_supply_module_registers_only_scheduled_workflows() -> None:
|
|
module = SupplyChainModule.from_catalog(
|
|
WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
)
|
|
|
|
assert tuple(item.workflow_id for item in module.workflow_definitions()) == SUPPLY_WORKFLOW_IDS
|
|
assert len(SUPPLY_WORKFLOW_IDS) == 4
|
|
|
|
|
|
def test_supply_workflows_use_project_owned_module_commands() -> None:
|
|
module = SupplyChainModule.from_catalog(
|
|
WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
)
|
|
|
|
for definition in module.workflow_definitions():
|
|
assert definition.steps
|
|
for step in definition.steps:
|
|
assert isinstance(step.action, DeferredModuleCommandStep)
|
|
assert step.timeout_seconds == SUPPLY_TIMEOUT_SECONDS == 2 * 60 * 60
|
|
assert step.max_attempts == 1
|
|
assert step.resources == (SUPPLY_RESOURCE,) == ("module:supply_chain",)
|
|
assert step.production_sink is True
|