mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-26 05:29:45 +00:00
rcourtman
6cd7e1c7c1
Patrol treated every guest's cluster-resources memory as cache-inclusive, so an LXC without a linked Pulse agent, which only ever gets that source, lost memory pressure entirely: no triage flag, forecast or anomaly at 97% of its limit, a "may include reclaimable cache" caveat in the seed, and older memory alerts sent to model review every run with any resolve verdict discarded. The premise holds for QEMU but not for LXC. Proxmox's container mem is cgroup usage with page cache already subtracted: PVE::CGroup's get_memory_stat returns memory.current minus memory.stat file (cgroup v2) or usage_in_bytes minus total_cache (v1), PVE::LXC::vmstatus reports it as mem, and pvestatd feeds that value to the RRD that /cluster/resources reads. A QEMU guest's listing mem is the host's view of guest RAM and does count the guest's cache. Qualification now takes the guest kind. A current, dated cluster-resources reading establishes pressure for a Proxmox container and stays unknown for QEMU; status-mem, status-freemem and derived-total-minus-used stay cache-inclusive for both, and retained readings still cannot prove pressure. The typed read state derives the kind from the resource type, so Patrol, the observer and AI queries agree. The calculateLXCMemory comment that called the listing value cache-inclusive is corrected. Contract-Neutral: monitoring change is a corrected comment in calculateLXCMemory and its tests; LXC memory collection and sources are unchanged
425 lines
22 KiB
Python
425 lines
22 KiB
Python
#!/usr/bin/env python3
|
||
"""Exercise the copied resident-memory recipe, not a Pulse memory repair.
|
||
|
||
The fixture adapter runs real awk over synthetic status files. A separate
|
||
check reads only a disposable child process's live Linux counters; it does not
|
||
inspect a host service, database, heap or container daemon.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import select
|
||
import shutil
|
||
import signal
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import unittest
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[2]
|
||
DOC = ROOT / "docs/TROUBLESHOOTING.md"
|
||
COUNTERS = "VmRSS: 8192 kB\nRssAnon: 6144 kB\nRssFile: 1536 kB\nRssShmem: 512 kB\nVmSwap: 1024 kB\n"
|
||
IDENTITY = "4321 Tue Oct 6 00:00:00 2026"
|
||
|
||
|
||
def section() -> str:
|
||
return DOC.read_text(encoding="utf-8").split(
|
||
"#### Memory use keeps growing\n", 1
|
||
)[1].split("\n#### ", 1)[0]
|
||
|
||
|
||
def recipe() -> str:
|
||
blocks = re.findall(r"```bash\n(.*?)```", section(), re.DOTALL)
|
||
if len(blocks) != 1:
|
||
raise AssertionError("expected one bounded memory recipe")
|
||
return blocks[0]
|
||
|
||
|
||
READER = '''#!/usr/bin/env python3
|
||
import json, os, pathlib, signal, subprocess, sys, time
|
||
name = pathlib.Path(sys.argv[0]).name
|
||
args = sys.argv[1:]
|
||
root = pathlib.Path(os.environ['RECIPE_FIXTURE'])
|
||
log = root / 'calls.jsonl'
|
||
previous = [json.loads(line) for line in log.read_text().splitlines()] if log.exists() else []
|
||
count = sum(call['name'] == name for call in previous)
|
||
with log.open('a') as stream:
|
||
stream.write(json.dumps({'name': name, 'args': args}) + '\\n')
|
||
if name == os.environ.get('HANG_TOOL'):
|
||
if os.environ.get('IGNORE_TERM'):
|
||
signal.signal(signal.SIGTERM, signal.SIG_IGN)
|
||
record = root / 'hung-reader.tmp'
|
||
record.write_text(json.dumps({'pid': os.getpid()}))
|
||
record.replace(root / 'hung-reader.json')
|
||
if os.environ.get('CLOSE_PIPES'):
|
||
os.close(1); os.close(2)
|
||
time.sleep(100)
|
||
if name == 'date':
|
||
assert args == ['-u', '+%Y-%m-%dT%H:%M:%SZ']
|
||
print('2026-10-06T00:00:00Z')
|
||
elif name == 'systemctl':
|
||
assert args == ['show', 'pulse', '--property=MainPID', '--value']
|
||
print(os.environ.get(f'PID_{count}', '4321'))
|
||
sys.exit(int(os.environ.get(f'SERVICE_EXIT_{count}', '0')))
|
||
elif name == 'ps':
|
||
assert args == ['-p', '4321', '-o', 'pid=,lstart=']
|
||
assert os.environ['TZ'] == 'UTC'
|
||
print(os.environ.get(f'IDENTITY_{count}', '4321 Tue Oct 6 00:00:00 2026'))
|
||
sys.exit(int(os.environ.get(f'PROCESS_EXIT_{count}', '0')))
|
||
elif name == 'awk':
|
||
assert args[-1] == '/proc/4321/status'
|
||
if os.environ.get('AWK_ERROR'):
|
||
print('VmRSS: 8192 kB')
|
||
print('private fixture reader error', file=sys.stderr)
|
||
sys.exit(3)
|
||
target = root / ('missing' if os.environ.get('STATUS_MISSING') else 'status')
|
||
sys.exit(subprocess.run([os.environ['REAL_AWK'], *args[:-1], str(target)]).returncode)
|
||
else:
|
||
raise AssertionError('unexpected command')
|
||
'''
|
||
|
||
|
||
def run_owned_recipe(copied: str, env: dict[str, str]) -> subprocess.CompletedProcess:
|
||
"""Keep the ten-second budget and observe actual owned-reader termination.
|
||
|
||
A pidfd binds observation and emergency cleanup to the fixture process,
|
||
even if its pipes close or its PID disappears. No service is signalled.
|
||
"""
|
||
with subprocess.Popen(["bash", "-c", copied], env=env, text=True,
|
||
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||
start_new_session=True) as process:
|
||
reader_fd = None
|
||
reader_record_missing = False
|
||
try:
|
||
if env.get("HANG_TOOL"):
|
||
record = Path(env["RECIPE_FIXTURE"]) / "hung-reader.json"
|
||
deadline = time.monotonic() + 2
|
||
while not record.exists() and process.poll() is None and time.monotonic() < deadline:
|
||
time.sleep(0.01)
|
||
if record.exists():
|
||
reader_fd = os.pidfd_open(json.loads(record.read_text())["pid"])
|
||
reader_exit = select.poll()
|
||
reader_exit.register(reader_fd, select.POLLIN)
|
||
else:
|
||
reader_record_missing = True
|
||
stdout, stderr = process.communicate(timeout=10)
|
||
if reader_record_missing:
|
||
raise AssertionError("owned hung-reader identity unobserved")
|
||
if reader_fd is not None and not reader_exit.poll(500):
|
||
raise AssertionError("owned reader survived the collection deadline")
|
||
except subprocess.TimeoutExpired:
|
||
# Never target a service or another test's group. TimeoutExpired
|
||
# remains an error, not a passing or empty sample.
|
||
if reader_fd is not None and not reader_exit.poll(0):
|
||
signal.pidfd_send_signal(reader_fd, signal.SIGKILL)
|
||
try:
|
||
os.killpg(process.pid, signal.SIGKILL)
|
||
except ProcessLookupError:
|
||
pass
|
||
process.communicate(timeout=5)
|
||
raise
|
||
finally:
|
||
if reader_fd is not None:
|
||
if not reader_exit.poll(0):
|
||
signal.pidfd_send_signal(reader_fd, signal.SIGKILL)
|
||
if not reader_exit.poll(1000):
|
||
raise AssertionError("test-owned reader cleanup unobserved")
|
||
os.close(reader_fd)
|
||
return subprocess.CompletedProcess(process.args, process.returncode, stdout, stderr)
|
||
|
||
|
||
class MemoryTroubleshootingDocsTest(unittest.TestCase):
|
||
def exercise(self, *, status: str = COUNTERS, copied: str | None = None,
|
||
**settings: str):
|
||
with tempfile.TemporaryDirectory() as temporary:
|
||
directory = Path(temporary)
|
||
for name in ("date", "systemctl", "ps", "awk"):
|
||
command = directory / name
|
||
command.write_text(READER, encoding="utf-8")
|
||
command.chmod(0o700)
|
||
(directory / "status").write_text(status, encoding="utf-8")
|
||
env = dict(os.environ, PATH=f"{directory}:{os.environ['PATH']}",
|
||
RECIPE_FIXTURE=str(directory), REAL_AWK=shutil.which("awk"), **settings)
|
||
result = run_owned_recipe(copied if copied is not None else recipe(), env)
|
||
log = directory / "calls.jsonl"
|
||
calls = [json.loads(line) for line in log.read_text().splitlines()] if log.exists() else []
|
||
return result, calls
|
||
|
||
def assert_no_memory_claim(self, result):
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertEqual(result.stdout, "")
|
||
self.assertNotIn("Pulse process (PID and UTC start)", result.stdout)
|
||
self.assertNotRegex(result.stdout, r"VmRSS:|RssAnon:|RssFile:|RssShmem:|VmSwap:")
|
||
|
||
def test_shipped_guide_is_the_exercised_source_and_shell_is_valid(self):
|
||
self.assertEqual(DOC.read_bytes(),
|
||
(ROOT / "frontend-modern/public/docs/TROUBLESHOOTING.md").read_bytes())
|
||
result = subprocess.run(["bash", "-n"], input=recipe(), text=True, capture_output=True)
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
|
||
def test_only_five_counters_and_stable_process_identity_leave_the_reader(self):
|
||
result, calls = self.exercise(status="Name: private-workload\nUid: 1001 1001\n"
|
||
+ COUNTERS + "VmSize: 1234567 kB\nThreads: 19\n")
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
self.assertIn("2026-10-06T00:00:00Z", result.stdout)
|
||
self.assertIn(IDENTITY, result.stdout)
|
||
self.assertIn(COUNTERS, result.stdout)
|
||
for private in ("private-workload", "Uid:", "VmSize:", "Threads:"):
|
||
self.assertNotIn(private, result.stdout)
|
||
self.assertEqual([call["name"] for call in calls],
|
||
["date", "systemctl", "ps", "awk", "systemctl", "ps"])
|
||
|
||
def test_missing_or_invalid_pid_stops_before_a_process_read(self):
|
||
for pid in ("", "0", "not-a-pid", "4321 extra"):
|
||
with self.subTest(pid=pid):
|
||
result, calls = self.exercise(PID_0=pid)
|
||
self.assert_no_memory_claim(result)
|
||
self.assertIn("sample unavailable", result.stderr)
|
||
self.assertEqual([call["name"] for call in calls], ["date", "systemctl"])
|
||
|
||
def test_service_and_process_failures_remain_failures(self):
|
||
for key, names in (
|
||
("SERVICE_EXIT_0", ["date", "systemctl"]),
|
||
("PROCESS_EXIT_0", ["date", "systemctl", "ps"]),
|
||
("SERVICE_EXIT_1", ["date", "systemctl", "ps", "awk", "systemctl"]),
|
||
("PROCESS_EXIT_1", ["date", "systemctl", "ps", "awk", "systemctl", "ps"]),
|
||
):
|
||
with self.subTest(key=key):
|
||
result, calls = self.exercise(**{key: "3"})
|
||
self.assert_no_memory_claim(result)
|
||
self.assertEqual(result.returncode, 3)
|
||
self.assertEqual([call["name"] for call in calls], names)
|
||
|
||
def test_empty_initial_identity_is_not_a_complete_sample(self):
|
||
result, calls = self.exercise(IDENTITY_0="")
|
||
self.assert_no_memory_claim(result)
|
||
self.assertIn("sample unavailable", result.stderr)
|
||
self.assertNotIn("awk", [call["name"] for call in calls])
|
||
|
||
def test_missing_unreadable_or_incomplete_status_does_not_print_partial_counters(self):
|
||
for status, settings in ((COUNTERS, {"STATUS_MISSING": "1"}),
|
||
("", {}), (COUNTERS.replace("RssAnon: 6144 kB\n", ""), {})):
|
||
with self.subTest(status=status, settings=settings):
|
||
result, calls = self.exercise(status=status, **settings)
|
||
self.assert_no_memory_claim(result)
|
||
self.assertEqual([call["name"] for call in calls],
|
||
["date", "systemctl", "ps", "awk"])
|
||
|
||
def test_malformed_units_numbers_duplicates_and_extra_fields_fail_closed(self):
|
||
for value in ("-1 kB", "unknown kB", "1 MB", "1", "1 kB extra"):
|
||
with self.subTest(value=value):
|
||
result, _ = self.exercise(status=COUNTERS.replace("8192 kB", value))
|
||
self.assert_no_memory_claim(result)
|
||
result, _ = self.exercise(status=COUNTERS + "VmRSS: 8192 kB\n")
|
||
self.assert_no_memory_claim(result)
|
||
|
||
def test_service_restart_or_pid_reuse_discards_the_captured_counters(self):
|
||
for settings in ({"PID_1": "9999"}, {"PID_1": "0"},
|
||
{"IDENTITY_1": "4321 Tue Oct 6 00:00:01 2026"},
|
||
{"IDENTITY_1": ""}):
|
||
with self.subTest(settings=settings):
|
||
result, _ = self.exercise(**settings)
|
||
self.assert_no_memory_claim(result)
|
||
self.assertIn("sample unavailable", result.stderr)
|
||
|
||
def test_measured_zero_is_retained_without_interpreting_missing_as_zero(self):
|
||
result, _ = self.exercise(status=re.sub(r"\d+ kB", "0 kB", COUNTERS))
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
self.assertEqual(result.stdout.count("0 kB"), 5)
|
||
|
||
def test_hung_service_reader_is_unavailable_before_the_test_deadline(self):
|
||
started = time.monotonic()
|
||
result, calls = self.exercise(HANG_TOOL="systemctl")
|
||
self.assertLess(time.monotonic() - started, 10)
|
||
self.assertEqual(result.returncode, 137, result.stderr)
|
||
self.assert_no_memory_claim(result)
|
||
self.assertIn("sample unavailable", result.stderr)
|
||
self.assertEqual([call["name"] for call in calls], ["date", "systemctl"])
|
||
|
||
def test_hung_counter_reader_never_publishes_identity_or_partial_memory(self):
|
||
started = time.monotonic()
|
||
result, calls = self.exercise(HANG_TOOL="awk")
|
||
self.assertLess(time.monotonic() - started, 10)
|
||
self.assertEqual(result.returncode, 137, result.stderr)
|
||
self.assert_no_memory_claim(result)
|
||
self.assertIn("sample unavailable", result.stderr)
|
||
self.assertEqual([call["name"] for call in calls], ["date", "systemctl", "ps", "awk"])
|
||
|
||
def test_missing_deadline_utility_is_not_a_sample_or_an_unbounded_fallback(self):
|
||
copied = recipe().replace("timeout", "pulse-missing-memory-deadline")
|
||
self.assertNotEqual(copied, recipe())
|
||
result, calls = self.exercise(copied=copied)
|
||
self.assertEqual(result.returncode, 1, result.stderr)
|
||
self.assert_no_memory_claim(result)
|
||
self.assertIn("sample unavailable", result.stderr)
|
||
self.assertEqual(calls, [])
|
||
|
||
def test_term_resistant_readers_are_stopped_even_when_their_pipes_close(self):
|
||
for tool in ("systemctl", "awk"):
|
||
with self.subTest(tool=tool):
|
||
started = time.monotonic()
|
||
result, calls = self.exercise(HANG_TOOL=tool, IGNORE_TERM="1", CLOSE_PIPES="1")
|
||
self.assertLess(time.monotonic() - started, 10)
|
||
self.assertEqual(result.returncode, 137, result.stderr)
|
||
self.assert_no_memory_claim(result)
|
||
self.assertEqual(calls[-1]["name"], tool)
|
||
|
||
def test_shell_exit_alone_does_not_pass_the_reader_termination_oracle(self):
|
||
copied = recipe().replace("--signal=KILL 8s", "--kill-after=1s 8s")
|
||
self.assertNotEqual(copied, recipe())
|
||
with self.assertRaisesRegex(AssertionError, "reader survived the collection deadline"):
|
||
self.exercise(copied=copied, HANG_TOOL="systemctl", IGNORE_TERM="1", CLOSE_PIPES="1")
|
||
|
||
def test_reader_failure_does_not_expose_raw_errors_or_partial_output(self):
|
||
result, _ = self.exercise(AWK_ERROR="1")
|
||
self.assert_no_memory_claim(result)
|
||
self.assertEqual(result.returncode, 3)
|
||
self.assertEqual(result.stderr, "Resident-memory sample unavailable (exit 3); stop sampling.\n")
|
||
|
||
@unittest.skipUnless(sys.platform.startswith("linux"), "Linux /proc recipe")
|
||
def test_real_linux_status_of_an_owned_disposable_process(self):
|
||
with tempfile.TemporaryDirectory() as temporary:
|
||
directory = Path(temporary)
|
||
child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
|
||
try:
|
||
command = directory / "systemctl"
|
||
command.write_text("#!/bin/sh\nprintf '%s\\n' \"$RECIPE_CHILD_PID\"\n", encoding="utf-8")
|
||
command.chmod(0o700)
|
||
env = dict(os.environ, PATH=f"{directory}:{os.environ['PATH']}",
|
||
RECIPE_CHILD_PID=str(child.pid))
|
||
result = run_owned_recipe(recipe(), env)
|
||
self.assertEqual(result.returncode, 0, result.stderr)
|
||
self.assertIn("Pulse process (PID and UTC start)", result.stdout)
|
||
for field in ("VmRSS", "RssAnon", "RssFile", "RssShmem", "VmSwap"):
|
||
self.assertRegex(result.stdout, rf"(?m)^{field}:\s+[0-9]+ kB$")
|
||
self.assertNotIn("Name:", result.stdout)
|
||
self.assertNotIn("Uid:", result.stdout)
|
||
finally:
|
||
child.terminate()
|
||
child.wait(timeout=5)
|
||
|
||
def test_guidance_distinguishes_scope_units_and_unsafe_shortcuts(self):
|
||
guide = " ".join(section().split())
|
||
for phrase in ("container, not just Pulse", "own cache accounting", "Go heap size",
|
||
"inside the Pulse container", "in both `systemctl` calls",
|
||
"1,024 bytes", "approximate and not an atomic snapshot",
|
||
"not specifically the Go heap", "different run",
|
||
"do not measure Pulse RSS", "do not assume its PID 1 is Pulse",
|
||
"requires GNU `timeout`", "ignore TERM", "No timestamp",
|
||
"does not elevate privileges or prompt", "do not add `sudo` inside it",
|
||
"do not remove the deadline", "does not stop or restart Pulse",
|
||
"Do not restart Pulse, drop caches, force garbage collection",
|
||
"Do not post a heap dump", "existing screenshots remain useful"):
|
||
self.assertIn(phrase, guide)
|
||
self.assertNotRegex(recipe(),
|
||
r"(?<!-)\b(?:sudo|restart|kill|rm|truncate|sqlite3|curl|printenv|cat)\b"
|
||
r"|--follow|--token|cmdline|environ|smaps")
|
||
|
||
def test_baseline_and_growth_are_distinct_without_manufactured_comparisons(self):
|
||
guide = " ".join(section().split())
|
||
for distinction in (
|
||
"A higher steady baseline is also worth reporting",
|
||
"continuing growth is not required",
|
||
"a plateau does not dismiss the increased requirement",
|
||
"not by itself a leak or its cause",
|
||
"Different runs at different ages, not verified recovery",
|
||
"comparable uptimes and workloads",
|
||
"otherwise keep the mismatch explicit",
|
||
"Do not restart, downgrade or change retention, polling or enabled features",
|
||
"CPU improvement is not evidence that the resident-memory baseline fell",
|
||
):
|
||
self.assertIn(distinction, guide)
|
||
faq = " ".join((ROOT / "docs/FAQ.md").read_text().split())
|
||
self.assertIn("higher steady baseline is worth reporting", faq)
|
||
self.assertIn("not a like-for-like comparison", faq)
|
||
self.assertIn("do not restart to create a comparison", faq)
|
||
self.assertIn("TROUBLESHOOTING.md#memory-use-keeps-growing", faq)
|
||
self.assertEqual((ROOT / "docs/FAQ.md").read_bytes(),
|
||
(ROOT / "frontend-modern/public/docs/FAQ.md").read_bytes())
|
||
|
||
|
||
class GuestMemoryInterpretationDocsTest(unittest.TestCase):
|
||
HEADING = "#### Guest memory is high but available memory is plentiful\n"
|
||
ANCHOR = "#guest-memory-is-high-but-available-memory-is-plentiful"
|
||
|
||
def guidance(self):
|
||
text = DOC.read_text(encoding="utf-8")
|
||
self.assertEqual(text.count(self.HEADING), 1)
|
||
return " ".join(text.split(self.HEADING, 1)[1].split("\n#### ", 1)[0].split())
|
||
|
||
def test_guest_and_process_entry_points_are_separate_and_mirrored(self):
|
||
for name in ("TROUBLESHOOTING.md", "FAQ.md", "CONFIGURATION.md"):
|
||
self.assertEqual((ROOT / "docs" / name).read_bytes(),
|
||
(ROOT / "frontend-modern/public/docs" / name).read_bytes())
|
||
for name in ("FAQ.md", "CONFIGURATION.md"):
|
||
self.assertIn("TROUBLESHOOTING.md" + self.ANCHOR,
|
||
(ROOT / "docs" / name).read_text())
|
||
self.assertIn(self.ANCHOR, section())
|
||
self.assertIn("Pulse server process", section())
|
||
self.assertIn("not a monitored guest", section())
|
||
|
||
def test_comparisons_keep_guest_kernel_units_and_cache_semantics(self):
|
||
guide = self.guidance()
|
||
for distinction in (
|
||
"same guest, kernel and time", "node's available memory is not the VM's",
|
||
"single application's RSS is not total guest usage",
|
||
"Do not sum process RSS", "not all cache is immediately reclaimable",
|
||
):
|
||
self.assertIn(distinction, guide)
|
||
self.assertIn("a guarantee of every allocation succeeding", guide)
|
||
self.assertIn("**20 GiB total** and **15.2 GiB available**", guide)
|
||
self.assertIn("(20 - 15.2) / 20 × 100 = 24%", guide)
|
||
self.assertIn("not a nominal configured limit", guide)
|
||
self.assertIn("do not mix GB with GiB or kernel `kB` (1,024 bytes)", guide)
|
||
self.assertIn("Do not subtract all `buff/cache`", guide)
|
||
self.assertIn("substitute `free` for `available`", guide)
|
||
# Existing collectors use named kernel fields, not free-table columns.
|
||
parser = (ROOT / "pkg/proxmox/client.go").read_text()
|
||
self.assertIn('"MemAvailable"', parser)
|
||
|
||
def test_missing_or_retained_evidence_and_controls_are_not_recovery(self):
|
||
guide = self.guidance()
|
||
for distinction in (
|
||
"Missing is not zero", "prior low value", "not current recovery",
|
||
"source and original observation time", "Unknown source, time or counters",
|
||
"not a correction to apply blindly", "Patrol finding has cleared",
|
||
"does not prove that the API-only path recovered",
|
||
"does not hide the reading, dismiss a separate Patrol finding or establish recovery",
|
||
"Do not disable all monitoring or change thresholds",
|
||
):
|
||
self.assertIn(distinction, guide)
|
||
patrol = (ROOT / "internal/ai/patrol_guest_memory.go").read_text()
|
||
evidence = (ROOT / "internal/unifiedresources/guest_memory.go").read_text()
|
||
# Patrol delegates interpretation and display to the shared selected-
|
||
# sample formatter. Keep the guide bound to that implementation, not
|
||
# the former location of its warning strings.
|
||
self.assertIn("unifiedresources.QualifyGuestMemoryForKind(", patrol)
|
||
self.assertIn("unifiedresources.GuestMemoryEvidence{", patrol)
|
||
self.assertIn("}).Format(r.percent)", patrol)
|
||
self.assertIn('"guest pressure unknown"', evidence)
|
||
self.assertIn('"N/A (guest memory unavailable; not evidence of recovery)"', evidence)
|
||
|
||
def test_guidance_uses_existing_evidence_without_new_guest_reads(self):
|
||
guide = self.guidance()
|
||
for boundary in (
|
||
"existing observations", "existing screenshots", "original sequence",
|
||
"Do not publish raw API responses", "full `/proc` output",
|
||
"do not run Diagnostics, guest-agent probes or another Patrol run",
|
||
"install an agent, restart the guest, drop caches or change memory limits",
|
||
"If the guest is frozen or unresponsive, stop these comparisons",
|
||
"VM_DISK_MONITORING.md#backup-safety",
|
||
):
|
||
self.assertIn(boundary, guide)
|
||
self.assertNotIn("```", guide, "interpret evidence already collected, not a new command")
|
||
self.assertIn("CONFIGURATION.md#metric-thresholds-off-and-inheritance", guide)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|