feat(notifications): expose canonical Telegram routing
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / tests (push) Has been cancelled

This commit is contained in:
ogt
2026-07-15 08:29:05 +08:00
parent 2d6416c322
commit 088d8a509d
8 changed files with 522 additions and 25 deletions

View File

@@ -0,0 +1,58 @@
from __future__ import annotations
from fastapi import FastAPI
from fastapi.testclient import TestClient
from src.api.v1 import agents
from src.api.v1.agents import router
def _client() -> TestClient:
app = FastAPI()
app.include_router(router, prefix="/api/v1")
return TestClient(app)
def test_canonical_routing_runtime_readback_endpoint(monkeypatch) -> None:
payload = {
"schema_version": "telegram_canonical_routing_runtime_readback_v1",
"status": "runtime_policy_enforced_product_delivery_receipts_partial",
"operation_boundaries": {
"telegram_send_performed": False,
"secret_value_read": False,
},
}
monkeypatch.setattr(
agents,
"build_canonical_telegram_routing_runtime_readback",
lambda: payload,
)
response = _client().get(
"/api/v1/agents/telegram-canonical-routing-runtime-readback"
)
assert response.status_code == 200
assert response.json() == payload
def test_canonical_routing_runtime_readback_rejects_invalid_source(
monkeypatch,
) -> None:
def invalid_source() -> dict:
raise ValueError("invalid registry")
monkeypatch.setattr(
agents,
"build_canonical_telegram_routing_runtime_readback",
invalid_source,
)
response = _client().get(
"/api/v1/agents/telegram-canonical-routing-runtime-readback"
)
assert response.status_code == 500
assert response.json() == {
"detail": "Telegram canonical routing runtime readback 無效"
}

View File

@@ -9,6 +9,7 @@ import pytest
import yaml
from src.services.notification_matrix import (
build_canonical_telegram_routing_runtime_readback,
get_trusted_alert_canonical_route_context,
load_canonical_telegram_routing_registry,
resolve_canonical_telegram_route,
@@ -30,6 +31,42 @@ EXPECTED_PRODUCTS = {
REPO_ROOT = Path(__file__).resolve().parents[3]
def test_runtime_readback_separates_historical_source_state_from_live_gate() -> None:
readback = build_canonical_telegram_routing_runtime_readback()
assert readback["status"] == (
"runtime_policy_enforced_product_delivery_receipts_partial"
)
assert readback["source_status"] == "source_policy_ready_runtime_not_applied"
assert readback["source_generated_at"] == "2026-07-14T15:50:04+08:00"
assert readback["runtime_generated_at"].endswith("+00:00")
assert readback["runtime_enforcement"] == {
"canonical_resolver_active": True,
"gateway_pre_send_gate_active": True,
"trusted_ingress_context_required": True,
"unknown_route_decision": "block",
"unknown_route_reason": "unknown_product_signal_or_severity",
"raw_numeric_destination_count": 0,
"destination_identity_exposed": False,
"all_product_delivery_receipts_ready": False,
}
assert readback["rollups"]["runtime_allowed_route_count"] == 4
assert readback["rollups"]["runtime_blocked_route_count"] == 16
assert len(readback["products"]) == 11
assert len(readback["routes"]) == 20
assert readback["operation_boundaries"] == {
"read_only": True,
"telegram_send_performed": False,
"bot_api_call_performed": False,
"runtime_route_changed": False,
"secret_value_read": False,
"raw_identifier_included": False,
}
serialized = json.dumps(readback, ensure_ascii=False)
assert '"chat_id"' not in serialized
assert '"bot_token"' not in serialized
def test_namespaced_product_ingress_route_is_explicit_and_policy_blocked() -> None:
context = get_trusted_alert_canonical_route_context(
{