test_core_improvements.py #1

#!/usr/bin/env python3
"""Offline coverage of bounded cleanup and retained/correlated run logs."""
import importlib.util
import fcntl
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch

import test_core_review as core
import test_runtime_review as runtime
import test_maintenance_review as maintenance


class LogRetention(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory(prefix="sdp-run-logs-")
        self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name)
        self.log = self.root / "log with spaces.log"

    def rotate(self, language, keep="2", size="0"):
        if language == "python":
            core.logger.prepare_run_log(self.log, keep, size)
        else:
            result = subprocess.run([core.BASH, "-c", 'source "$HELPER"; sdp_prepare_run_log "$FILE"'],
                                    env=dict(os.environ, HELPER=str(core.BIN / "sdp_structured_log.sh"),
                                             FILE=str(self.log), SDP_RUN_LOG_KEEP=keep,
                                             SDP_RUN_LOG_MAX_BYTES=size), capture_output=True, text=True)
            self.assertEqual(result.returncode, 0, result.stderr)

    def test_rotation_retains_latest_two_runs_in_both_languages(self):
        for language in ["python", "bash"]:
            with self.subTest(language=language):
                for path in self.root.iterdir():
                    path.unlink()
                unrelated = self.log.with_name(self.log.name + ".run-unrelated")
                unrelated.write_text("leave this alone")
                for number in range(4):
                    self.log.write_text(str(number))
                    self.rotate(language)
                archives = [p for p in self.root.iterdir() if p != unrelated]
                self.assertEqual(sorted(p.read_text() for p in archives), ["2", "3"])
                self.assertEqual(unrelated.read_text(), "leave this alone")

    def test_size_budget_keeps_latest_failure_even_if_large(self):
        for language in ["python", "bash"]:
            with self.subTest(language=language):
                for path in self.root.iterdir():
                    path.unlink()
                self.log.write_text("old")
                self.rotate(language, "0", "5")
                self.log.write_text("large failed run diagnostics")
                self.rotate(language, "0", "5")
                archives = list(self.root.iterdir())
                self.assertEqual(len(archives), 1)
                self.assertEqual(archives[0].read_text(), "large failed run diagnostics")

    def test_future_dated_legacy_archive_never_displaces_current_failure(self):
        for language in ["python", "bash"]:
            with self.subTest(language=language):
                for path in self.root.iterdir():
                    path.unlink()
                legacy = self.log.with_name(self.log.name + ".run-20990101T000000Z-legacy")
                legacy.write_text("old diagnostics with future timestamp")
                os.utime(legacy, (4070908800, 4070908800))
                self.log.write_text("current failure")
                self.rotate(language, "1", "1")
                archives = list(self.root.iterdir())
                self.assertEqual(len(archives), 1)
                self.assertEqual(archives[0].read_text(), "current failure")

    def test_mixed_languages_keep_rotation_order_with_identical_timestamps(self):
        for number, language in enumerate(["python", "bash", "python", "bash", "python"]):
            # Emulate a coarse clock/filesystem and future-dated metadata.
            for path in self.root.iterdir():
                os.utime(path, (4070908800, 4070908800))
            self.log.write_text(str(number))
            self.rotate(language)
        self.assertEqual(sorted(p.read_text() for p in self.root.iterdir()), ["3", "4"])

    def test_invalid_limits_preserve_previous_log(self):
        self.log.write_text("previous failure")
        with self.assertRaises(ValueError):
            core.logger.prepare_run_log(self.log, "invalid")
        result = subprocess.run([core.BASH, "-c", 'source "$HELPER"; sdp_prepare_run_log "$FILE"'],
                                env=dict(os.environ, HELPER=str(core.BIN / "sdp_structured_log.sh"),
                                         FILE=str(self.log), SDP_RUN_LOG_KEEP="-1"), capture_output=True)
        self.assertEqual(result.returncode, 1)
        self.assertEqual(self.log.read_text(), "previous failure")

    def test_symlink_targets_are_not_modified(self):
        target = self.root / "valuable"
        target.write_text("evidence")
        self.log.symlink_to(target)
        with self.assertRaises(ValueError):
            core.logger.prepare_run_log(self.log)
        self.assertEqual(target.read_text(), "evidence")

    def test_overflow_limits_do_not_rotate_or_delete_logs(self):
        self.log.write_text("previous failure")
        huge = "18446744073709551620"
        with self.assertRaises(ValueError):
            core.logger.prepare_run_log(self.log, huge)
        result = subprocess.run([core.BASH, "-c", 'source "$HELPER"; sdp_prepare_run_log "$FILE"'],
                                env=dict(os.environ, HELPER=str(core.BIN / "sdp_structured_log.sh"),
                                         FILE=str(self.log), SDP_RUN_LOG_KEEP=huge), capture_output=True)
        self.assertEqual(result.returncode, 1)
        self.assertEqual(self.log.read_text(), "previous failure")

    def test_python_run_id_correlates_context_json_and_prom_comment(self):
        env = {"LOGS": str(self.root), "SDP_METRICS_DIR": str(self.root)}
        with patch.dict(os.environ, env):
            with core.logger.run("sample.py", mode="jsonl") as event:
                self.assertEqual(core.logger.current_run_id(), event.run_id)
                event.add(run_id="cannot override")
            self.assertEqual(json.loads((self.root / "sample.jsonl").read_text())["run_id"], event.run_id)
            prom = core.logger.SDPEvent("sample.py", mode="prom")
            prom.emit(0)
            content = (self.root / "p4_sdp_sample.prom").read_text()
            self.assertIn("# run_id " + prom.run_id, content)
            self.assertNotIn('run_id="', content)

    def test_shell_run_id_correlates_text_and_json(self):
        f = core.CoreReview(methodName="runTest")
        f.setUp()
        self.addCleanup(f.doCleanups)
        script = f.root / "correlate.sh"
        script.write_text('source "' + str(core.BIN / "sdp_structured_log.sh") + '"\nsdp_run_log_init correlate.sh\necho diagnostic\n')
        result = f.run_bash(script)
        self.assertEqual(result.returncode, 0, result.stderr)
        event = json.loads((f.logs / "correlate.jsonl").read_text())
        self.assertIn("run_id=" + event["run_id"], (f.logs / "correlate.log").read_text())

    def test_maintenance_preserves_failure_and_correlates_next_run(self):
        env = {"LOGS": str(self.root), "SDP_RUN_LOG_KEEP": "2"}
        log = self.root / "maintenance_sample.log"
        log.write_text("previous failure diagnostics")
        with patch.dict(os.environ, env), patch.dict(sys.modules, {"sdp_structured_log": core.logger}):
            with core.logger.run("maintenance_sample.py") as event:
                with maintenance.sdputils.maintenance_run("maintenance_sample.py", "1"):
                    print("current diagnostics")
        self.assertIn("run_id=" + event.run_id, log.read_text())
        self.assertEqual(next(self.root.glob(log.name + ".run-*")).read_text(), "previous failure diagnostics")
        self.assertEqual(json.loads((self.root / "maintenance_sample.jsonl").read_text())["run_id"], event.run_id)

    def test_verify_preserves_all_previous_diagnostics_before_work(self):
        spec = importlib.util.spec_from_file_location("improved_verify", core.BIN / "p4verify.py")
        module = importlib.util.module_from_spec(spec)
        sys.modules[spec.name] = module
        self.addCleanup(sys.modules.pop, spec.name, None)
        spec.loader.exec_module(module)
        module.sdplog = core.logger
        names = ["verify_output.log", "p4verify.log", "p4verify_errors.log", "p4verify-shelves.log", "p4verify_autotune.csv"]
        for name in names:
            (self.root / name).write_text("previous failure in " + name)
        verify = module.P4Verify.__new__(module.P4Verify)
        verify.instance = "1"
        verify.debug = False
        verify.L = None
        verify._sdp_event = core.logger.SDPEvent("p4verify.py")
        verify.load_sdp_environment = lambda: None
        class StopBeforeExternalWork(Exception):
            pass
        def stop():
            raise StopBeforeExternalWork()
        verify.check_backup_functions_present = stop
        with patch.dict(os.environ, {"LOGS": str(self.root)}):
            with self.assertRaises(StopBeforeExternalWork):
                verify.run()
        for name in names:
            archives = list(self.root.glob(name + ".run-*"))
            self.assertEqual(len(archives), 1, name)
            self.assertEqual(archives[0].read_text(), "previous failure in " + name)

    def test_competing_verifier_does_not_rotate_active_logs_or_start_work(self):
        spec = importlib.util.spec_from_file_location("locked_verify", core.BIN / "p4verify.py")
        module = importlib.util.module_from_spec(spec)
        sys.modules[spec.name] = module
        self.addCleanup(sys.modules.pop, spec.name, None)
        spec.loader.exec_module(module)
        verify = module.P4Verify.__new__(module.P4Verify)
        verify.instance, verify.L = "1", None
        verify._sdp_event = core.logger.SDPEvent("p4verify.py")
        verify.load_sdp_environment = lambda: None
        active = self.root / "verify_output.log"
        active.write_text("running verification diagnostics")
        with (self.root / "p4verify.lock").open("a") as lock, patch.dict(os.environ, {"LOGS": str(self.root)}):
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
            verify._run_locked = lambda _: self.fail("Competing invocation started work")
            verify.run()
            self.assertEqual(verify._sdp_event.status, "skipped")
        self.assertEqual(active.read_text(), "running verification diagnostics")
        self.assertEqual(list(self.root.glob("*.run-*")), [])
        with patch.dict(os.environ, {"LOGS": str(self.root)}):
            verify._run_locked = lambda _: None
            verify.run()  # Persistent filename does not prevent a later run.


