awoooi/apps/api/tests/test_telegram_callback_monitoring_contract.py

from __future__ import annotations

import re
from pathlib import Path

import pytest
import yaml

from src.core.metrics import (
    TELEGRAM_CALLBACK_INGRESS_LAST_RECEIPT_TIMESTAMP,
)
from src.services import telegram_gateway as telegram_gateway_module
from src.services.controlled_alert_target_router import resolve_typed_alert_target
from src.services.telegram_gateway import TelegramGateway

ROOT = Path(__file__).resolve().parents[3]
CATALOG = ROOT / "apps" / "api" / "src" / "services" / "awooop_ansible_audit_service.py"
POST_VERIFIER = (
    ROOT / "apps" / "api" / "src" / "services" / "awooop_ansible_post_verifier.py"
)
PLAYBOOK = (
    ROOT
    / "infra"
    / "ansible"
    / "playbooks"
    / "110-telegram-callback-ingress-sensor.yml"
)
RULE_SOURCES = (
    ROOT / "ops" / "monitoring" / "alerts-unified.yml",
    ROOT / "ops" / "monitoring" / "alerts.yml",
    ROOT / "k8s" / "monitoring" / "alert-chain-monitor.yaml",
)


class _ConfiguredGateway(TelegramGateway):
    @property
    def bot_token(self) -> str:
        return "configured"

    @property
    def chat_id(self) -> str:
        return "configured"


class _Redis:
    def __init__(self) -> None:
        self.values: dict[str, str] = {}

    async def setex(self, key: str, _ttl: int, value: str) -> None:
        self.values[key] = value

    async def get(self, key: str) -> str | None:
        return self.values.get(key)


def _rule(path: Path) -> dict:
    payload = yaml.safe_load(path.read_text(encoding="utf-8"))
    groups = payload["spec"]["groups"] if "spec" in payload else payload["groups"]
    return next(
        rule
        for group in groups
        for rule in group.get("rules", [])
        if rule.get("alert") == "TelegramCallbackIngressUnverified"
    )


def test_callback_ingress_sensor_is_identical_and_exact_typed_in_all_sources() -> None:
    rules = [_rule(path) for path in RULE_SOURCES]
    expressions = {re.sub(r"\s+", "", str(rule["expr"])) for rule in rules}

    assert len(expressions) == 1
    expression = expressions.pop()
    assert "awoooi_telegram_callback_ingress_last_receipt_timestamp_seconds" in expression
    assert ">300" in expression
    assert "absent(" in expression

    for rule in rules:
        assert rule["for"] == "2m"
        assert rule["labels"] == {
            "severity": "warning",
            "layer": "docker-188",
            "component": "openclaw",
            "host": "188",
            "team": "platform",
            "auto_repair": "true",
            "alert_category": "ai_agent",
            "canonical_asset_id": "service:openclaw:host188",
            "domain_router": "docker_container",
        }

        route = resolve_typed_alert_target(
            alertname="TelegramCallbackIngressUnverified",
            target_resource=rule["labels"]["component"],
            namespace="default",
            labels=rule["labels"],
            alert_category=rule["labels"]["alert_category"],
        )
        assert route["canonical_asset_id"] == "service:openclaw:host188"
        assert route["allowed_catalog_ids"] == [
            "ansible:188-openclaw-callback-forwarder"
        ]
        assert route["cross_domain_fallback_allowed"] is False


def test_sensor_has_exact_host110_catalog_rollback_and_independent_verifier() -> None:
    catalog = CATALOG.read_text(encoding="utf-8")
    verifier = POST_VERIFIER.read_text(encoding="utf-8")
    playbook = PLAYBOOK.read_text(encoding="utf-8")

    assert '"catalog_id": "ansible:110-telegram-callback-ingress-sensor"' in catalog
    assert '"inventory_hosts": ["host_110"]' in catalog
    assert '"auto_apply_enabled": False' in catalog
    assert '"canonical_asset_id": "service:prometheus"' in catalog
    assert '"ansible:110-telegram-callback-ingress-sensor": (' in verifier
    assert "host_110_telegram_callback_ingress_rule_healthy" in verifier
    assert "host_110_telegram_callback_ingress_metric_visible" in verifier
    assert "host_110_telegram_callback_ingress_rule_hash_converged" in verifier
    assert "promtool" in playbook
    assert "callback-sensor.bak" in playbook
    assert "rescue:" in playbook
    assert "Require exact rollback proof after any started apply" in playbook
    assert "TelegramCallbackIngressUnverified" in playbook

    play = yaml.safe_load(playbook)[0]
    assert play["hosts"] == "host_110"
    assert "--force-recreate" in playbook
    assert "prometheus_runtime_rules" in playbook
    assert "sensor_rollback_hashes" in playbook


@pytest.mark.asyncio
async def test_durable_callback_receipt_updates_prometheus_timestamp(
    monkeypatch,
) -> None:
    redis = _Redis()
    monkeypatch.setattr(telegram_gateway_module, "get_redis", lambda: redis)
    TELEGRAM_CALLBACK_INGRESS_LAST_RECEIPT_TIMESTAMP.set(0)

    gateway = _ConfiguredGateway()
    await gateway.record_callback_ingress_receipt(
        source="openclaw_188_hmac_forwarder",
        update_id=99,
    )

    assert TELEGRAM_CALLBACK_INGRESS_LAST_RECEIPT_TIMESTAMP._value.get() > 0


@pytest.mark.asyncio
async def test_separate_scraped_replica_refreshes_shared_receipt_metric(
    monkeypatch,
) -> None:
    redis = _Redis()
    monkeypatch.setattr(telegram_gateway_module, "get_redis", lambda: redis)
    writer = _ConfiguredGateway()
    scraped_replica = _ConfiguredGateway()

    await writer.record_callback_ingress_receipt(
        source="openclaw_188_hmac_forwarder",
        update_id=100,
    )
    TELEGRAM_CALLBACK_INGRESS_LAST_RECEIPT_TIMESTAMP.set(0)

    assert await scraped_replica.refresh_callback_ingress_receipt() is True
    assert TELEGRAM_CALLBACK_INGRESS_LAST_RECEIPT_TIMESTAMP._value.get() > 0