655 lines
22 KiB
Python
655 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gyxx_flow.catalog import CatalogError, WorkflowCatalog
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_catalog_maps_29_project_scheduled_tasks() -> None:
|
|
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
scheduled = catalog.scheduled_workflows()
|
|
|
|
assert len(scheduled) == 29
|
|
assert len({item.source_task_name for item in scheduled}) == 29
|
|
assert Counter(item.module for item in scheduled) == {
|
|
"content_marketing": 9,
|
|
"product_commerce": 12,
|
|
"shop_intelligence": 4,
|
|
"supply_chain": 4,
|
|
}
|
|
|
|
|
|
def test_catalog_schedules_douyin_price_appeal() -> None:
|
|
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
schedule = catalog.schedule_for("shop.douyin_price_appeal")
|
|
|
|
assert len(catalog.workflows) == 29
|
|
assert catalog.manual_workflows() == ()
|
|
assert "shop.douyin_price_appeal" in {
|
|
schedule.workflow_id for schedule in catalog.schedules
|
|
}
|
|
assert catalog.unavailable_workflows() == ()
|
|
assert schedule.kind == "daily"
|
|
assert schedule.effective_times == ("08:00", "16:00", "22:00")
|
|
|
|
|
|
def test_tmall_video_upload_is_an_idempotent_store_selected_scheduled_workflow() -> None:
|
|
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
workflow = next(
|
|
item for item in catalog.workflows if item.workflow_id == "product.video_upload"
|
|
)
|
|
|
|
assert workflow.trigger == "scheduled"
|
|
assert [step.step_id for step in workflow.steps] == ["upload"]
|
|
step = workflow.steps[0]
|
|
assert step.entry == "upload_video_to_guanghe.py"
|
|
assert step.args == (
|
|
"--execute",
|
|
"--keep-browser-open-on-failure",
|
|
)
|
|
assert step.replay_policy == "idempotent"
|
|
assert step.timeout_seconds == 21600
|
|
assert workflow.topic_config is not None
|
|
assert workflow.topic_config.default_keyword == "我的夏日焕新清单"
|
|
schedule = catalog.schedule_for(workflow.workflow_id)
|
|
assert schedule.kind == "interval_days"
|
|
assert schedule.at == "14:00"
|
|
assert schedule.every_days == 3
|
|
assert schedule.anchor_date == "2026-09-04"
|
|
assert schedule.enabled is True
|
|
|
|
|
|
def test_jd_video_upload_is_an_idempotent_scheduled_workflow() -> None:
|
|
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
workflow = next(
|
|
item
|
|
for item in catalog.workflows
|
|
if item.workflow_id == "product.jd_video_upload"
|
|
)
|
|
|
|
assert workflow.trigger == "scheduled"
|
|
assert [step.step_id for step in workflow.steps] == ["upload"]
|
|
step = workflow.steps[0]
|
|
assert step.entry == "upload_video_to_jd.py"
|
|
assert step.args == (
|
|
"--anchor-record-id",
|
|
"recvrXTRWWFXeJ",
|
|
"--manual-login",
|
|
"--execute",
|
|
)
|
|
assert step.replay_policy == "idempotent"
|
|
assert step.timeout_seconds == 21600
|
|
schedule = catalog.schedule_for(workflow.workflow_id)
|
|
assert schedule.kind == "interval_days"
|
|
assert schedule.at == "14:00"
|
|
assert schedule.every_days == 3
|
|
assert schedule.anchor_date == "2026-09-04"
|
|
assert schedule.enabled is True
|
|
|
|
|
|
def test_tmall_baibu_apply_is_a_six_step_hourly_workflow() -> None:
|
|
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
workflow = next(
|
|
item
|
|
for item in catalog.workflows
|
|
if item.workflow_id == "product.tmall_baibu_apply"
|
|
)
|
|
|
|
assert workflow.trigger == "scheduled"
|
|
assert [step.step_id for step in workflow.steps] == [
|
|
"old_chain_bag",
|
|
"old_all_bag",
|
|
"old_all_3c",
|
|
"old_chain_3c",
|
|
"new_chain_bag",
|
|
"new_all_bag",
|
|
]
|
|
assert [step.entry for step in workflow.steps] == [
|
|
"tmall_baibu_apply_old_chain_bag.py",
|
|
"tmall_baibu_apply_old_all_bag.py",
|
|
"tmall_baibu_apply_old_all_3c.py",
|
|
"tmall_baibu_apply_old_chain_3c.py",
|
|
"tmall_baibu_apply_new_chain_bag.py",
|
|
"tmall_baibu_apply_new_all_bag.py",
|
|
]
|
|
assert workflow.steps[0].depends_on == ()
|
|
assert all(
|
|
step.depends_on == (previous.step_id,)
|
|
for previous, step in zip(workflow.steps, workflow.steps[1:])
|
|
)
|
|
assert all(step.run_after_failure for step in workflow.steps[1:])
|
|
assert all(step.replay_policy == "repeatable" for step in workflow.steps)
|
|
assert [step.step_id for step in workflow.steps if step.has_hyperlink] == [
|
|
"old_chain_bag",
|
|
"old_chain_3c",
|
|
"new_chain_bag",
|
|
]
|
|
schedule = catalog.schedule_for(workflow.workflow_id)
|
|
assert schedule.kind == "daily"
|
|
assert schedule.enabled is True
|
|
assert schedule.business_date_offset_days == 0
|
|
assert schedule.effective_times == tuple(f"{hour:02d}:00" for hour in range(24))
|
|
|
|
|
|
def test_product_daily_passes_the_previous_business_day_explicitly() -> None:
|
|
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
workflow = next(
|
|
item for item in catalog.workflows if item.workflow_id == "product.daily"
|
|
)
|
|
|
|
assert workflow.steps[0].args == (
|
|
"--target-date",
|
|
"{business_date}",
|
|
"--stages",
|
|
"collect,import,analyze,export,insert",
|
|
)
|
|
assert catalog.schedule_for(workflow.workflow_id).business_date_offset_days == -1
|
|
|
|
|
|
def test_style_analysis_uses_the_previous_complete_business_day() -> None:
|
|
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
schedule = catalog.schedule_for("product.style_analysis.interval")
|
|
|
|
assert schedule.kind == "interval_days"
|
|
assert schedule.every_days == 3
|
|
assert schedule.business_date_offset_days == -1
|
|
|
|
|
|
def test_erp_all_shop_daily_runs_at_nine_for_the_previous_day() -> None:
|
|
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
workflow = next(
|
|
item
|
|
for item in catalog.workflows
|
|
if item.workflow_id == "product.erp_all_shop_daily"
|
|
)
|
|
|
|
assert workflow.steps[0].args == (
|
|
"--from",
|
|
"{business_date}",
|
|
"--to",
|
|
"{business_date}",
|
|
"--execute",
|
|
"--force",
|
|
)
|
|
schedule = catalog.schedule_for(workflow.workflow_id)
|
|
assert schedule.at == "09:00"
|
|
assert schedule.business_date_offset_days == -1
|
|
|
|
|
|
def test_every_scheduled_workflow_has_one_valid_asia_shanghai_schedule() -> None:
|
|
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
|
|
assert catalog.timezone == "Asia/Shanghai"
|
|
assert len(catalog.schedules) == 29
|
|
for workflow in catalog.scheduled_workflows():
|
|
schedule = catalog.schedule_for(workflow.workflow_id)
|
|
assert schedule.workflow_id == workflow.workflow_id
|
|
assert schedule.kind in {"daily", "weekly", "monthly", "interval_days"}
|
|
assert schedule.at.count(":") == 1
|
|
assert sum(schedule.enabled for schedule in catalog.schedules) == 25
|
|
|
|
|
|
def test_scheduled_workflows_use_python_entries_and_content_tasks_are_graphs() -> None:
|
|
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
|
|
for workflow in catalog.scheduled_workflows():
|
|
assert workflow.steps, f"scheduled workflow must declare LangGraph steps: {workflow.workflow_id}"
|
|
assert all(step.name and step.description for step in workflow.steps)
|
|
assert all(
|
|
step.data_flow
|
|
and step.data_flow.sources
|
|
and step.data_flow.processing
|
|
and step.data_flow.destinations
|
|
for step in workflow.steps
|
|
)
|
|
entries = [step.entry for step in workflow.steps]
|
|
assert all(
|
|
not entry.casefold().endswith((".bat", ".cmd", ".ps1"))
|
|
for entry in entries
|
|
)
|
|
|
|
daily = next(
|
|
workflow
|
|
for workflow in catalog.workflows
|
|
if workflow.workflow_id == "content.metrics.daily"
|
|
)
|
|
assert [step.step_id for step in daily.steps] == [
|
|
"collect_collaborators",
|
|
"refresh_self_mapping",
|
|
"collect_self_bilibili",
|
|
"collect_self_douyin",
|
|
"sync",
|
|
]
|
|
assert daily.steps[0].depends_on == ()
|
|
assert daily.steps[1].depends_on == ("collect_collaborators",)
|
|
assert daily.steps[2].depends_on == ("refresh_self_mapping",)
|
|
assert daily.steps[3].depends_on == ("collect_self_bilibili",)
|
|
assert daily.steps[4].depends_on == ("collect_self_douyin",)
|
|
assert "--include-self-operated" not in daily.steps[0].args
|
|
assert all(step.replay_policy == "repeatable" for step in daily.steps)
|
|
assert catalog.schedule_for(daily.workflow_id).effective_times == (
|
|
"01:00",
|
|
"05:00",
|
|
)
|
|
|
|
backfill = next(
|
|
workflow
|
|
for workflow in catalog.workflows
|
|
if workflow.workflow_id == "content.metrics.backfill"
|
|
)
|
|
assert [step.step_id for step in backfill.steps] == [
|
|
"collect_collaborators",
|
|
"sync",
|
|
]
|
|
assert backfill.steps[0].entry == "run_all.py"
|
|
assert backfill.steps[0].args == ("--daily-scope",)
|
|
assert backfill.steps[1].entry == "data/tools/sync_metrics_to_cmt_notes.py"
|
|
assert backfill.steps[1].depends_on == ("collect_collaborators",)
|
|
assert backfill.steps[1].run_after_failure is True
|
|
backfill_schedule = catalog.schedule_for(backfill.workflow_id)
|
|
assert backfill_schedule.kind == "weekly"
|
|
assert backfill_schedule.days == ("Monday",)
|
|
assert backfill_schedule.at == "14:00"
|
|
|
|
marketing_report = next(
|
|
workflow
|
|
for workflow in catalog.workflows
|
|
if workflow.workflow_id == "content.marketing_report.daily"
|
|
)
|
|
report_flow = marketing_report.steps[0].data_flow
|
|
assert report_flow is not None
|
|
assert {source.system for source in report_flow.sources} == {
|
|
"PostgreSQL",
|
|
"飞书",
|
|
}
|
|
assert "调用 Hermes 生成营销分析" in report_flow.processing
|
|
assert any(
|
|
destination.system == "飞书" and destination.condition
|
|
for destination in report_flow.destinations
|
|
)
|
|
|
|
notes_master = next(
|
|
workflow
|
|
for workflow in catalog.workflows
|
|
if workflow.workflow_id == "content.notes_master.daily"
|
|
)
|
|
assert notes_master.steps[0].entry == "data/tools/sync_notes_master.py"
|
|
assert notes_master.steps[0].args == ("--execute",)
|
|
assert catalog.schedule_for(notes_master.workflow_id).at == "09:00"
|
|
|
|
comments = next(
|
|
workflow
|
|
for workflow in catalog.workflows
|
|
if workflow.workflow_id == "content.comments.weekly"
|
|
)
|
|
assert [step.step_id for step in comments.steps] == [
|
|
"bilibili",
|
|
"xiaohongshu",
|
|
"douyin",
|
|
]
|
|
assert all(step.depends_on == () for step in comments.steps)
|
|
assert "三平台并行启动" in comments.note
|
|
|
|
main_image = next(
|
|
workflow
|
|
for workflow in catalog.workflows
|
|
if workflow.workflow_id == "product.main_image.weekly"
|
|
)
|
|
assert [step.step_id for step in main_image.steps] == ["jd", "tmall"]
|
|
assert [step.entry for step in main_image.steps] == [
|
|
"run_weekly_jd_main_image.py",
|
|
"run_weekly_main_image.py",
|
|
]
|
|
assert all(step.depends_on == () for step in main_image.steps)
|
|
assert all(step.run_after_failure is False for step in main_image.steps)
|
|
assert "并行执行京东与天猫" in main_image.note
|
|
assert catalog.schedule_for(main_image.workflow_id).at == "08:30"
|
|
|
|
shop_metrics = next(
|
|
workflow
|
|
for workflow in catalog.workflows
|
|
if workflow.workflow_id == "shop.metrics.weekly"
|
|
)
|
|
assert [step.step_id for step in shop_metrics.steps] == ["jd", "dy", "tm"]
|
|
assert shop_metrics.steps[0].args == ("--platform", "jd", "--skip-feishu")
|
|
assert shop_metrics.steps[1].args == ("--platform", "dy", "--skip-feishu")
|
|
assert all(
|
|
destination.system != "飞书 Base"
|
|
for step in shop_metrics.steps
|
|
for destination in step.data_flow.destinations
|
|
)
|
|
assert "店铺经营指标周采集不写飞书 Base" in shop_metrics.note
|
|
|
|
|
|
def test_shop_steps_preserve_declared_replay_policy_at_module_boundary() -> None:
|
|
from gyxx_flow.modules.shop_intelligence import ShopIntelligenceModule
|
|
|
|
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
definitions = ShopIntelligenceModule.from_catalog(catalog).workflow_definitions()
|
|
|
|
policies = {
|
|
workflow.workflow_id: {step.replay_policy for step in workflow.steps}
|
|
for workflow in definitions
|
|
}
|
|
|
|
assert policies["shop.metrics.weekly"] == {"idempotent"}
|
|
assert policies["shop.competitor.weekly"] == {"idempotent"}
|
|
assert policies["shop.jd_self_operated.daily"] == {"idempotent"}
|
|
assert policies["shop.douyin_price_appeal"] == {"idempotent"}
|
|
|
|
|
|
def test_product_daily_includes_the_idempotent_import_stage() -> None:
|
|
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
workflow = next(
|
|
item
|
|
for item in catalog.workflows
|
|
if item.workflow_id == "product.daily"
|
|
)
|
|
|
|
assert "collect,import,analyze,export,insert" in workflow.steps[0].args
|
|
assert not any(
|
|
item.workflow_id == "product.import.daily" for item in catalog.workflows
|
|
)
|
|
|
|
|
|
def test_catalog_loads_optional_step_timeout_seconds(tmp_path: Path) -> None:
|
|
(tmp_path / "workflows.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 3,
|
|
"workflows": [
|
|
{
|
|
"id": "content.timeout_probe",
|
|
"module": "content_marketing",
|
|
"trigger": "manual",
|
|
"execution": {
|
|
"steps": [
|
|
{
|
|
"id": "run",
|
|
"entry": "run.py",
|
|
"timeout_seconds": 12.5,
|
|
}
|
|
]
|
|
},
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(tmp_path / "schedules.json").write_text(
|
|
'{"schema_version":1,"timezone":"Asia/Shanghai","schedules":[]}',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
catalog = WorkflowCatalog.load(tmp_path)
|
|
|
|
assert catalog.workflows[0].steps[0].timeout_seconds == 12.5
|
|
|
|
|
|
@pytest.mark.parametrize("timeout_seconds", [0, -1, True, "60", float("inf")])
|
|
def test_catalog_rejects_invalid_step_timeout_seconds(
|
|
tmp_path: Path,
|
|
timeout_seconds: object,
|
|
) -> None:
|
|
(tmp_path / "workflows.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 3,
|
|
"workflows": [
|
|
{
|
|
"id": "content.timeout_probe",
|
|
"module": "content_marketing",
|
|
"trigger": "manual",
|
|
"execution": {
|
|
"steps": [
|
|
{
|
|
"id": "run",
|
|
"entry": "run.py",
|
|
"timeout_seconds": timeout_seconds,
|
|
}
|
|
]
|
|
},
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(tmp_path / "schedules.json").write_text(
|
|
'{"schema_version":1,"timezone":"Asia/Shanghai","schedules":[]}',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with pytest.raises(CatalogError, match="workflow step timeout_seconds is invalid"):
|
|
WorkflowCatalog.load(tmp_path)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value", "message"),
|
|
[
|
|
("name", "", "workflow step name is invalid"),
|
|
("description", 42, "workflow step description is invalid"),
|
|
],
|
|
)
|
|
def test_catalog_rejects_invalid_step_explanation(
|
|
tmp_path: Path,
|
|
field: str,
|
|
value: object,
|
|
message: str,
|
|
) -> None:
|
|
step = {
|
|
"id": "run",
|
|
"entry": "run.py",
|
|
"name": "运行任务",
|
|
"description": "执行测试任务。",
|
|
field: value,
|
|
}
|
|
(tmp_path / "workflows.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 3,
|
|
"workflows": [
|
|
{
|
|
"id": "a.one",
|
|
"module": "content_marketing",
|
|
"trigger": "manual",
|
|
"execution": {"steps": [step]},
|
|
}
|
|
],
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(tmp_path / "schedules.json").write_text(
|
|
'{"schema_version":1,"timezone":"Asia/Shanghai","schedules":[]}',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with pytest.raises(CatalogError, match=message):
|
|
WorkflowCatalog.load(tmp_path)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("data_flow", "message"),
|
|
[
|
|
([], "data_flow must be an object"),
|
|
(
|
|
{"sources": [], "processing": ["处理"], "destinations": []},
|
|
"sources is invalid",
|
|
),
|
|
(
|
|
{
|
|
"sources": [{"label": "输入"}],
|
|
"processing": [],
|
|
"destinations": [{"label": "输出"}],
|
|
},
|
|
"processing is invalid",
|
|
),
|
|
(
|
|
{
|
|
"sources": [{"label": "输入"}],
|
|
"processing": ["处理"],
|
|
"destinations": [{"label": ""}],
|
|
},
|
|
"destinations is invalid",
|
|
),
|
|
],
|
|
)
|
|
def test_catalog_rejects_invalid_step_data_flow(
|
|
tmp_path: Path,
|
|
data_flow: object,
|
|
message: str,
|
|
) -> None:
|
|
(tmp_path / "workflows.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 3,
|
|
"workflows": [
|
|
{
|
|
"id": "a.one",
|
|
"module": "content_marketing",
|
|
"trigger": "manual",
|
|
"execution": {
|
|
"steps": [
|
|
{
|
|
"id": "run",
|
|
"entry": "run.py",
|
|
"name": "运行任务",
|
|
"description": "执行测试任务。",
|
|
"data_flow": data_flow,
|
|
}
|
|
]
|
|
},
|
|
}
|
|
],
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(tmp_path / "schedules.json").write_text(
|
|
'{"schema_version":1,"timezone":"Asia/Shanghai","schedules":[]}',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with pytest.raises(CatalogError, match=message):
|
|
WorkflowCatalog.load(tmp_path)
|
|
|
|
|
|
def test_catalog_rejects_duplicate_workflow_ids(tmp_path: Path) -> None:
|
|
(tmp_path / "workflows.json").write_text(
|
|
'{"schema_version":3,"workflows":['
|
|
'{"id":"a.one","module":"content_marketing","trigger":"manual",'
|
|
'"execution":{"entry":"run.py"}},'
|
|
'{"id":"a.one","module":"content_marketing","trigger":"manual",'
|
|
'"execution":{"entry":"other.py"}}]}',
|
|
encoding="utf-8",
|
|
)
|
|
(tmp_path / "schedules.json").write_text(
|
|
'{"schema_version":1,"timezone":"Asia/Shanghai","schedules":[]}',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with pytest.raises(CatalogError, match="duplicate workflow id"):
|
|
WorkflowCatalog.load(tmp_path)
|
|
|
|
|
|
def test_catalog_rejects_absolute_execution_entry(tmp_path: Path) -> None:
|
|
(tmp_path / "workflows.json").write_text(
|
|
'{"schema_version":3,"workflows":['
|
|
'{"id":"a.one","module":"content_marketing","trigger":"manual",'
|
|
'"execution":{"entry":"D:/old/run.py"}}]}',
|
|
encoding="utf-8",
|
|
)
|
|
(tmp_path / "schedules.json").write_text(
|
|
'{"schema_version":1,"timezone":"Asia/Shanghai","schedules":[]}',
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with pytest.raises(CatalogError, match="relative"):
|
|
WorkflowCatalog.load(tmp_path)
|
|
|
|
|
|
def _write_single_scheduled_catalog(tmp_path: Path, at: object) -> None:
|
|
(tmp_path / "workflows.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 3,
|
|
"workflows": [
|
|
{
|
|
"id": "shop.price_appeal",
|
|
"module": "shop_intelligence",
|
|
"trigger": "scheduled",
|
|
"execution": {"entry": "run.py"},
|
|
"provenance": {"task_name": "price appeal"},
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(tmp_path / "schedules.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 1,
|
|
"timezone": "Asia/Shanghai",
|
|
"schedules": [
|
|
{
|
|
"workflow_id": "shop.price_appeal",
|
|
"kind": "daily",
|
|
"at": at,
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def test_catalog_loads_multiple_daily_schedule_times(tmp_path: Path) -> None:
|
|
_write_single_scheduled_catalog(tmp_path, ["08:00", "16:00", "22:00"])
|
|
|
|
schedule = WorkflowCatalog.load(tmp_path).schedules[0]
|
|
|
|
assert schedule.at == "08:00"
|
|
assert schedule.at_times == ("08:00", "16:00", "22:00")
|
|
assert schedule.effective_times == ("08:00", "16:00", "22:00")
|
|
|
|
|
|
def test_catalog_preserves_legacy_single_schedule_time(tmp_path: Path) -> None:
|
|
_write_single_scheduled_catalog(tmp_path, "08:00")
|
|
|
|
schedule = WorkflowCatalog.load(tmp_path).schedules[0]
|
|
|
|
assert schedule.at == "08:00"
|
|
assert schedule.at_times == ()
|
|
assert schedule.effective_times == ("08:00",)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"at",
|
|
[
|
|
[],
|
|
["08:00", "08:00"],
|
|
["08:00", "24:00"],
|
|
["08:00", 16],
|
|
{"morning": "08:00"},
|
|
],
|
|
)
|
|
def test_catalog_rejects_invalid_multiple_schedule_times(
|
|
tmp_path: Path,
|
|
at: object,
|
|
) -> None:
|
|
_write_single_scheduled_catalog(tmp_path, at)
|
|
|
|
with pytest.raises(CatalogError, match="invalid schedule time"):
|
|
WorkflowCatalog.load(tmp_path)
|