class BoundedCacheCleanup(unittest.TestCase):
    def fixture(self):
        suite = runtime.CacheCleanupReview(methodName="runTest")
        suite.setUp()
        self.addCleanup(suite.doCleanups)
        f = suite.fixture
        for n in range(4):
            (f.home / ("depots/depot " + str(n))).mkdir()
        f.write_executable(f.mockbin / "find", "#!" + sys.executable + r'''
import fcntl, json, os, sys, time
from pathlib import Path
root = Path(os.environ["SDP_TEST_ROOT"])
if sys.argv[1] == str(root / "metrics"):
    sys.exit(0)
state = root / "workers.json"
with (root / "workers.lock").open("a") as lock:
    fcntl.flock(lock, fcntl.LOCK_EX)
    data = json.loads(state.read_text()) if state.exists() else {"active": 0, "maximum": 0, "done": 0}
    data["active"] += 1
    data["maximum"] = max(data["active"], data["maximum"])
    state.write_text(json.dumps(data))
time.sleep(0.15)
with (root / "workers.lock").open("a") as lock:
    fcntl.flock(lock, fcntl.LOCK_EX)
    data = json.loads(state.read_text())
    data["active"] -= 1
    data["done"] += 1
    state.write_text(json.dumps(data))
sys.stdout.write("..")
''')
        return f

    def test_worker_limit_is_obeyed_and_all_counts_are_collected(self):
        f = self.fixture()
        result = f.run_script("cacheclean.sh", "1", extra={"SDP_CACHE_CLEAN_WORKERS": "2"})
        self.assertEqual(result.returncode, 0, result.stderr + f.log("cacheclean"))
        state = json.loads((f.root / "workers.json").read_text())
        self.assertEqual(state, {"active": 0, "maximum": 2, "done": 5})
        self.assertEqual(f.event("cacheclean")["files_removed"], 10)
        self.assertEqual(f.event("cacheclean")["depot_count"], 5)
        self.assertIn("seconds", f.log("cacheclean"))

    def test_invalid_worker_limit_runs_no_deletions(self):
        f = self.fixture()
        for limit in ["0", "257", "18446744073709551620", "invalid"]:
            with self.subTest(limit=limit):
                result = f.run_script("cacheclean.sh", "1", extra={"SDP_CACHE_CLEAN_WORKERS": limit})
                self.assertEqual(result.returncode, 1)
                self.assertFalse((f.root / "workers.json").exists())


if __name__ == "__main__":
    unittest.main()