606 lines
20 KiB
Python
606 lines
20 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import threading
|
||
from pathlib import Path
|
||
from typing import Any, Mapping
|
||
from urllib.request import Request, urlopen
|
||
|
||
import pytest
|
||
|
||
from gyxx_flow.adapters.workflow_config import (
|
||
PostgresWorkflowConfigStore,
|
||
WorkflowConfigConflictError,
|
||
merge_lifecycle_style_records,
|
||
normalize_config,
|
||
normalize_style_config,
|
||
)
|
||
from gyxx_flow.console import (
|
||
ConsoleConflictError,
|
||
WorkflowConsoleService,
|
||
create_console_server,
|
||
)
|
||
from gyxx_flow.core.config import Settings
|
||
from gyxx_flow.modules.product_commerce.config.style_config_loader import (
|
||
StyleConfigLoader,
|
||
)
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||
|
||
|
||
class FakeDynamicConfigStore:
|
||
def __init__(self) -> None:
|
||
self.records = [
|
||
{
|
||
"id": 1,
|
||
"style_name": "极星pro",
|
||
"erp_codes": ["10398"],
|
||
"platforms": [
|
||
{
|
||
"id": 11,
|
||
"platform": "天猫",
|
||
"item_ids": ["832214449817"],
|
||
"persona_bitable_url": "",
|
||
"enabled": True,
|
||
}
|
||
],
|
||
"enabled": True,
|
||
"revision": 1,
|
||
}
|
||
]
|
||
|
||
def snapshot(
|
||
self,
|
||
*,
|
||
search: str = "",
|
||
platform: str = "",
|
||
enabled: bool | None = None,
|
||
) -> dict[str, Any]:
|
||
del search, platform, enabled
|
||
return {
|
||
"styles": self.records,
|
||
"summary": {"total": 1, "enabled": 1, "platform_configs": 1, "matched": 1},
|
||
"platforms": [{"name": "天猫", "count": 1}],
|
||
"source": "postgresql",
|
||
}
|
||
|
||
def runtime_records(self) -> list[dict[str, Any]]:
|
||
return self.records
|
||
|
||
def create(self, payload: Mapping[str, Any]) -> dict[str, Any]:
|
||
return {"id": 2, **normalize_style_config(payload), "revision": 1}
|
||
|
||
def update(
|
||
self,
|
||
config_id: int,
|
||
payload: Mapping[str, Any],
|
||
*,
|
||
expected_revision: int,
|
||
) -> dict[str, Any]:
|
||
if expected_revision != 1:
|
||
raise WorkflowConfigConflictError("配置已被其他人更新，请重新载入")
|
||
return {
|
||
"id": config_id,
|
||
**normalize_style_config(payload),
|
||
"revision": 2,
|
||
}
|
||
|
||
def delete(self, config_id: int, *, expected_revision: int) -> None:
|
||
del config_id
|
||
if expected_revision != 1:
|
||
raise WorkflowConfigConflictError("配置已被其他人更新，请重新载入")
|
||
|
||
|
||
def _payload() -> dict[str, Any]:
|
||
return {
|
||
"style_name": "极星pro",
|
||
"brand": "光影行星",
|
||
"erp_codes": "10398，10398, 10400",
|
||
"note": "",
|
||
"sales_bitable_url": "https://example.feishu.cn/base/abc?table=tbl1",
|
||
"main_image_bitable_url": "",
|
||
"creator_bitable_url": "",
|
||
"self_creator_bitable_url": "",
|
||
"weekly_note_analysis_url": "",
|
||
"style_analysis_bitable_url": "",
|
||
"style_content": "",
|
||
"enabled": True,
|
||
"platforms": [
|
||
{
|
||
"platform": "天猫",
|
||
"item_ids": ["832214449817", "832214449817"],
|
||
"persona_bitable_url": "",
|
||
"enabled": True,
|
||
}
|
||
],
|
||
}
|
||
|
||
|
||
def test_normalize_config_cleans_ids_and_markdown_urls() -> None:
|
||
payload = {
|
||
"platform": "天猫",
|
||
"style_name": "极星pro",
|
||
"brand": "光影行星",
|
||
"erp_codes": "10398，10398, 10400",
|
||
"item_ids": ["832214449817", "832214449817"],
|
||
"note": "",
|
||
"sales_bitable_url": "https://example.feishu.cn/base/abc?table=tbl1",
|
||
"main_image_bitable_url": "",
|
||
"creator_bitable_url": "",
|
||
"tm_persona_bitable_url": "",
|
||
"jd_persona_bitable_url": "",
|
||
"dy_persona_bitable_url": "",
|
||
"self_creator_bitable_url": "",
|
||
"weekly_note_analysis_url": "",
|
||
"style_analysis_bitable_url": "",
|
||
"style_content": "",
|
||
"enabled": True,
|
||
}
|
||
payload["sales_bitable_url"] = "[销量表](https://example.feishu.cn/base/abc?table=tbl1)"
|
||
|
||
normalized = normalize_config(payload)
|
||
|
||
assert normalized["erp_codes"] == ["10398", "10400"]
|
||
assert normalized["item_ids"] == ["832214449817"]
|
||
assert normalized["sales_bitable_url"] == (
|
||
"https://example.feishu.cn/base/abc?table=tbl1"
|
||
)
|
||
|
||
|
||
def test_normalize_style_config_owns_erp_once_and_platform_ids_per_child() -> None:
|
||
normalized = normalize_style_config(_payload())
|
||
|
||
assert normalized["erp_codes"] == ["10398", "10400"]
|
||
assert normalized["platforms"] == [
|
||
{
|
||
"platform": "天猫",
|
||
"item_ids": ["832214449817"],
|
||
"persona_bitable_url": "",
|
||
"enabled": True,
|
||
}
|
||
]
|
||
|
||
|
||
def test_runtime_records_use_lifecycle_identity_and_keep_dynamic_targets() -> None:
|
||
dynamic_records = [
|
||
{
|
||
"style_name": "款式甲",
|
||
"brand": "旧品牌",
|
||
"erp_codes": ["OLD-ERP"],
|
||
"platform": "天猫",
|
||
"item_ids": ["OLD-TM"],
|
||
"sales_bitable_url": "https://example.feishu.cn/base/sales?table=tblSales",
|
||
"tm_persona_bitable_url": "tm-persona",
|
||
}
|
||
]
|
||
lifecycle_rows = [
|
||
{
|
||
"id": 8,
|
||
"name": "款式甲",
|
||
"brand": "光影行星",
|
||
"erp_style_codes": ["NEW-ERP"],
|
||
"platform_product_ids": {
|
||
"tm": ["NEW-TM"],
|
||
"jd": ["NEW-JD"],
|
||
"jd_self": [],
|
||
"dy": [],
|
||
},
|
||
}
|
||
]
|
||
|
||
records = merge_lifecycle_style_records(dynamic_records, lifecycle_rows)
|
||
|
||
by_platform = {row["platform"]: row for row in records}
|
||
assert by_platform["天猫"]["item_ids"] == ["NEW-TM"]
|
||
assert by_platform["京东旗舰店"]["item_ids"] == ["NEW-JD"]
|
||
assert by_platform["京东自营"]["item_ids"] == []
|
||
assert by_platform["抖音"]["item_ids"] == []
|
||
assert all(row["erp_codes"] == ["NEW-ERP"] for row in records)
|
||
assert by_platform["天猫"]["sales_bitable_url"].endswith("table=tblSales")
|
||
assert by_platform["天猫"]["tm_persona_bitable_url"] == "tm-persona"
|
||
|
||
|
||
def test_runtime_records_empty_lifecycle_ids_clear_legacy_platform_ids() -> None:
|
||
records = merge_lifecycle_style_records(
|
||
[
|
||
{
|
||
"style_name": "研发款",
|
||
"erp_codes": ["ERP-1"],
|
||
"platform": "天猫",
|
||
"item_ids": ["LEGACY-TM"],
|
||
}
|
||
],
|
||
[
|
||
{
|
||
"id": 9,
|
||
"name": "研发款",
|
||
"erp_style_codes": ["ERP-1"],
|
||
"platform_product_ids": {
|
||
"tm": [],
|
||
"jd": [],
|
||
"jd_self": [],
|
||
"dy": [],
|
||
},
|
||
}
|
||
],
|
||
)
|
||
|
||
assert all(not row["item_ids"] for row in records)
|
||
|
||
|
||
def test_lifecycle_merge_respects_disabled_dynamic_platform() -> None:
|
||
records = merge_lifecycle_style_records(
|
||
[
|
||
{
|
||
"style_name": "款式甲",
|
||
"platform": "天猫",
|
||
"item_ids": ["TM-1"],
|
||
"enabled": False,
|
||
},
|
||
{
|
||
"style_name": "款式甲",
|
||
"platform": "京东旗舰店",
|
||
"item_ids": ["JD-1"],
|
||
"enabled": True,
|
||
},
|
||
],
|
||
[
|
||
{
|
||
"id": 10,
|
||
"name": "款式甲",
|
||
"erp_style_codes": ["ERP-1"],
|
||
"platform_product_ids": {
|
||
"tm": ["TM-1"],
|
||
"jd": ["JD-1"],
|
||
"jd_self": [],
|
||
"dy": [],
|
||
},
|
||
}
|
||
],
|
||
)
|
||
|
||
assert {row["platform"] for row in records} == {"京东旗舰店", "京东自营", "抖音"}
|
||
|
||
|
||
def test_normalize_style_config_supports_extensible_business_targets() -> None:
|
||
payload = _payload()
|
||
payload["destinations"] = [
|
||
{
|
||
"key": "sales_bitable_url",
|
||
"label": "销量表",
|
||
"url": "https://example.feishu.cn/base/sales?table=tblSales",
|
||
"description": "销量日报写入目标",
|
||
"enabled": True,
|
||
},
|
||
{
|
||
"key": "style_quality_analysis",
|
||
"label": "款式质量分析",
|
||
"url": "https://example.feishu.cn/base/quality?table=tblQuality",
|
||
"description": "未来分析逻辑写入的结果表",
|
||
"enabled": True,
|
||
},
|
||
]
|
||
|
||
normalized = normalize_style_config(payload)
|
||
|
||
assert [item["key"] for item in normalized["destinations"]] == [
|
||
"sales_bitable_url",
|
||
"style_quality_analysis",
|
||
]
|
||
assert normalized["sales_bitable_url"] == (
|
||
"https://example.feishu.cn/base/sales?table=tblSales"
|
||
)
|
||
assert normalized["destinations"][1]["description"] == (
|
||
"未来分析逻辑写入的结果表"
|
||
)
|
||
|
||
|
||
def test_postgres_store_omits_empty_dsn_for_split_connection_fields(
|
||
monkeypatch,
|
||
) -> None:
|
||
from gyxx_flow.adapters import workflow_config
|
||
|
||
calls: list[dict[str, Any]] = []
|
||
|
||
class FakeConnection:
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *_args) -> None:
|
||
pass
|
||
|
||
def close(self) -> None:
|
||
pass
|
||
|
||
def connect(**kwargs):
|
||
calls.append(kwargs)
|
||
return FakeConnection()
|
||
|
||
monkeypatch.setattr(workflow_config.psycopg, "connect", connect)
|
||
store = PostgresWorkflowConfigStore(
|
||
{
|
||
"PG_HOST": "db.example.test",
|
||
"PG_PORT": "5432",
|
||
"PG_DB": "app",
|
||
"PG_USER": "app",
|
||
"PG_PASSWORD": "test-only",
|
||
}
|
||
)
|
||
|
||
with store._connection():
|
||
pass
|
||
|
||
assert calls[0]["host"] == "db.example.test"
|
||
assert "conninfo" not in calls[0]
|
||
|
||
|
||
def test_migration_seed_captures_all_feishu_records() -> None:
|
||
seed = json.loads(
|
||
(PROJECT_ROOT / "config" / "workflow-dynamic-config-seed.json").read_text(
|
||
encoding="utf-8"
|
||
)
|
||
)
|
||
|
||
assert seed["record_count"] == 215
|
||
assert len(seed["records"]) == 215
|
||
assert len({item["source_record_id"] for item in seed["records"]}) == 215
|
||
assert any(
|
||
item["config"]["style_name"] == "极星pro"
|
||
and item["config"]["platform"] == "天猫"
|
||
for item in seed["records"]
|
||
)
|
||
|
||
|
||
def test_style_loader_reads_database_provider_and_never_needs_feishu(
|
||
tmp_path: Path,
|
||
) -> None:
|
||
loader = StyleConfigLoader(
|
||
cache_path=tmp_path / "style-cache.json",
|
||
records_provider=lambda: [
|
||
{
|
||
**normalize_config(
|
||
{
|
||
"platform": "天猫",
|
||
"style_name": "极星pro",
|
||
"brand": "光影行星",
|
||
"erp_codes": ["10398", "10400"],
|
||
"item_ids": ["832214449817"],
|
||
"note": "",
|
||
"sales_bitable_url": "https://example.feishu.cn/base/abc?table=tbl1",
|
||
"main_image_bitable_url": "",
|
||
"creator_bitable_url": "",
|
||
"tm_persona_bitable_url": "",
|
||
"jd_persona_bitable_url": "",
|
||
"dy_persona_bitable_url": "",
|
||
"self_creator_bitable_url": "",
|
||
"weekly_note_analysis_url": "",
|
||
"style_analysis_bitable_url": "",
|
||
"style_content": "",
|
||
"enabled": True,
|
||
}
|
||
),
|
||
"sales_bitable_url": (
|
||
"https://example.feishu.cn/base/abc?table=tbl1&view=vew1"
|
||
),
|
||
}
|
||
],
|
||
)
|
||
|
||
loaded = loader.load()
|
||
|
||
assert loaded["source"] == "database"
|
||
assert loaded["styles"]["极星pro"]["tm"]["item_ids"] == [
|
||
"832214449817"
|
||
]
|
||
assert loaded["styles"]["极星pro"]["erp_codes"] == ["10398", "10400"]
|
||
|
||
|
||
def test_style_loader_preserves_extensible_business_targets(tmp_path: Path) -> None:
|
||
destinations = [
|
||
{
|
||
"key": "style_quality_analysis",
|
||
"label": "款式质量分析",
|
||
"url": "https://example.feishu.cn/base/quality?table=tblQuality",
|
||
"description": "质量分析结果表",
|
||
"enabled": True,
|
||
}
|
||
]
|
||
loader = StyleConfigLoader(
|
||
cache_path=tmp_path / "style-cache.json",
|
||
records_provider=lambda: [
|
||
{
|
||
"platform": "天猫",
|
||
"style_name": "极星pro",
|
||
"brand": "光影行星",
|
||
"erp_codes": ["10398"],
|
||
"item_ids": ["832214449817"],
|
||
"sales_bitable_url": (
|
||
"https://example.feishu.cn/base/sales?table=tblSales"
|
||
),
|
||
"destinations": destinations,
|
||
}
|
||
],
|
||
)
|
||
|
||
loaded = loader.load()
|
||
|
||
assert loaded["styles"]["极星pro"]["destinations"] == destinations
|
||
|
||
|
||
def test_grouped_database_keeps_original_platform_and_target_contracts() -> None:
|
||
shared = {
|
||
"style_name": "款式甲",
|
||
"brand": "光影行星",
|
||
"erp_codes": ["ERP-1"],
|
||
"sales_bitable_url": "https://example.feishu.cn/base/sales?table=tblSales",
|
||
"main_image_bitable_url": "https://example.feishu.cn/base/image?table=tblImage",
|
||
"tm_persona_bitable_url": "https://example.feishu.cn/base/tm?table=tblTm",
|
||
"jd_persona_bitable_url": "https://example.feishu.cn/base/jd?table=tblJd",
|
||
"dy_persona_bitable_url": "https://example.feishu.cn/base/dy?table=tblDy",
|
||
"style_analysis_bitable_url": "https://example.feishu.cn/base/report?table=tblReport",
|
||
}
|
||
rows = [
|
||
{**shared, "platform": "天猫", "item_ids": ["TM-1"]},
|
||
{**shared, "platform": "京东旗舰店", "item_ids": ["JD-SPU-1"]},
|
||
{**shared, "platform": "京东自营", "item_ids": ["JD-SKU-1"]},
|
||
{**shared, "platform": "抖音", "item_ids": ["DY-1"]},
|
||
]
|
||
loader = StyleConfigLoader(records_provider=lambda: rows)
|
||
|
||
style = loader._aggregate(loader._load_records_from_database())["styles"]["款式甲"]
|
||
|
||
assert style["erp_codes"] == ["ERP-1"]
|
||
assert style["tm"]["item_ids"] == ["TM-1"]
|
||
assert style["jd"]["spus"] == ["JD-SPU-1"]
|
||
assert style["jd_self"]["skus"] == ["JD-SKU-1"]
|
||
assert style["dy"]["item_ids"] == ["DY-1"]
|
||
assert style["sales_bitable"]["table_id"] == "tblSales"
|
||
assert style["main_image_bitable"]["table_id"] == "tblImage"
|
||
assert style["persona_bitable"]["table_id"] == "tblTm"
|
||
assert style["jd_persona_bitable"]["table_id"] == "tblJd"
|
||
assert style["dy_persona_bitable"]["table_id"] == "tblDy"
|
||
assert style["style_analysis_bitable"]["table_id"] == "tblReport"
|
||
|
||
|
||
def test_console_dynamic_config_crud_maps_optimistic_conflicts(
|
||
tmp_path: Path,
|
||
) -> None:
|
||
service = WorkflowConsoleService(
|
||
Settings(project_root=PROJECT_ROOT, data_root=tmp_path),
|
||
dynamic_configs=FakeDynamicConfigStore(),
|
||
)
|
||
|
||
assert service.dynamic_config_snapshot()["summary"]["total"] == 1
|
||
created = service.create_dynamic_config(_payload())
|
||
assert created["revision"] == 1
|
||
updated = service.update_dynamic_config(1, _payload(), expected_revision=1)
|
||
assert updated["revision"] == 2
|
||
with pytest.raises(ConsoleConflictError, match="其他人更新"):
|
||
service.update_dynamic_config(1, _payload(), expected_revision=9)
|
||
|
||
|
||
def test_console_assets_include_dynamic_config_crud_page() -> None:
|
||
html = (PROJECT_ROOT / "src" / "gyxx_flow" / "web" / "index.html").read_text(
|
||
encoding="utf-8"
|
||
)
|
||
script = (PROJECT_ROOT / "src" / "gyxx_flow" / "web" / "app.js").read_text(
|
||
encoding="utf-8"
|
||
)
|
||
|
||
assert 'id="dynamic-config-view"' in html
|
||
assert 'id="dynamic-config-form"' in html
|
||
assert "/api/dynamic-configs" in script
|
||
assert "#/configs/products" in script
|
||
|
||
|
||
def test_content_summary_targets_are_loaded_from_project_database(monkeypatch) -> None:
|
||
from gyxx_flow.modules.content_marketing import weekly_summary_all
|
||
|
||
monkeypatch.setattr(
|
||
weekly_summary_all,
|
||
"runtime_workflow_configs",
|
||
lambda: [
|
||
{
|
||
"style_name": "极星pro",
|
||
"weekly_note_analysis_url": (
|
||
"https://example.feishu.cn/base/base1?table=table1&view=view1"
|
||
),
|
||
}
|
||
],
|
||
)
|
||
|
||
assert weekly_summary_all.load_index_styles() == {
|
||
"极星pro": (
|
||
"base1",
|
||
"table1",
|
||
"https://example.feishu.cn/base/base1?table=table1&view=view1",
|
||
)
|
||
}
|
||
|
||
|
||
def test_content_mapping_rejects_legacy_static_cache(
|
||
monkeypatch, tmp_path: Path
|
||
) -> None:
|
||
from gyxx_flow.modules.content_marketing import feishu_mapping
|
||
|
||
cache_path = tmp_path / "mapping.json"
|
||
cache_path.write_text(
|
||
json.dumps({"platform": "xhs_pgy", "tables": [{"name": "旧款"}]}),
|
||
encoding="utf-8",
|
||
)
|
||
monkeypatch.setattr(feishu_mapping, "MAPPING_PATH", cache_path)
|
||
monkeypatch.setattr(feishu_mapping, "refresh_mapping", lambda **_kwargs: None)
|
||
|
||
with pytest.raises(RuntimeError, match="项目数据库不可用"):
|
||
feishu_mapping.load_mapping(force_refresh=False)
|
||
|
||
|
||
def test_retired_feishu_master_is_not_referenced_by_runtime_modules() -> None:
|
||
retired_token = "TtoCb1NuQaDy3NsZWTpc0GIvnph"
|
||
runtime_root = PROJECT_ROOT / "src" / "gyxx_flow"
|
||
offenders = []
|
||
for path in runtime_root.rglob("*.py"):
|
||
if path == runtime_root / "migration" / "feishu_style_config_seed.py":
|
||
continue
|
||
if retired_token in path.read_text(encoding="utf-8"):
|
||
offenders.append(path.relative_to(PROJECT_ROOT).as_posix())
|
||
|
||
assert offenders == []
|
||
|
||
|
||
def test_dynamic_config_http_crud_uses_revision_preconditions(tmp_path: Path) -> None:
|
||
server = create_console_server(
|
||
Settings(project_root=PROJECT_ROOT, data_root=tmp_path),
|
||
host="127.0.0.1",
|
||
port=0,
|
||
dynamic_configs=FakeDynamicConfigStore(),
|
||
)
|
||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||
thread.start()
|
||
base_url = f"http://127.0.0.1:{server.server_address[1]}"
|
||
|
||
def request(
|
||
method: str,
|
||
path: str,
|
||
payload: dict[str, Any] | None = None,
|
||
revision: int | None = None,
|
||
) -> tuple[int, dict[str, Any]]:
|
||
data = json.dumps(payload).encode("utf-8") if payload is not None else None
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"X-GYXX-Console": "1",
|
||
"Origin": base_url,
|
||
}
|
||
if revision is not None:
|
||
headers["If-Match"] = str(revision)
|
||
response = urlopen(
|
||
Request(base_url + path, data=data, headers=headers, method=method),
|
||
timeout=5,
|
||
)
|
||
return response.status, json.loads(response.read().decode("utf-8"))
|
||
|
||
try:
|
||
status, snapshot = request("GET", "/api/dynamic-configs")
|
||
assert status == 200
|
||
assert snapshot["summary"]["total"] == 1
|
||
|
||
status, created = request("POST", "/api/dynamic-configs", _payload())
|
||
assert status == 201
|
||
assert created["revision"] == 1
|
||
|
||
status, updated = request(
|
||
"PUT", "/api/dynamic-configs/1", _payload(), revision=1
|
||
)
|
||
assert status == 200
|
||
assert updated["revision"] == 2
|
||
|
||
status, deleted = request(
|
||
"DELETE", "/api/dynamic-configs/1", revision=1
|
||
)
|
||
assert status == 200
|
||
assert deleted == {"deleted": True, "id": 1}
|
||
finally:
|
||
server.shutdown()
|
||
server.server_close()
|
||
thread.join(timeout=5)
|