mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-11-04 01:23:48 +00:00
pulse-triage[bot]
34e3502630
Explain the queue-response failure reported on stable without treating an unread dataset as an empty queue or an authentication failure. Keep native queue monitoring and private, bounded evidence as the fallback rather than DNS, privilege or mail mutations. Contract-Neutral: Existing PMG help and safety regression coverage only; no collector, API, schema, permission or navigation behaviour changes. Change-source: pulse-maintainer
214 lines
12 KiB
Python
214 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Keep existing PMG guidance tied to the actual navigation and read limits.
|
|
|
|
Static source/document checks only: no gateway, mail, credentials, network,
|
|
service, notification or live diagnostic operation is performed.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
import re
|
|
import unittest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
DOC = ROOT / "docs/MAIL_GATEWAY.md"
|
|
|
|
|
|
def frontend(path):
|
|
return (ROOT / "frontend-modern/src" / path).read_text(encoding="utf-8")
|
|
|
|
|
|
def plain(text):
|
|
return " ".join(text.replace("**", "").split())
|
|
|
|
|
|
def go_block(text, opening):
|
|
"""Read one known source block, including its nested bodies."""
|
|
start = text.index(opening) + len(opening)
|
|
depth = 1
|
|
for end in range(start, len(text)):
|
|
depth += (text[end] == "{") - (text[end] == "}")
|
|
if depth == 0:
|
|
return text[start:end]
|
|
raise AssertionError(f"Unclosed source block: {opening}")
|
|
|
|
|
|
class PMGDocsTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.text = DOC.read_text(encoding="utf-8")
|
|
self.guide = plain(self.text)
|
|
|
|
def test_shipped_guide_matches(self):
|
|
self.assertEqual(DOC.read_bytes(),
|
|
(ROOT / "frontend-modern/public/docs/MAIL_GATEWAY.md").read_bytes())
|
|
|
|
def test_connection_instructions_match_existing_picker_and_form(self):
|
|
picker = frontend("components/Settings/InfrastructureSourcePicker.tsx")
|
|
authentication = frontend("components/Settings/NodeModalAuthenticationSection.tsx")
|
|
self.assertIn("Show more sources", picker)
|
|
self.assertIn("Show more sources", self.guide)
|
|
self.assertIn("Settings → Infrastructure → Add infrastructure", self.guide)
|
|
self.assertNotIn("**Add Node**", self.text)
|
|
self.assertIn("Username & Password", authentication)
|
|
self.assertIn("Username & Password", self.guide)
|
|
self.assertIn("modalProps.nodeType !== 'pmg'", authentication)
|
|
for phrase in ("not the PVE/PBS API-token setup", "minimum read permissions",
|
|
"password only in the private settings form",
|
|
"do not put it in a command, URL or issue report",
|
|
"Keep HTTPS and certificate verification enabled"):
|
|
self.assertIn(phrase, self.guide)
|
|
|
|
def test_navigation_and_dataset_detail_match_the_current_surface(self):
|
|
model = frontend("features/proxmox/proxmoxPageModel.ts")
|
|
route = re.search(r"id: 'mail', label: '([^']+)', path: '([^']+)'", model)
|
|
self.assertIsNotNone(route)
|
|
self.assertIn(f"**Proxmox → {route[1]}** (`{route[2]}`)", self.text)
|
|
table = frontend("features/proxmox/ProxmoxMailGatewayTable.tsx")
|
|
self.assertIn("No instances match current filters", table)
|
|
self.assertIn("clear its search and status filters", self.guide)
|
|
self.assertNotIn("on the **Infrastructure** page", self.text)
|
|
drawer = frontend("features/proxmox/ProxmoxMailGatewayDrawer.tsx")
|
|
self.assertIn(".slice(0, 8)", drawer)
|
|
self.assertIn("top eight reported domains, not the entire domain inventory", self.guide)
|
|
# The mail-flow card names the window its counts cover.
|
|
self.assertRegex(drawer, r"stats(?:\(\))?\??\.timeframe")
|
|
self.assertIn("compare the same window in PMG", self.guide)
|
|
self.assertIn("Expand the intended gateway's row", self.guide)
|
|
|
|
def test_documented_thresholds_exist_and_keep_their_units(self):
|
|
config = (ROOT / "internal/alerts/config/types.go").read_text()
|
|
fields = config.split("type PMGThresholdConfig struct {", 1)[1].split("\n}", 1)[0]
|
|
for key in ("queueTotalWarning", "deferredQueueWarn", "holdQueueWarn",
|
|
"oldestMessageWarnMins", "quarantineSpamWarn", "quarantineVirusWarn",
|
|
"quarantineGrowthWarnPct", "quarantineGrowthWarnMin"):
|
|
self.assertIn(f'json:"{key}"', fields)
|
|
self.assertNotRegex(fields, r'json:"(?:spamRate|deliveryFailures)')
|
|
self.assertIn("<ThresholdsTableProxmoxPMGSection", frontend(
|
|
"components/Alerts/ThresholdsTableProxmoxTab.tsx"))
|
|
title = re.search(r"ALERT_THRESHOLDS_SECTION_TITLE_PMG = '([^']+)'",
|
|
frontend("utils/alertThresholdsPresentation.ts"))
|
|
self.assertIsNotNone(title)
|
|
self.assertIn(f"**{title[1]}**", self.text)
|
|
for phrase in ("Alerts → Thresholds → Proxmox", "oldest-message age in minutes",
|
|
"percentage plus minimum message growth",
|
|
"do not expose a user-configurable spam-rate or delivery-failure threshold"):
|
|
self.assertIn(phrase, self.guide)
|
|
self.assertIn("m.checkPMGAnomalies(pmg, pmgDefaults)",
|
|
(ROOT / "internal/alerts/pmg.go").read_text())
|
|
self.assertIn("connectivity and mail anomaly checks", self.guide)
|
|
|
|
def test_partial_collection_is_not_a_healthy_zero(self):
|
|
poll = (ROOT / "internal/monitoring/monitor_pbs_pmg.go").read_text()
|
|
poll = poll.split("func (m *Monitor) pollPMGInstance(", 1)[1]
|
|
self.assertLess(poll.index('pmgInst.ConnectionHealth = "healthy"'),
|
|
poll.index("client.GetMailStatistics"))
|
|
self.assertIn("Healthy is not complete collection", self.guide)
|
|
self.assertIn("Some missing fields can display as zero or an empty section", self.guide)
|
|
self.assertIn("does not prove that every dataset was read", self.guide)
|
|
self.assertIn("Unavailable evidence is unknown, not zero", self.guide)
|
|
|
|
def test_saved_scope_guidance_matches_each_owning_gate(self):
|
|
poll = (ROOT / "internal/monitoring/monitor_pbs_pmg.go").read_text()
|
|
poll = poll.split("func (m *Monitor) pollPMGInstance(", 1)[1]
|
|
for condition, methods, description in (
|
|
("MailStatsEnabled()", ("GetMailStatistics", "GetMailCount", "GetSpamScores"),
|
|
"Mail statistics, hourly mail counts and spam-score distribution"),
|
|
("MonitorQueues", ("GetQueueStatus",), "Each collected cluster node's queue status"),
|
|
("MonitorQuarantine", ("GetQuarantineStatus",), "Both spam and virus quarantine counts"),
|
|
("MonitorDomainStats", ("ListRelayDomains", "GetDomainStatistics"),
|
|
"Relay-domain inventory and domain statistics"),
|
|
):
|
|
body = go_block(poll, f"if instanceCfg.{condition} {{")
|
|
for method in methods:
|
|
self.assertIn(f"client.{method}(", body)
|
|
self.assertEqual(body.count(f"client.{method}("), poll.count(f"client.{method}("))
|
|
self.assertIn(description, self.guide)
|
|
for phrase in ("saved switches control these subsequent ordinary poll requests",
|
|
"Edits apply to the next poll", "not requests already in flight",
|
|
"does not dismiss an existing alert or establish recovery"):
|
|
self.assertIn(phrase, self.guide)
|
|
|
|
def test_legacy_scope_and_explicit_false_readback_are_not_overpromised(self):
|
|
cfg = (ROOT / "internal/config/config.go").read_text()
|
|
default = go_block(cfg, "func (i PMGInstance) MailStatsEnabled() bool {")
|
|
self.assertIn("!i.MonitoringConfigured", default)
|
|
for flag in ("MonitorQueues", "MonitorQuarantine", "MonitorDomainStats"):
|
|
self.assertIn(f"!i.{flag}", default)
|
|
wire = (ROOT / "internal/api/configapi/config_handlers.go").read_text()
|
|
response = go_block(wire, "type NodeResponse struct {")
|
|
for flag in ("monitorMailStats", "monitorQueues", "monitorQuarantine", "monitorDomainStats"):
|
|
self.assertIn(f'json:"{flag}"', response)
|
|
self.assertNotIn(f'json:"{flag},omitempty"', response)
|
|
for phrase in ("Saving all four options off preserves that choice when settings are reopened",
|
|
"keeps the legacy mail-statistics default",
|
|
"until its collection scope is explicitly saved",
|
|
"cannot distinguish an old opt-out from an unset default"):
|
|
self.assertIn(phrase, self.guide)
|
|
|
|
def test_dataset_opt_out_is_not_a_connection_pause(self):
|
|
poll = (ROOT / "internal/monitoring/monitor_pbs_pmg.go").read_text()
|
|
poll = poll.split("func (m *Monitor) pollPMGInstance(", 1)[1]
|
|
for condition in ("MailStatsEnabled()", "MonitorQueues", "MonitorQuarantine", "MonitorDomainStats"):
|
|
body = go_block(poll, f"if instanceCfg.{condition} {{")
|
|
for method in ("GetVersion", "GetClusterStatus", "ListBackups"):
|
|
self.assertIn(f"client.{method}(", poll)
|
|
self.assertNotIn(f"client.{method}(", body)
|
|
for phrase in ("Version, cluster and configuration-backup reads are independent",
|
|
"all four off does not mean no PMG requests"):
|
|
self.assertIn(phrase, self.guide)
|
|
self.assertIn("if instanceCfg.Disabled", poll)
|
|
self.assertLess(poll.index("if instanceCfg.Disabled"), poll.index("client.GetVersion"))
|
|
for phrase in ("pause the PMG connection", "does not cancel a request already in flight",
|
|
"arrange independent coverage", "only the connection you deliberately paused"):
|
|
self.assertIn(phrase, self.guide)
|
|
|
|
def test_published_warning_is_bounded_to_the_named_release(self):
|
|
for phrase in ("Published v6.5.0 does not contain this scope correction",
|
|
"In that version, mail, queue and quarantine requests can still run",
|
|
"domain statistics alone follow their switch",
|
|
"assume development-source behaviour is present in an installed release"):
|
|
self.assertIn(phrase, self.guide)
|
|
|
|
def test_incomplete_collection_does_not_promise_zero_or_recovery(self):
|
|
for phrase in ("records downstream read failures as incomplete collection",
|
|
"retaining the successful version connection",
|
|
"Missing readings do not become observed zero totals",
|
|
"Partial queue data can still raise the highest configured severity",
|
|
"cannot clear or downgrade an existing alert",
|
|
"quarantine recovery requires both categories",
|
|
"do not make every empty display a verified zero"):
|
|
self.assertIn(phrase, self.guide)
|
|
|
|
def test_diagnostics_preserve_mail_privacy_and_separate_failure_classes(self):
|
|
for phrase in ("refusal is not a password verdict", "Certificate validation error",
|
|
"401/403", "specific failed request", "configured polling interval",
|
|
"Do not send test mail", "release quarantine",
|
|
"Do not enable Debug", "bounded Pulse log readers",
|
|
"addresses, subjects, message contents, full API responses"):
|
|
self.assertIn(phrase, self.guide)
|
|
self.assertNotRegex(self.text, r"```(?:bash|sh|powershell|http)\b")
|
|
self.assertIn("Pulse's own alert email uses a separate SMTP destination", self.guide)
|
|
self.assertIn("does not guarantee a notification was routed or received", self.guide)
|
|
self.assertIn("TROUBLESHOOTING.md#test-succeeds-but-real-alerts-are-missing", self.text)
|
|
|
|
def test_decoder_failure_preserves_native_coverage_without_gateway_changes(self):
|
|
section = self.guide.split("Response decoding is not a network verdict", 1)[1]
|
|
section = section.split("For a new connection", 1)[0]
|
|
for phrase in ("pmg_queue_status", "failed to decode response",
|
|
"cannot unmarshal array", "received a response it could not interpret",
|
|
"does not, by itself, establish", "queue is empty",
|
|
"separate transport or permission error",
|
|
"Test Connection does not resolve the decoding failure",
|
|
"PMG's own existing queue view", "short redacted error",
|
|
"do not attach the full queue response",
|
|
"Do not change DNS, disable certificate verification, grant broader permissions",
|
|
"retry, delete or release mail",
|
|
"Turning off queue collection removes coverage",
|
|
"does not repair decoding"):
|
|
self.assertIn(phrase, section)
|
|
self.assertNotRegex(section, r"https?://|```|\b(?:curl|wget)\b")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|