44 lines
1.5 KiB
Python
44 lines
1.5 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
from gyxx_flow.modules.shop_intelligence.config import TM_PLATFORM
|
|
from gyxx_flow.modules.shop_intelligence.writers import shop_base_writer
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_scheduled_content_workflows_do_not_auto_retry_failed_collection() -> None:
|
|
payload = json.loads(
|
|
(PROJECT_ROOT / "config" / "workflows.json").read_text(encoding="utf-8")
|
|
)
|
|
workflows = {item["id"]: item for item in payload["workflows"]}
|
|
commands = json.loads(
|
|
(PROJECT_ROOT / "config" / "commands.json").read_text(encoding="utf-8")
|
|
)["commands"]
|
|
scheduled_entries = {
|
|
step["entry"]
|
|
for workflow in workflows.values()
|
|
if workflow["module"] == "content_marketing"
|
|
and workflow["trigger"] == "scheduled"
|
|
for step in workflow["execution"]["steps"]
|
|
}
|
|
|
|
assert "data/tools/retry_failed.py" not in scheduled_entries
|
|
assert "content.retry_failed" not in workflows
|
|
assert "content.failed.retry" in {item["id"] for item in commands}
|
|
assert "data/tools/sync_metrics_to_cmt_notes.py" in scheduled_entries
|
|
|
|
|
|
def test_tmall_shop_metrics_never_write_to_feishu() -> None:
|
|
assert shop_base_writer.write_shop(TM_PLATFORM, "missing.json") is True
|
|
runner = (
|
|
PROJECT_ROOT
|
|
/ "src"
|
|
/ "gyxx_flow"
|
|
/ "modules"
|
|
/ "shop_intelligence"
|
|
/ "runners"
|
|
/ "run_shop.py"
|
|
).read_text(encoding="utf-8")
|
|
assert "write_shop(TM_PLATFORM" not in runner
|