awoooi/apps/api/tests/test_signoz_canonical_health_route.py

from __future__ import annotations

import subprocess
import sys
from datetime import UTC, datetime
from pathlib import Path
from types import SimpleNamespace

import pytest
import yaml

from src.api.v1 import health as health_module
from src.api.v1 import monitoring as monitoring_module
from src.core import deep_linking as deep_linking_module
from src.core.config import Settings
from src.services import signoz_client as signoz_client_module
from src.services.host_aggregator import HOST_CONFIGS

REPO_ROOT = Path(__file__).resolve().parents[3]
ALERT_CHAIN_SMOKE = REPO_ROOT / "scripts/alert_chain_smoke_test.py"
PROMETHEUS_ROUTE_DEPLOY = (
    REPO_ROOT / "scripts/ops/deploy-signoz-prometheus-route.sh"
)
LEGACY_PROMETHEUS_DEPLOY = REPO_ROOT / "k8s/monitoring/deploy-prometheus-config.sh"
INTERNAL_URL = "http://192.168.0.110:8080"
PUBLIC_URL = "https://signoz.wooo.work"
LEGACY_URL = "192.168.0.188:3301"
INTERNAL_HEALTH_URL = f"{INTERNAL_URL}/api/v1/health"


def _configmap(relative_path: str) -> dict:
    return yaml.safe_load((REPO_ROOT / relative_path).read_text(encoding="utf-8"))


def test_settings_split_internal_health_from_public_deep_links() -> None:
    fields = Settings.model_fields

    assert fields["SIGNOZ_INTERNAL_URL"].default == INTERNAL_URL
    assert fields["SIGNOZ_PUBLIC_URL"].default == PUBLIC_URL
    assert "SIGNOZ_URL" not in fields
    assert fields["OTEL_EXPORTER_OTLP_ENDPOINT"].default == "192.168.0.188:24317"


