124 lines
3.5 KiB
Python
124 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
from types import ModuleType, SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from gyxx_flow.adapters import ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[3]
|
|
COMMAND_PATH = (
|
|
PROJECT_ROOT
|
|
/ "src"
|
|
/ "gyxx_flow"
|
|
/ "modules"
|
|
/ "product_commerce"
|
|
/ "commands"
|
|
/ "run_alerts.py"
|
|
)
|
|
|
|
|
|
def _load_command(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
observed: dict[str, object],
|
|
) -> ModuleType:
|
|
legacy = ModuleType("run_alerts_with_retry")
|
|
|
|
def fake_main() -> int:
|
|
observed["argv"] = tuple(sys.argv)
|
|
return 0
|
|
|
|
legacy.main = fake_main # type: ignore[attr-defined]
|
|
monkeypatch.setitem(sys.modules, "run_alerts_with_retry", legacy)
|
|
spec = importlib.util.spec_from_file_location("test_run_alerts_command", COMMAND_PATH)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_alert_command_uses_canonical_acceptance_recipient(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
observed: dict[str, object] = {}
|
|
monkeypatch.setenv("GYXX_BUSINESS_DATE", "2026-08-01")
|
|
monkeypatch.setenv("GYXX_WORKFLOW_ACCEPTANCE", "1")
|
|
monkeypatch.delenv("GYXX_NOTIFICATION_RECIPIENT_OPEN_ID", raising=False)
|
|
monkeypatch.delenv("GYXX_ALERT_RECIPIENT_OPEN_ID", raising=False)
|
|
module = _load_command(monkeypatch, observed)
|
|
|
|
assert module.main() == 0
|
|
assert observed["argv"] == (
|
|
sys.argv[0],
|
|
"--end-date",
|
|
"2026-08-01",
|
|
"--openid",
|
|
ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID,
|
|
)
|
|
|
|
|
|
def test_alert_command_accepts_canonical_production_recipient(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
observed: dict[str, object] = {}
|
|
monkeypatch.setenv("GYXX_BUSINESS_DATE", "2026-08-01")
|
|
monkeypatch.delenv("GYXX_WORKFLOW_ACCEPTANCE", raising=False)
|
|
monkeypatch.setenv("GYXX_NOTIFICATION_RECIPIENT_OPEN_ID", "ou_configured")
|
|
monkeypatch.setenv("GYXX_ALERT_RECIPIENT_OPEN_ID", "ou_legacy")
|
|
module = _load_command(monkeypatch, observed)
|
|
|
|
assert module.main() == 0
|
|
assert observed["argv"][-1] == "ou_configured"
|
|
|
|
|
|
def test_alert_command_passes_all_dynamic_recipients(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
observed: dict[str, object] = {}
|
|
monkeypatch.setenv("GYXX_BUSINESS_DATE", "2026-08-01")
|
|
monkeypatch.delenv("GYXX_WORKFLOW_ACCEPTANCE", raising=False)
|
|
module = _load_command(monkeypatch, observed)
|
|
monkeypatch.setattr(
|
|
module,
|
|
"resolve_notification_route",
|
|
lambda *_args, **_kwargs: SimpleNamespace(
|
|
enabled=True,
|
|
open_ids=("ou_first", "ou_second"),
|
|
),
|
|
)
|
|
|
|
assert module.main() == 0
|
|
assert observed["argv"] == (
|
|
sys.argv[0],
|
|
"--end-date",
|
|
"2026-08-01",
|
|
"--openid",
|
|
"ou_first",
|
|
"--openid",
|
|
"ou_second",
|
|
)
|
|
|
|
|
|
def test_alert_command_uses_safe_no_notify_when_route_is_disabled(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
observed: dict[str, object] = {}
|
|
monkeypatch.setenv("GYXX_BUSINESS_DATE", "2026-08-01")
|
|
module = _load_command(monkeypatch, observed)
|
|
monkeypatch.setattr(
|
|
module,
|
|
"resolve_notification_route",
|
|
lambda *_args, **_kwargs: SimpleNamespace(enabled=False, open_ids=()),
|
|
)
|
|
|
|
assert module.main() == 0
|
|
assert observed["argv"] == (
|
|
sys.argv[0],
|
|
"--end-date",
|
|
"2026-08-01",
|
|
"--no-notify",
|
|
)
|