402 lines
14 KiB
Python
402 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gyxx_flow.core.config import Settings
|
|
from gyxx_flow.notification_routing import (
|
|
ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID,
|
|
ANALYZER_NOTIFICATION_APP_ID,
|
|
GYXX_NOTIFICATION_APP_PROFILE,
|
|
GYXX_NOTIFICATION_HERMES_PROFILE,
|
|
GYXX_NOTIFICATION_RECIPIENTS_JSON,
|
|
GYXX_NOTIFICATION_ROUTE_MODE,
|
|
GYXX_NOTIFICATION_ROUTE_WORKFLOW_ID,
|
|
NOTIFICATION_CAPABILITIES,
|
|
NotificationRoutingConflictError,
|
|
NotificationRoutingError,
|
|
NotificationRoutingPreconditionError,
|
|
NotificationRoutingStore,
|
|
notification_environment_for_workflow,
|
|
resolve_notification_route,
|
|
)
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
@pytest.fixture
|
|
def routing_settings(tmp_path: Path) -> Settings:
|
|
project_root = tmp_path / "project"
|
|
config_root = project_root / "config"
|
|
config_root.mkdir(parents=True)
|
|
shutil.copyfile(
|
|
PROJECT_ROOT / "config" / "notification-routing.json",
|
|
config_root / "notification-routing.json",
|
|
)
|
|
return Settings(project_root=project_root, data_root=tmp_path / "data")
|
|
|
|
|
|
def _environment(settings: Settings) -> dict[str, str]:
|
|
return {
|
|
"GYXX_PROJECT_ROOT": str(settings.project_root),
|
|
"GYXX_DATA_ROOT": str(settings.data_root),
|
|
}
|
|
|
|
|
|
def _route(
|
|
*,
|
|
enabled: bool = True,
|
|
app_profile: str = "hermes-analyzer",
|
|
person_ids: list[str] | None = None,
|
|
) -> dict[str, object]:
|
|
return {
|
|
"enabled": enabled,
|
|
"app_profile": app_profile,
|
|
"person_ids": ["he_yingwei", "wang_yunlong"]
|
|
if person_ids is None
|
|
else person_ids,
|
|
}
|
|
|
|
|
|
def test_baseline_contains_all_provided_people_and_routable_capabilities(
|
|
routing_settings: Settings,
|
|
) -> None:
|
|
payload, revision = NotificationRoutingStore(routing_settings).snapshot()
|
|
|
|
assert len(payload["people"]) == 16
|
|
assert payload["app_profile"] == "hermes-analyzer"
|
|
assert (
|
|
payload["app_profiles"]["hermes-analyzer"]["app_id"]
|
|
== ANALYZER_NOTIFICATION_APP_ID
|
|
)
|
|
assert all(person["mobile"] is None for person in payload["people"].values())
|
|
assert set(payload["capabilities"]) == set(NOTIFICATION_CAPABILITIES)
|
|
assert payload["routes"] == {}
|
|
assert len(revision) == 64
|
|
|
|
wang = payload["people"]["wang_yunlong"]
|
|
assert wang["bindings"]["collector"]["open_id"] == (
|
|
"ou_8ee224968aa26a74c7d30ba27fed5eeb"
|
|
)
|
|
assert wang["bindings"]["hermes-analyzer"]["open_id"] == (
|
|
"ou_7ad5fc8012e2f741afc5346e05ffd447"
|
|
)
|
|
provided = {
|
|
person["name"]: binding["open_id"]
|
|
for person_id, person in payload["people"].items()
|
|
if person_id != "wang_yunlong"
|
|
for profile, binding in person["bindings"].items()
|
|
if profile == "hermes-analyzer"
|
|
}
|
|
assert provided == {
|
|
"何颖威": "ou_89bcff110ccbb09a23548dc0fb3d880c",
|
|
"冯任运": "ou_c1faf3d3498201d27cd73c388acde067",
|
|
"张育基": "ou_339c1d396b397c97b08e1bf377963d40",
|
|
"李静娴": "ou_2eda5eec112109ae6d19a1f6813eadcb",
|
|
"黄坤平": "ou_b76e4cbdb24fe28cebd45ad091b60224",
|
|
"马兆基": "ou_54141d75b24d22ddc8c1c4d66e00b6e9",
|
|
"尹江涛": "ou_96fea927e6eef671f019428cc5075e69",
|
|
"余鹏辉": "ou_531782527ea7164a4abfb23dc1c37b2b",
|
|
"何嘉琪": "ou_bfcfa9585a0cc9c0a989816c12de661f",
|
|
"黄韶基": "ou_fa8d81a16527ad06352dbecc575285b8",
|
|
"谢钧宇": "ou_0d407e24fcf61bf6bb3a176bc84006e9",
|
|
"赵静": "ou_24cc944d6e43c69c59d6560ad4e2ae6e",
|
|
"李凤仪": "ou_e557867b29d756e57d3a1b656e9d3ce1",
|
|
"陈燕": "ou_63945c9e1b6d4c9ced7cce1c495d8bea",
|
|
"邓红梅": "ou_3799e39cfc5f78da1b7c54c2f677f5e3",
|
|
}
|
|
assert all(
|
|
binding["verified"] is False and binding["source"] == "provided"
|
|
for person_id, person in payload["people"].items()
|
|
if person_id != "wang_yunlong"
|
|
for profile, binding in person["bindings"].items()
|
|
if profile == "hermes-analyzer"
|
|
)
|
|
|
|
|
|
def test_business_adapter_boundary_reexports_notification_resolution() -> None:
|
|
from gyxx_flow.adapters import (
|
|
ResolvedNotificationRoute as ExportedResolvedNotificationRoute,
|
|
)
|
|
from gyxx_flow.adapters import resolve_notification_route as exported_resolver
|
|
from gyxx_flow.notification_routing import ResolvedNotificationRoute
|
|
|
|
assert ExportedResolvedNotificationRoute is ResolvedNotificationRoute
|
|
assert exported_resolver is resolve_notification_route
|
|
|
|
|
|
def test_enabled_route_supports_multiple_people_and_deduplicates_ids(
|
|
routing_settings: Settings,
|
|
) -> None:
|
|
store = NotificationRoutingStore(routing_settings)
|
|
payload, revision = store.snapshot()
|
|
payload["routes"]["content.marketing_report.daily"] = _route(
|
|
person_ids=["he_yingwei", "wang_yunlong", "he_yingwei"]
|
|
)
|
|
|
|
updated, updated_revision = store.update(
|
|
payload,
|
|
expected_revision=f'"{revision}"',
|
|
)
|
|
resolved = resolve_notification_route(
|
|
"content.marketing_report.daily",
|
|
("ou_legacy",),
|
|
_environment(routing_settings),
|
|
)
|
|
|
|
assert store.path == (
|
|
routing_settings.data_root / "state" / "notifications" / "routing.json"
|
|
)
|
|
assert store.path.is_file()
|
|
assert updated_revision != revision
|
|
assert updated["routes"]["content.marketing_report.daily"]["person_ids"] == [
|
|
"he_yingwei",
|
|
"wang_yunlong",
|
|
]
|
|
assert resolved.configured is True
|
|
assert resolved.enabled is True
|
|
assert resolved.app_profile == "hermes-analyzer"
|
|
assert resolved.open_ids == (
|
|
"ou_89bcff110ccbb09a23548dc0fb3d880c",
|
|
"ou_7ad5fc8012e2f741afc5346e05ffd447",
|
|
)
|
|
|
|
|
|
def test_disabled_route_returns_no_recipients_and_absent_route_is_legacy(
|
|
routing_settings: Settings,
|
|
) -> None:
|
|
legacy = resolve_notification_route(
|
|
"product.alert.daily",
|
|
("ou_old", "ou_old", "ou_other"),
|
|
_environment(routing_settings),
|
|
)
|
|
assert legacy.configured is False
|
|
assert legacy.enabled is True
|
|
assert legacy.open_ids == ("ou_old", "ou_other")
|
|
|
|
store = NotificationRoutingStore(routing_settings)
|
|
payload, revision = store.snapshot()
|
|
payload["routes"]["product.alert.daily"] = _route(
|
|
enabled=False,
|
|
person_ids=[],
|
|
)
|
|
store.update(payload, expected_revision=revision)
|
|
|
|
disabled = resolve_notification_route(
|
|
"product.alert.daily",
|
|
("ou_old",),
|
|
_environment(routing_settings),
|
|
)
|
|
assert disabled.configured is True
|
|
assert disabled.enabled is False
|
|
assert disabled.open_ids == ()
|
|
|
|
|
|
def test_acceptance_recipient_does_not_reenable_an_explicitly_disabled_route(
|
|
routing_settings: Settings,
|
|
) -> None:
|
|
store = NotificationRoutingStore(routing_settings)
|
|
payload, revision = store.snapshot()
|
|
payload["routes"]["product.alert.daily"] = _route(
|
|
enabled=False,
|
|
person_ids=[],
|
|
)
|
|
store.update(payload, expected_revision=revision)
|
|
environment = {
|
|
**_environment(routing_settings),
|
|
"GYXX_WORKFLOW_ACCEPTANCE": "1",
|
|
"GYXX_NOTIFICATION_RECIPIENT_OPEN_ID": (
|
|
"ou_8ee224968aa26a74c7d30ba27fed5eeb"
|
|
),
|
|
}
|
|
|
|
resolved = resolve_notification_route(
|
|
"product.alert.daily",
|
|
("ou_legacy",),
|
|
environment,
|
|
)
|
|
|
|
assert resolved.enabled is False
|
|
assert resolved.open_ids == ()
|
|
|
|
|
|
def test_app_profile_identity_is_required_and_open_ids_remain_app_scoped(
|
|
routing_settings: Settings,
|
|
) -> None:
|
|
store = NotificationRoutingStore(routing_settings)
|
|
payload, revision = store.snapshot()
|
|
invalid = deepcopy(payload)
|
|
invalid["routes"]["product.market_rank"] = _route(
|
|
app_profile="collector",
|
|
person_ids=["he_yingwei"],
|
|
)
|
|
with pytest.raises(NotificationRoutingError, match="analyzer Hermes"):
|
|
store.update(invalid, expected_revision=revision)
|
|
assert not store.path.exists()
|
|
|
|
invalid_default = deepcopy(payload)
|
|
invalid_default["app_profile"] = "collector"
|
|
with pytest.raises(NotificationRoutingError, match="analyzer Hermes"):
|
|
store.update(invalid_default, expected_revision=revision)
|
|
assert not store.path.exists()
|
|
|
|
|
|
def test_analyzer_capability_uses_analyzer_scoped_acceptance_recipient(
|
|
routing_settings: Settings,
|
|
) -> None:
|
|
resolved = resolve_notification_route(
|
|
"content.marketing_report.daily",
|
|
("ou_89bcff110ccbb09a23548dc0fb3d880c",),
|
|
{
|
|
**_environment(routing_settings),
|
|
"GYXX_WORKFLOW_ACCEPTANCE": "1",
|
|
"GYXX_NOTIFICATION_RECIPIENT_OPEN_ID": (
|
|
"ou_8ee224968aa26a74c7d30ba27fed5eeb"
|
|
),
|
|
},
|
|
)
|
|
|
|
assert resolved.open_ids == (ANALYZER_ACCEPTANCE_RECIPIENT_OPEN_ID,)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"snapshot_workflow",
|
|
[None, "product.alert.daily"],
|
|
)
|
|
def test_environment_snapshot_requires_the_exact_workflow_id(
|
|
snapshot_workflow: str | None,
|
|
) -> None:
|
|
environment = {
|
|
GYXX_NOTIFICATION_ROUTE_MODE: "enabled",
|
|
GYXX_NOTIFICATION_APP_PROFILE: "hermes-analyzer",
|
|
GYXX_NOTIFICATION_HERMES_PROFILE: "data-analyzer",
|
|
GYXX_NOTIFICATION_RECIPIENTS_JSON: (
|
|
'["ou_89bcff110ccbb09a23548dc0fb3d880c"]'
|
|
),
|
|
}
|
|
if snapshot_workflow is not None:
|
|
environment[GYXX_NOTIFICATION_ROUTE_WORKFLOW_ID] = snapshot_workflow
|
|
|
|
with pytest.raises(NotificationRoutingError, match="another workflow"):
|
|
resolve_notification_route(
|
|
"content.marketing_report.daily",
|
|
(),
|
|
environment,
|
|
)
|
|
|
|
|
|
def test_unsupported_workflow_cannot_gain_a_route(
|
|
routing_settings: Settings,
|
|
) -> None:
|
|
store = NotificationRoutingStore(routing_settings)
|
|
payload, revision = store.snapshot()
|
|
payload["routes"]["content.metrics.daily"] = _route()
|
|
|
|
with pytest.raises(NotificationRoutingError, match="does not support"):
|
|
store.update(payload, expected_revision=revision)
|
|
assert not store.path.exists()
|
|
|
|
|
|
def test_unsafe_people_fields_and_cross_person_identity_collisions_are_rejected(
|
|
routing_settings: Settings,
|
|
) -> None:
|
|
store = NotificationRoutingStore(routing_settings)
|
|
payload, revision = store.snapshot()
|
|
unsafe = deepcopy(payload)
|
|
unsafe["people"]["he_yingwei"]["name"] = "何颖威\u200b"
|
|
with pytest.raises(NotificationRoutingError, match="person name is invalid"):
|
|
store.update(unsafe, expected_revision=revision)
|
|
|
|
collision = deepcopy(payload)
|
|
collision["people"]["feng_renyun"]["bindings"]["hermes-analyzer"][
|
|
"open_id"
|
|
] = collision["people"]["he_yingwei"]["bindings"]["hermes-analyzer"][
|
|
"open_id"
|
|
]
|
|
with pytest.raises(NotificationRoutingError, match="multiple people"):
|
|
store.update(collision, expected_revision=revision)
|
|
assert not store.path.exists()
|
|
|
|
|
|
def test_corrupt_runtime_state_fails_closed_instead_of_using_defaults(
|
|
routing_settings: Settings,
|
|
) -> None:
|
|
store = NotificationRoutingStore(routing_settings)
|
|
store.path.parent.mkdir(parents=True)
|
|
store.path.write_text("{broken", encoding="utf-8")
|
|
|
|
with pytest.raises(NotificationRoutingError, match="JSON is invalid"):
|
|
resolve_notification_route(
|
|
"content.marketing_report.daily",
|
|
("ou_legacy",),
|
|
_environment(routing_settings),
|
|
)
|
|
|
|
|
|
def test_update_requires_current_revision_and_preserves_state_on_conflict(
|
|
routing_settings: Settings,
|
|
) -> None:
|
|
store = NotificationRoutingStore(routing_settings)
|
|
payload, revision = store.snapshot()
|
|
payload["routes"]["product.alert.daily"] = _route(
|
|
person_ids=["wang_yunlong"]
|
|
)
|
|
|
|
with pytest.raises(NotificationRoutingPreconditionError):
|
|
store.update(payload, expected_revision=None)
|
|
with pytest.raises(NotificationRoutingConflictError):
|
|
store.update(payload, expected_revision="0" * 64)
|
|
assert not store.path.exists()
|
|
|
|
_updated, current_revision = store.update(payload, expected_revision=revision)
|
|
before = store.path.read_bytes()
|
|
with pytest.raises(NotificationRoutingConflictError):
|
|
store.update(payload, expected_revision=revision)
|
|
assert store.path.read_bytes() == before
|
|
assert current_revision != revision
|
|
|
|
|
|
def test_environment_snapshot_is_stable_after_routing_file_changes(
|
|
routing_settings: Settings,
|
|
) -> None:
|
|
store = NotificationRoutingStore(routing_settings)
|
|
payload, revision = store.snapshot()
|
|
payload["routes"]["supply.replenishment.weekly"] = _route(
|
|
person_ids=["he_yingwei", "wang_yunlong"]
|
|
)
|
|
payload, revision = store.update(payload, expected_revision=revision)
|
|
|
|
snapshot_environment = notification_environment_for_workflow(
|
|
"supply.replenishment.weekly",
|
|
project_root=routing_settings.project_root,
|
|
data_root=routing_settings.data_root,
|
|
environment={},
|
|
)
|
|
payload["routes"]["supply.replenishment.weekly"] = _route(
|
|
person_ids=["feng_renyun"]
|
|
)
|
|
store.update(payload, expected_revision=revision)
|
|
|
|
resolved = resolve_notification_route(
|
|
"supply.replenishment.weekly",
|
|
(),
|
|
snapshot_environment,
|
|
)
|
|
assert snapshot_environment[GYXX_NOTIFICATION_ROUTE_MODE] == "enabled"
|
|
assert (
|
|
snapshot_environment[GYXX_NOTIFICATION_HERMES_PROFILE]
|
|
== "data-analyzer"
|
|
)
|
|
assert json.loads(snapshot_environment[GYXX_NOTIFICATION_RECIPIENTS_JSON]) == [
|
|
"ou_89bcff110ccbb09a23548dc0fb3d880c",
|
|
"ou_7ad5fc8012e2f741afc5346e05ffd447",
|
|
]
|
|
assert resolved.open_ids == (
|
|
"ou_89bcff110ccbb09a23548dc0fb3d880c",
|
|
"ou_7ad5fc8012e2f741afc5346e05ffd447",
|
|
)
|