from __future__ import annotations from src.api.v1.monitoring import public_monitoring_tool_payload from src.services.public_redaction import redact_public_lan_text, redact_public_lan_topology def test_redact_public_lan_text_replaces_internal_endpoints_with_aliases() -> None: value = ( "image=192.168.0.110:5000/library/api " "scanner=http://192.168.0.112:8080/health " "ollama=`192.168.0.188:11434` " "unknown=192.168.0.222:1234" ) redacted = redact_public_lan_text(value) assert "192.168.0." not in redacted assert "host:public-gateway/registry/library/api" in redacted assert "scanner=host:kali-readonly/scanner/health" in redacted assert "ollama=`host:observability-a/ollama`" in redacted assert "unknown=host:internal-node" in redacted def test_redact_public_lan_topology_recurses_json_values() -> None: payload = { "safe_key": "unchanged", "nested": [{"endpoint": "192.168.0.188:3301"}], } redacted = redact_public_lan_topology(payload) assert redacted["safe_key"] == "unchanged" assert redacted["nested"][0]["endpoint"] == "host:observability-a/signoz" def test_redact_public_lan_text_replaces_internal_work_context_terms() -> None: redacted = redact_public_lan_text("不得顯示工作視窗對話、批准!繼續 或 source_thread_id") assert "工作視窗" not in redacted assert "批准!" not in redacted assert "source_thread_id" not in redacted assert "內部協作環境" in redacted def test_public_monitoring_tool_payload_drops_internal_probe_url() -> None: payload = public_monitoring_tool_payload( { "name": "Grafana", "status": "up", "url": "http://192.168.0.110:3002", } ) assert "url" not in payload def test_public_monitoring_tool_payload_uses_public_route_when_available() -> None: payload = public_monitoring_tool_payload( { "name": "SigNoz", "status": "up", "url": "http://192.168.0.188:3301", } ) assert payload["url"] == "https://signoz.wooo.work"