579 lines
17 KiB
Python
579 lines
17 KiB
Python
from __future__ import annotations
|
||
|
||
import subprocess
|
||
import sys
|
||
from datetime import date
|
||
from pathlib import Path
|
||
from types import ModuleType
|
||
|
||
import check_nine_day_decline as decline
|
||
import pytest
|
||
import run_alerts_with_retry as retry
|
||
|
||
PAIR = ("tm", "测试款")
|
||
|
||
|
||
def _segment(sales: int, *, days: int = 3) -> dict:
|
||
return {
|
||
"sales": sales,
|
||
"present_days": days,
|
||
"erp_style_codes": ["STYLE-1"],
|
||
}
|
||
|
||
|
||
def _event() -> dict:
|
||
return {
|
||
"style_name": "测试款",
|
||
"platform": "tm",
|
||
"erp_style_codes": ["STYLE-1"],
|
||
"window_start": "2026-07-24",
|
||
"window_end": "2026-08-01",
|
||
"seg1_sales": 300,
|
||
"seg2_sales": 210,
|
||
"seg3_sales": 90,
|
||
"drop_pct_1to2": 30.0,
|
||
"drop_pct_2to3": 57.1,
|
||
"drop_pct_1to3": 70.0,
|
||
}
|
||
|
||
|
||
def _prepare_main(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||
monkeypatch.setattr(
|
||
sys,
|
||
"argv",
|
||
[
|
||
"check_nine_day_decline.py",
|
||
"--end-date",
|
||
"2026-08-01",
|
||
"--openid",
|
||
"ou_owner",
|
||
"--data-root",
|
||
str(tmp_path),
|
||
],
|
||
)
|
||
monkeypatch.setattr(decline, "managed_data_path", lambda value: Path(value))
|
||
monkeypatch.setattr(decline, "append_log", lambda line: None)
|
||
monkeypatch.setattr(
|
||
decline,
|
||
"_persist_detected_events",
|
||
lambda events, *_args: len(events),
|
||
)
|
||
monkeypatch.delenv("GYXX_WORKFLOW_ACCEPTANCE", raising=False)
|
||
|
||
|
||
def test_detect_decline_uses_three_inclusive_postgres_sum_windows(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
observed: list[list[tuple[date, date]]] = []
|
||
|
||
def fake_load(windows):
|
||
observed.append(windows)
|
||
return [
|
||
{PAIR: _segment(300)},
|
||
{PAIR: _segment(210)},
|
||
{PAIR: _segment(90)},
|
||
]
|
||
|
||
monkeypatch.setattr(decline, "_load_segment_snapshots", fake_load)
|
||
|
||
events = decline.detect_decline("2026-08-01", 9)
|
||
|
||
assert observed == [
|
||
[
|
||
(date(2026, 7, 24), date(2026, 7, 26)),
|
||
(date(2026, 7, 27), date(2026, 7, 29)),
|
||
(date(2026, 7, 30), date(2026, 8, 1)),
|
||
]
|
||
]
|
||
assert len(events) == 1
|
||
assert events[0]["window_start"] == "2026-07-24"
|
||
assert events[0]["window_end"] == "2026-08-01"
|
||
assert [events[0][f"seg{index}_sales"] for index in range(1, 4)] == [
|
||
300,
|
||
210,
|
||
90,
|
||
]
|
||
assert events[0]["segment_windows"] == [
|
||
{"start": "2026-07-24", "end": "2026-07-26", "days": 3},
|
||
{"start": "2026-07-27", "end": "2026-07-29", "days": 3},
|
||
{"start": "2026-07-30", "end": "2026-08-01", "days": 3},
|
||
]
|
||
|
||
|
||
def test_detect_decline_rejects_incomplete_three_day_segment(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
monkeypatch.setattr(
|
||
decline,
|
||
"_load_segment_snapshots",
|
||
lambda windows: [
|
||
{PAIR: _segment(300)},
|
||
{PAIR: _segment(210, days=2)},
|
||
{PAIR: _segment(90)},
|
||
],
|
||
)
|
||
|
||
assert decline.detect_decline("2026-08-01", 9) == []
|
||
|
||
|
||
def test_segment_loader_uses_postgres_range_sum_on_one_connection(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
fake_db = ModuleType("db")
|
||
connection = object()
|
||
observed: list[tuple[object, str, str]] = []
|
||
|
||
class ConnectionContext:
|
||
def __enter__(self):
|
||
return connection
|
||
|
||
def __exit__(self, *args):
|
||
return False
|
||
|
||
def fake_fetch(conn, start, end):
|
||
observed.append((conn, start, end))
|
||
return {}
|
||
|
||
fake_db.get_conn = ConnectionContext # type: ignore[attr-defined]
|
||
fake_db.fetch_daily_for_range = fake_fetch # type: ignore[attr-defined]
|
||
monkeypatch.setitem(sys.modules, "db", fake_db)
|
||
windows = decline._segment_windows("2026-08-01", 9)
|
||
|
||
assert decline._load_segment_snapshots(windows) == [{}, {}, {}]
|
||
assert observed == [
|
||
(connection, "2026-07-24", "2026-07-26"),
|
||
(connection, "2026-07-27", "2026-07-29"),
|
||
(connection, "2026-07-30", "2026-08-01"),
|
||
]
|
||
|
||
|
||
def test_build_decline_notification_message_is_deterministic() -> None:
|
||
message = decline.build_decline_notification_message(
|
||
[_event()],
|
||
end_date="2026-08-01",
|
||
window_days=9,
|
||
)
|
||
|
||
assert "销量连续下滑告警（截至 2026-08-01，3 段 × 3 天）" in message
|
||
assert "[tm] 测试款" in message
|
||
assert "300 → 210 → 90" in message
|
||
assert "累计 -70.0%" in message
|
||
assert "ERP" in message
|
||
|
||
|
||
def test_alert_delivery_uuid_is_stable_and_recipient_scoped() -> None:
|
||
first = decline._notification_delivery_uuid(
|
||
[_event()],
|
||
end_date="2026-08-01",
|
||
window_days=9,
|
||
openid="ou_first",
|
||
)
|
||
same = decline._notification_delivery_uuid(
|
||
[_event()],
|
||
end_date="2026-08-01",
|
||
window_days=9,
|
||
openid="ou_first",
|
||
)
|
||
second = decline._notification_delivery_uuid(
|
||
[_event()],
|
||
end_date="2026-08-01",
|
||
window_days=9,
|
||
openid="ou_second",
|
||
)
|
||
|
||
assert first == same
|
||
assert first != second
|
||
|
||
|
||
def test_decline_notification_uses_shared_lark_user_sender(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
captured: dict[str, object] = {}
|
||
monkeypatch.delenv("GYXX_WORKFLOW_ACCEPTANCE", raising=False)
|
||
|
||
def fake_send(**kwargs):
|
||
captured.update(kwargs)
|
||
return {"message_id": "om_real123"}
|
||
|
||
monkeypatch.setattr(decline, "send_lark_bot_message", fake_send)
|
||
|
||
response = decline.send_decline_notification(
|
||
[_event()],
|
||
"2026-08-01",
|
||
9,
|
||
openid="ou_owner",
|
||
)
|
||
|
||
assert response["message_id"] == "om_real123"
|
||
assert captured["user_id"] == "ou_owner"
|
||
assert captured["profile"] == "hermes-analyzer"
|
||
assert captured["idempotency_key"] == decline._notification_delivery_uuid(
|
||
[_event()],
|
||
end_date="2026-08-01",
|
||
window_days=9,
|
||
openid="ou_owner",
|
||
)
|
||
assert "测试款" in str(captured["text"])
|
||
|
||
|
||
def test_main_fails_when_postgres_detection_read_fails(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
_prepare_main(monkeypatch, tmp_path)
|
||
monkeypatch.setattr(
|
||
decline,
|
||
"detect_decline",
|
||
lambda *args: (_ for _ in ()).throw(RuntimeError("postgres unavailable")),
|
||
)
|
||
|
||
assert decline.main() == 1
|
||
|
||
|
||
def test_main_zero_events_has_explicit_postcondition(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
capsys: pytest.CaptureFixture[str],
|
||
) -> None:
|
||
_prepare_main(monkeypatch, tmp_path)
|
||
monkeypatch.setattr(decline, "detect_decline", lambda *args: [])
|
||
|
||
assert decline.main() == 0
|
||
output = capsys.readouterr().out
|
||
assert "detection_source=postgresql events=0" in output
|
||
assert "notification=not_required persistence=not_required" in output
|
||
|
||
|
||
def test_main_dedup_query_failure_is_fail_closed(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
_prepare_main(monkeypatch, tmp_path)
|
||
monkeypatch.setattr(decline, "detect_decline", lambda *args: [_event()])
|
||
monkeypatch.setattr(
|
||
decline,
|
||
"_already_notified",
|
||
lambda *args: (_ for _ in ()).throw(RuntimeError("db timeout")),
|
||
)
|
||
notify = pytest.fail
|
||
monkeypatch.setattr(decline, "send_decline_notification", notify)
|
||
|
||
assert decline.main() == 5
|
||
|
||
|
||
def test_main_does_not_accept_lark_response_without_real_message_id(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
_prepare_main(monkeypatch, tmp_path)
|
||
monkeypatch.setattr(decline, "detect_decline", lambda *args: [_event()])
|
||
monkeypatch.setattr(decline, "_already_notified", lambda *args: False)
|
||
monkeypatch.setattr(
|
||
decline,
|
||
"send_decline_notification",
|
||
lambda *args, **kwargs: {"ok": True},
|
||
)
|
||
persist = pytest.fail
|
||
monkeypatch.setattr(decline, "_persist_events", persist)
|
||
|
||
assert decline.main() == 6
|
||
|
||
|
||
def test_main_returns_nonzero_when_receipt_persistence_fails(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
_prepare_main(monkeypatch, tmp_path)
|
||
monkeypatch.setattr(decline, "detect_decline", lambda *args: [_event()])
|
||
monkeypatch.setattr(decline, "_already_notified", lambda *args: False)
|
||
monkeypatch.setattr(
|
||
decline,
|
||
"send_decline_notification",
|
||
lambda *args, **kwargs: {"message_id": "om_real123"},
|
||
)
|
||
monkeypatch.setattr(
|
||
decline,
|
||
"_persist_events",
|
||
lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("write failed")),
|
||
)
|
||
|
||
assert decline.main() == 4
|
||
|
||
|
||
def test_main_success_requires_and_persists_feishu_receipt(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
capsys: pytest.CaptureFixture[str],
|
||
) -> None:
|
||
_prepare_main(monkeypatch, tmp_path)
|
||
observed: dict[str, object] = {}
|
||
monkeypatch.setattr(decline, "detect_decline", lambda *args: [_event()])
|
||
monkeypatch.setattr(decline, "_already_notified", lambda *args: False)
|
||
monkeypatch.setattr(
|
||
decline,
|
||
"send_decline_notification",
|
||
lambda *args, **kwargs: {"message_id": "om_real123"},
|
||
)
|
||
|
||
def fake_persist(*args, **kwargs):
|
||
observed.update(kwargs)
|
||
return 1
|
||
|
||
monkeypatch.setattr(decline, "_persist_events", fake_persist)
|
||
|
||
assert decline.main() == 0
|
||
assert observed["message_id"] == "om_real123"
|
||
assert observed["openid"] == "ou_owner"
|
||
output = capsys.readouterr().out
|
||
assert "notification=sent feishu_message_id=om_real123" in output
|
||
assert "persistence=committed history_rows=1" in output
|
||
|
||
|
||
def test_main_no_notify_still_detects_hits_and_succeeds(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
capsys: pytest.CaptureFixture[str],
|
||
) -> None:
|
||
_prepare_main(monkeypatch, tmp_path)
|
||
monkeypatch.setattr(
|
||
sys,
|
||
"argv",
|
||
[
|
||
"check_nine_day_decline.py",
|
||
"--end-date",
|
||
"2026-08-01",
|
||
"--no-notify",
|
||
"--data-root",
|
||
str(tmp_path),
|
||
],
|
||
)
|
||
monkeypatch.setattr(decline, "detect_decline", lambda *args: [_event()])
|
||
persisted: list[tuple[list[dict], str, int]] = []
|
||
monkeypatch.setattr(
|
||
decline,
|
||
"_persist_detected_events",
|
||
lambda events, end_date, window_days: (
|
||
persisted.append((events, end_date, window_days)) or len(events)
|
||
),
|
||
)
|
||
monkeypatch.setattr(
|
||
decline,
|
||
"_already_notified",
|
||
lambda *args: pytest.fail("disabled notification must not query receipts"),
|
||
)
|
||
monkeypatch.setattr(
|
||
decline,
|
||
"send_decline_notification",
|
||
lambda *args, **kwargs: pytest.fail("disabled notification must not send"),
|
||
)
|
||
|
||
assert decline.main() == 0
|
||
assert persisted == [([_event()], "2026-08-01", 9)]
|
||
output = capsys.readouterr().out
|
||
assert "events=1 notification=disabled persistence=committed" in output
|
||
assert "detection_rows=1" in output
|
||
|
||
|
||
def test_main_blocks_notification_when_detection_output_cannot_be_persisted(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
_prepare_main(monkeypatch, tmp_path)
|
||
monkeypatch.setattr(decline, "detect_decline", lambda *args: [_event()])
|
||
monkeypatch.setattr(
|
||
decline,
|
||
"_persist_detected_events",
|
||
lambda *_args: (_ for _ in ()).throw(RuntimeError("write failed")),
|
||
)
|
||
monkeypatch.setattr(
|
||
decline,
|
||
"send_decline_notification",
|
||
lambda *args, **kwargs: pytest.fail("must not send without durable output"),
|
||
)
|
||
|
||
assert decline.main() == 5
|
||
|
||
|
||
def test_main_sends_and_persists_each_recipient_independently(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
_prepare_main(monkeypatch, tmp_path)
|
||
monkeypatch.setattr(
|
||
sys,
|
||
"argv",
|
||
[
|
||
"check_nine_day_decline.py",
|
||
"--end-date",
|
||
"2026-08-01",
|
||
"--openid",
|
||
"ou_first",
|
||
"--openid",
|
||
"ou_second",
|
||
"--data-root",
|
||
str(tmp_path),
|
||
],
|
||
)
|
||
monkeypatch.setattr(decline, "detect_decline", lambda *args: [_event()])
|
||
dedup_queries: list[str] = []
|
||
persisted: list[tuple[str, str]] = []
|
||
|
||
def fake_already_notified(_style, _date, recipient):
|
||
dedup_queries.append(recipient)
|
||
return False
|
||
|
||
def fake_notify(*_args, openid, **_kwargs):
|
||
return {"message_id": f"om_{openid.removeprefix('ou_')}"}
|
||
|
||
def fake_persist(*_args, openid, message_id, **_kwargs):
|
||
persisted.append((openid, message_id))
|
||
return 1
|
||
|
||
monkeypatch.setattr(decline, "_already_notified", fake_already_notified)
|
||
monkeypatch.setattr(decline, "send_decline_notification", fake_notify)
|
||
monkeypatch.setattr(decline, "_persist_events", fake_persist)
|
||
|
||
assert decline.main() == 0
|
||
assert dedup_queries == ["ou_first", "ou_second"]
|
||
assert persisted == [
|
||
("ou_first", "om_first"),
|
||
("ou_second", "om_second"),
|
||
]
|
||
|
||
|
||
def test_alert_schema_and_upsert_are_recipient_scoped() -> None:
|
||
repository_root = Path(__file__).resolve().parents[3]
|
||
schema = (
|
||
repository_root
|
||
/ "src"
|
||
/ "gyxx_flow"
|
||
/ "modules"
|
||
/ "product_commerce"
|
||
/ "db"
|
||
/ "schema.sql"
|
||
).read_text(encoding="utf-8")
|
||
db_module = (
|
||
repository_root
|
||
/ "src"
|
||
/ "gyxx_flow"
|
||
/ "modules"
|
||
/ "product_commerce"
|
||
/ "db"
|
||
/ "__init__.py"
|
||
).read_text(encoding="utf-8")
|
||
|
||
assert "UNIQUE (style_name, trigger_date, recipient)" in schema
|
||
assert "DROP CONSTRAINT IF EXISTS fact_alert_history_style_name_trigger_date_key" in schema
|
||
assert "ON CONFLICT (style_name, trigger_date, recipient)" in db_module
|
||
assert "def ensure_alert_history_recipient_scope" in db_module
|
||
assert "fact_alert_history_style_name_trigger_date_key" in db_module
|
||
|
||
|
||
def test_detected_output_migrates_recipient_scope_before_upsert(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
fake_db = ModuleType("db")
|
||
connection = object()
|
||
calls: list[tuple[str, object]] = []
|
||
|
||
class ConnectionContext:
|
||
def __enter__(self):
|
||
return connection
|
||
|
||
def __exit__(self, *args):
|
||
return False
|
||
|
||
fake_db.get_conn = ConnectionContext # type: ignore[attr-defined]
|
||
fake_db.ensure_alert_history_recipient_scope = ( # type: ignore[attr-defined]
|
||
lambda conn: calls.append(("migrate", conn))
|
||
)
|
||
fake_db.upsert_decline_event = ( # type: ignore[attr-defined]
|
||
lambda conn, event: calls.append(("upsert", (conn, event)))
|
||
)
|
||
monkeypatch.setitem(sys.modules, "db", fake_db)
|
||
|
||
assert decline._persist_detected_events([_event()], "2026-08-01", 9) == 1
|
||
assert calls[0] == ("migrate", connection)
|
||
assert calls[1][0] == "upsert"
|
||
|
||
|
||
class _FailedCollector:
|
||
def __init__(self, *args):
|
||
self.pages: list[tuple] = []
|
||
|
||
def add_page(self, *args, **kwargs):
|
||
self.pages.append((args, kwargs))
|
||
|
||
def write(self):
|
||
return None
|
||
|
||
def summary_line(self):
|
||
return "failure"
|
||
|
||
|
||
def test_retry_wrapper_does_not_repeat_after_possible_success(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
calls: list[list[str]] = []
|
||
monkeypatch.setattr(
|
||
sys,
|
||
"argv",
|
||
[
|
||
"run_alerts_with_retry.py",
|
||
"--end-date",
|
||
"2026-08-01",
|
||
"--openid",
|
||
"ou_owner",
|
||
],
|
||
)
|
||
monkeypatch.setattr(retry, "FailedCollector", _FailedCollector)
|
||
monkeypatch.setattr(retry, "consolidate_failed_logs", lambda value: None)
|
||
monkeypatch.setattr(retry, "build_child_environment", lambda *args: {})
|
||
monkeypatch.setattr(
|
||
retry.time,
|
||
"sleep",
|
||
lambda seconds: pytest.fail("non-retryable result must not sleep"),
|
||
)
|
||
|
||
def fake_run(command, **kwargs):
|
||
calls.append(command)
|
||
return subprocess.CompletedProcess(command, 4)
|
||
|
||
monkeypatch.setattr(retry.subprocess, "run", fake_run)
|
||
|
||
assert retry.main() == 4
|
||
assert len(calls) == 1
|
||
|
||
|
||
def test_retry_wrapper_can_retry_fail_closed_dedup_read(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
return_codes = iter((5, 0))
|
||
sleeps: list[int] = []
|
||
monkeypatch.setattr(
|
||
sys,
|
||
"argv",
|
||
[
|
||
"run_alerts_with_retry.py",
|
||
"--end-date",
|
||
"2026-08-01",
|
||
"--openid",
|
||
"ou_owner",
|
||
],
|
||
)
|
||
monkeypatch.setattr(retry, "FailedCollector", _FailedCollector)
|
||
monkeypatch.setattr(retry, "consolidate_failed_logs", lambda value: None)
|
||
monkeypatch.setattr(retry, "build_child_environment", lambda *args: {})
|
||
monkeypatch.setattr(retry.time, "sleep", sleeps.append)
|
||
monkeypatch.setattr(
|
||
retry.subprocess,
|
||
"run",
|
||
lambda command, **kwargs: subprocess.CompletedProcess(
|
||
command,
|
||
next(return_codes),
|
||
),
|
||
)
|
||
|
||
assert retry.main() == 0
|
||
assert sleeps == [retry.RUN_SLEEP_SECONDS]
|