@pytest.mark.asyncio
async def test_health_check_uses_exact_canonical_internal_path(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    calls: list[tuple[str, str, str]] = []
    expected = SimpleNamespace(status="up")

    async def fake_http_health_check(name: str, base_url: str, path: str):
        calls.append((name, base_url, path))
        return expected

    monkeypatch.setattr(health_module.settings, "SIGNOZ_INTERNAL_URL", INTERNAL_URL)
    monkeypatch.setattr(
        health_module,
        "_http_health_check",
        fake_http_health_check,
    )

    assert await health_module.check_signoz() is expected
    assert calls == [("signoz", INTERNAL_URL, "/api/v1/health")]


@pytest.mark.asyncio
async def test_monitoring_probe_uses_internal_route_but_returns_replaceable_url(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    requested_urls: list[str] = []

    class FakeClient:
        async def get(self, url: str, *, timeout: float):
            requested_urls.append(url)
            assert timeout == monitoring_module.TIMEOUT
            return SimpleNamespace(status_code=200)

    monkeypatch.setattr(monitoring_module.settings, "SIGNOZ_INTERNAL_URL", INTERNAL_URL)

    probe = await monitoring_module._probe_signoz(FakeClient())

    assert requested_urls == [f"{INTERNAL_URL}/api/v1/health"]
    assert probe["url"] == INTERNAL_URL
    assert monitoring_module.public_monitoring_tool_payload(probe)["url"] == PUBLIC_URL


def test_user_facing_deep_links_use_public_route(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(deep_linking_module.settings, "SIGNOZ_PUBLIC_URL", PUBLIC_URL)
    monkeypatch.setattr(signoz_client_module.settings, "SIGNOZ_PUBLIC_URL", PUBLIC_URL)
    monkeypatch.setattr(
        signoz_client_module.settings, "SIGNOZ_INTERNAL_URL", INTERNAL_URL
    )

    trace_id = "0af7651916cd43dd8448eb211c80319c"
    assert deep_linking_module.DeepLinking.signoz_trace_url(trace_id) == (
        f"{PUBLIC_URL}/trace/{trace_id}"
    )
    assert deep_linking_module.DeepLinking.signoz_service_url() == (
        f"{PUBLIC_URL}/services/awoooi-api"
    )
    assert deep_linking_module.DeepLinking.signoz_logs_url(trace_id).startswith(
        f"{PUBLIC_URL}/logs?"
    )

    client = signoz_client_module.SignOzClient()
    generated = client.generate_trace_url(
        "awoooi-api",
        alert_timestamp=datetime(2026, 7, 15, tzinfo=UTC),
    )
    assert generated.startswith(f"{PUBLIC_URL}/traces?service=awoooi-api&")
    assert INTERNAL_URL not in generated


@pytest.mark.parametrize(
    "relative_path",
    [
        "k8s/awoooi-prod/04-configmap.yaml",
        "k8s/awoooi-dev/02-configmap.yaml",
    ],
)
def test_configmaps_publish_split_routes_and_preserve_otlp(
    relative_path: str,
) -> None:
    data = _configmap(relative_path)["data"]

    assert data["SIGNOZ_INTERNAL_URL"] == INTERNAL_URL
    assert data["SIGNOZ_PUBLIC_URL"] == PUBLIC_URL
    assert "SIGNOZ_URL" not in data
    assert data["OTEL_EXPORTER_OTLP_ENDPOINT"] == "http://192.168.0.188:24317"


def test_prod_network_policy_allows_exact_canonical_signoz_health_route() -> None:
    policy_path = REPO_ROOT / "k8s/awoooi-prod/02-network-policy.yaml"
    policies = [
        document
        for document in yaml.safe_load_all(policy_path.read_text(encoding="utf-8"))
        if document
    ]
    policy = next(
        document
        for document in policies
        if document["metadata"]["name"] == "allow-required-egress"
    )

    canonical_rules = [
        rule
        for rule in policy["spec"]["egress"]
        if rule.get("to") == [{"ipBlock": {"cidr": "192.168.0.110/32"}}]
    ]
    assert len(canonical_rules) == 1
    assert {"protocol": "TCP", "port": 8080} in canonical_rules[0]["ports"]

    legacy_rules = [
        rule
        for rule in policy["spec"]["egress"]
        if rule.get("to") == [{"ipBlock": {"cidr": "192.168.0.188/32"}}]
    ]
    assert len(legacy_rules) == 1
    assert {"protocol": "TCP", "port": 3301} not in legacy_rules[0]["ports"]
    assert {"protocol": "TCP", "port": 24317} in legacy_rules[0]["ports"]
    assert {"protocol": "TCP", "port": 24318} in legacy_rules[0]["ports"]


def test_active_api_source_has_no_legacy_signoz_ui_route() -> None:
    source_root = REPO_ROOT / "apps/api/src"
    offenders = [
        str(path.relative_to(REPO_ROOT))
        for path in source_root.rglob("*.py")
        if path.name != "public_redaction.py"
        and LEGACY_URL in path.read_text(encoding="utf-8")
    ]

    assert offenders == []


def test_cd_alert_chain_smoke_uses_canonical_public_signoz_route() -> None:
    text = ALERT_CHAIN_SMOKE.read_text(encoding="utf-8")

    assert '"SIGNOZ_PUBLIC_URL"' in text
    assert PUBLIC_URL in text
    assert LEGACY_URL not in text


def test_host_asset_map_places_signoz_on_the_canonical_query_host() -> None:
    canonical_services = HOST_CONFIGS["192.168.0.110"]["services"]
    legacy_services = HOST_CONFIGS["192.168.0.188"]["services"]

    assert ("SigNoz", 8080, "http", "/api/v1/health") in canonical_services
    assert not any(service[0] == "SigNoz" for service in legacy_services)


def test_monitoring_registry_splits_query_health_from_otlp_transport() -> None:
    registry = _configmap("ops/monitoring/service-registry.yaml")
    services = {service["name"]: service for service in registry["services"]}

    query = services["signoz-ui"]
    assert query["host"] == "192.168.0.110"
    assert query["port"] == 8080
    assert query["health_endpoint"] == "/api/v1/health"
    assert query["health_type"] == "http"
    assert query["monitoring"]["prometheus"] is True
    assert query["monitoring"]["prometheus_scrape"] is False

    collector = services["signoz-collector"]
    assert collector["host"] == "192.168.0.188"
    assert collector["port"] == 24317
    assert collector["health_type"] == "grpc"
    assert collector["monitoring"]["prometheus_scrape"] is False
    assert collector["monitoring"]["external_verifier"] == (
        "scripts/reboot-recovery/verify-signoz-otel-grpc-runtime.sh"
    )


def test_stateful_registry_places_clickhouse_on_canonical_host() -> None:
    source_registry = _configmap("ops/config/service-registry.yaml")
    configmap = _configmap("k8s/awoooi-prod/15-service-registry-configmap.yaml")
    runtime_registry = yaml.safe_load(configmap["data"]["service-registry.yaml"])

    for registry in (source_registry, runtime_registry):
        services = {service["name"]: service for service in registry["services"]}
        clickhouse = services["signoz-clickhouse"]
        assert clickhouse["host"] == "192.168.0.110"
        assert clickhouse["stateful_level"] == "BLOCK"
        assert clickhouse["alert_only"] is True


def test_active_prometheus_targets_canonical_http_health_not_legacy_tcp() -> None:
    config = _configmap("k8s/monitoring/prometheus.yml")
    jobs = {job["job_name"]: job for job in config["scrape_configs"]}
    http_job = jobs["blackbox-http"]
    http_targets = http_job["static_configs"][0]["targets"]
    tcp_targets = jobs["blackbox-tcp"]["static_configs"][0]["targets"]

    assert INTERNAL_HEALTH_URL in http_targets
    assert LEGACY_URL not in tcp_targets
    assert http_job["relabel_configs"] == [
        {
            "source_labels": ["__address__"],
            "target_label": "__param_target",
        },
        {
            "source_labels": ["__param_target"],
            "target_label": "instance",
        },
        {
            "target_label": "__address__",
            "replacement": "192.168.0.110:9115",
        },
    ]


def test_signoz_prometheus_route_deploy_is_target_first_and_rollback_safe() -> None:
    text = PROMETHEUS_ROUTE_DEPLOY.read_text(encoding="utf-8")

    assert 'MODE="${1:-apply}"' in text
    assert "promtool check config" in text
    assert "cp -p \"${REMOTE_CONFIG}\" \"${BACKUP_CONFIG}\"" in text
    assert 'action=target_first_config_apply' in text
    assert text.index("promtool check config") < text.index(
        'action=target_first_config_apply'
    )
    assert text.index('action=target_first_config_apply') < text.index(
        "terminal=target_up_two_scrapes"
    )
    assert "rollback_deploy" in text
    assert INTERNAL_HEALTH_URL in text
    assert LEGACY_URL in text


def test_legacy_prometheus_deployer_routes_to_bounded_host110_replacement() -> None:
    text = LEGACY_PROMETHEUS_DEPLOY.read_text(encoding="utf-8")

    assert "superseded_by=global_product_governance_v2" in text
    assert "expiry=2026-07-15" in text
    assert "scripts/ops/deploy-signoz-prometheus-route.sh" in text
    assert "192.168.0.188" not in text


@pytest.mark.parametrize(
    "relative_path",
    [
        "ops/monitoring/alerts-unified.yml",
        "ops/monitoring/alerts.yml",
    ],
)
def test_signoz_down_alert_is_exact_and_fails_closed_on_missing_series(
    relative_path: str,
) -> None:
    payload = _configmap(relative_path)
    rules = [
        rule
        for group in payload["groups"]
        for rule in group.get("rules", [])
        if rule.get("alert") == "SignOzDown"
    ]

    assert len(rules) == 1
    rule = rules[0]
    assert 'job="blackbox-http"' in rule["expr"]
    assert INTERNAL_HEALTH_URL in rule["expr"]
    assert "absent(" in rule["expr"]
    assert LEGACY_URL not in rule["expr"]
    assert rule["labels"]["host"] == "110"
    assert rule["labels"]["target_host"] == "192.168.0.110"


def test_legacy_188_rule_plane_does_not_duplicate_canonical_signoz_alert() -> None:
    payload = _configmap("k8s/monitoring/k3s-alerts.yaml")
    signoz_rules = [
        rule
        for group in payload["groups"]
        for rule in group.get("rules", [])
        if rule.get("alert") == "SignOzDown"
    ]
    text = (REPO_ROOT / "k8s/monitoring/k3s-alerts.yaml").read_text(encoding="utf-8")

    assert signoz_rules == []
    assert LEGACY_URL not in text


def test_monitoring_generator_uses_blackbox_without_scraping_signoz_spa(
    tmp_path: Path,
) -> None:
    subprocess.run(
        [
            sys.executable,
            str(REPO_ROOT / "ops/monitoring/generate_monitoring.py"),
            "--output-dir",
            str(tmp_path),
        ],
        cwd=REPO_ROOT,
        check=True,
        capture_output=True,
        text=True,
    )

    scrape = _configmap(str(tmp_path / "prometheus-scrape-generated.yaml"))
    blackbox = yaml.safe_load(
        (tmp_path / "blackbox-targets-generated.yaml").read_text(encoding="utf-8")
    )
    generated_jobs = {job["job_name"] for job in scrape["scrape_configs"]}
    assert "signoz-ui" not in generated_jobs
    assert "signoz-collector" not in generated_jobs
    assert any(
        target["labels"].get("service") == "signoz-ui"
        and target["url"] == INTERNAL_HEALTH_URL
        for target in blackbox
    )