251 lines
8.5 KiB
Python
251 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
from pathlib import Path
|
|
|
|
from gyxx_flow.adapters.native import DeferredModuleCommandStep
|
|
from gyxx_flow.catalog import WorkflowCatalog
|
|
from gyxx_flow.modules.product_commerce import (
|
|
PRODUCT_MANUAL_WORKFLOW_IDS,
|
|
PRODUCT_RESOURCE,
|
|
PRODUCT_SCHEDULED_WORKFLOW_IDS,
|
|
PRODUCT_TIMEOUT_SECONDS,
|
|
PRODUCT_WORKFLOW_IDS,
|
|
TMALL_BAIBU_IMPORT_MODE_ALL,
|
|
TMALL_BAIBU_IMPORT_MODE_WITHOUT_HYPERLINKS,
|
|
ProductCommerceModule,
|
|
)
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_product_module_registers_scheduled_and_manual_entries() -> None:
|
|
module = ProductCommerceModule.from_catalog(
|
|
WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
)
|
|
assert tuple(item.workflow_id for item in module.workflow_definitions()) == PRODUCT_WORKFLOW_IDS
|
|
assert len(PRODUCT_SCHEDULED_WORKFLOW_IDS) == 12
|
|
assert PRODUCT_MANUAL_WORKFLOW_IDS == ()
|
|
|
|
|
|
def test_tmall_baibu_import_mode_filters_hyperlink_steps_and_rewires_dependencies() -> None:
|
|
catalog = WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
|
|
all_steps = next(
|
|
definition
|
|
for definition in ProductCommerceModule.from_catalog(
|
|
catalog,
|
|
tmall_baibu_import_mode=TMALL_BAIBU_IMPORT_MODE_ALL,
|
|
).workflow_definitions()
|
|
if definition.workflow_id == "product.tmall_baibu_apply"
|
|
).steps
|
|
filtered_steps = next(
|
|
definition
|
|
for definition in ProductCommerceModule.from_catalog(
|
|
catalog,
|
|
tmall_baibu_import_mode=TMALL_BAIBU_IMPORT_MODE_WITHOUT_HYPERLINKS,
|
|
).workflow_definitions()
|
|
if definition.workflow_id == "product.tmall_baibu_apply"
|
|
).steps
|
|
|
|
assert [step.step_id for step in all_steps] == [
|
|
"old_chain_bag",
|
|
"old_all_bag",
|
|
"old_all_3c",
|
|
"old_chain_3c",
|
|
"new_chain_bag",
|
|
"new_all_bag",
|
|
]
|
|
assert [step.step_id for step in filtered_steps] == [
|
|
"old_all_bag",
|
|
"old_all_3c",
|
|
"new_all_bag",
|
|
]
|
|
assert [step.depends_on for step in filtered_steps] == [
|
|
(),
|
|
("old_all_bag",),
|
|
("old_all_3c",),
|
|
]
|
|
assert all(step.replay_policy == "repeatable" for step in all_steps)
|
|
assert all(step.replay_policy == "repeatable" for step in filtered_steps)
|
|
|
|
def test_product_workflows_use_project_owned_module_commands() -> None:
|
|
module = ProductCommerceModule.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 == PRODUCT_TIMEOUT_SECONDS == 6 * 60 * 60
|
|
assert step.max_attempts == 1
|
|
if definition.workflow_id == "product.ecommerce_costs.daily":
|
|
assert step.resources == (
|
|
f"{PRODUCT_RESOURCE}:{definition.workflow_id}:{step.step_id}",
|
|
)
|
|
elif definition.workflow_id == "product.main_image.weekly":
|
|
assert step.resources == (f"{PRODUCT_RESOURCE}:{step.step_id}",)
|
|
elif definition.workflow_id in {
|
|
"product.daily",
|
|
"product.persona.daily",
|
|
"product.style_analysis.interval",
|
|
"product.video_upload",
|
|
"product.jd_video_upload",
|
|
"product.tmall_baibu_apply",
|
|
}:
|
|
assert step.resources == (
|
|
f"{PRODUCT_RESOURCE}:{definition.workflow_id}",
|
|
)
|
|
else:
|
|
assert step.resources == (PRODUCT_RESOURCE,) == (
|
|
"module:product_commerce",
|
|
)
|
|
assert step.production_sink is True
|
|
|
|
|
|
def test_workflow_scoped_product_resources_are_distinct() -> None:
|
|
module = ProductCommerceModule.from_catalog(
|
|
WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
)
|
|
resources = {
|
|
definition.workflow_id: definition.steps[0].resources
|
|
for definition in module.workflow_definitions()
|
|
if definition.workflow_id
|
|
in {
|
|
"product.daily",
|
|
"product.persona.daily",
|
|
"product.style_analysis.interval",
|
|
"product.ecommerce_costs.daily",
|
|
"product.video_upload",
|
|
"product.jd_video_upload",
|
|
"product.tmall_baibu_apply",
|
|
}
|
|
}
|
|
|
|
assert resources == {
|
|
"product.daily": ("module:product_commerce:product.daily",),
|
|
"product.persona.daily": (
|
|
"module:product_commerce:product.persona.daily",
|
|
),
|
|
"product.style_analysis.interval": (
|
|
"module:product_commerce:product.style_analysis.interval",
|
|
),
|
|
"product.ecommerce_costs.daily": (
|
|
"module:product_commerce:product.ecommerce_costs.daily:tmall_download",
|
|
),
|
|
"product.video_upload": (
|
|
"module:product_commerce:product.video_upload",
|
|
),
|
|
"product.jd_video_upload": (
|
|
"module:product_commerce:product.jd_video_upload",
|
|
),
|
|
"product.tmall_baibu_apply": (
|
|
"module:product_commerce:product.tmall_baibu_apply",
|
|
),
|
|
}
|
|
|
|
|
|
def test_persona_workflow_allows_idempotent_recovery_after_partial_platform_runs() -> None:
|
|
module = ProductCommerceModule.from_catalog(
|
|
WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
)
|
|
|
|
workflow = next(
|
|
item
|
|
for item in module.workflow_definitions()
|
|
if item.workflow_id == "product.persona.daily"
|
|
)
|
|
|
|
assert workflow.steps[0].replay_policy == "idempotent"
|
|
|
|
|
|
def test_combined_main_image_workflow_runs_platforms_independently() -> None:
|
|
module = ProductCommerceModule.from_catalog(
|
|
WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
)
|
|
|
|
workflow = next(
|
|
item
|
|
for item in module.workflow_definitions()
|
|
if item.workflow_id == "product.main_image.weekly"
|
|
)
|
|
assert [step.step_id for step in workflow.steps] == ["jd", "tmall"]
|
|
assert all(step.critical is True for step in workflow.steps)
|
|
assert all(step.depends_on == () for step in workflow.steps)
|
|
assert all(step.run_after_failure is False for step in workflow.steps)
|
|
assert [step.resources for step in workflow.steps] == [
|
|
("module:product_commerce:jd",),
|
|
("module:product_commerce:tmall",),
|
|
]
|
|
|
|
|
|
def test_tmall_video_upload_workflow_allows_idempotent_selected_store_rescans() -> None:
|
|
module = ProductCommerceModule.from_catalog(
|
|
WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
)
|
|
|
|
workflow = next(
|
|
item
|
|
for item in module.workflow_definitions()
|
|
if item.workflow_id == "product.video_upload"
|
|
)
|
|
assert len(workflow.steps) == 1
|
|
step = workflow.steps[0]
|
|
assert step.step_id == "upload"
|
|
assert step.replay_policy == "idempotent"
|
|
assert step.action.command_entry == "upload_video_to_guanghe.py"
|
|
assert step.action.command_args == (
|
|
"--execute",
|
|
"--keep-browser-open-on-failure",
|
|
)
|
|
|
|
|
|
def test_jd_video_upload_workflow_allows_idempotent_rescans() -> None:
|
|
module = ProductCommerceModule.from_catalog(
|
|
WorkflowCatalog.load(PROJECT_ROOT / "config")
|
|
)
|
|
|
|
workflow = next(
|
|
item
|
|
for item in module.workflow_definitions()
|
|
if item.workflow_id == "product.jd_video_upload"
|
|
)
|
|
assert len(workflow.steps) == 1
|
|
step = workflow.steps[0]
|
|
assert step.step_id == "upload"
|
|
assert step.replay_policy == "idempotent"
|
|
assert step.action.command_entry == "upload_video_to_jd.py"
|
|
assert step.action.command_args == (
|
|
"--anchor-record-id",
|
|
"recvrXTRWWFXeJ",
|
|
"--manual-login",
|
|
"--execute",
|
|
)
|
|
|
|
|
|
def test_product_runtime_compatibility_package_points_to_flattened_module() -> None:
|
|
module_root = (
|
|
PROJECT_ROOT
|
|
/ "src"
|
|
/ "gyxx_flow"
|
|
/ "modules"
|
|
/ "product_commerce"
|
|
).resolve()
|
|
compatibility = importlib.import_module(
|
|
"gyxx_flow.modules.product_commerce.runtime"
|
|
)
|
|
legacy_module = importlib.import_module(
|
|
"gyxx_flow.modules.product_commerce.runtime.market_rank_limits"
|
|
)
|
|
canonical_module = importlib.import_module(
|
|
"gyxx_flow.modules.product_commerce.market_rank_limits"
|
|
)
|
|
|
|
assert tuple(Path(path).resolve() for path in compatibility.__path__) == (
|
|
module_root,
|
|
)
|
|
assert Path(legacy_module.__file__).resolve() == Path(
|
|
canonical_module.__file__
|
|
).resolve()
|