test_maintenance_logging.py #2

#!/usr/bin/env python3
"""Isolated maintenance logging regressions; no installed /p4 or live server.

Run with SDP_TEST_BASH=/path/to/bash python3 -m unittest discover -s sdp/test
-p test_maintenance_logging.py. Requires Bash 4.3+ like the deployed logger.
All absolute /p4 paths are rewritten in temporary copies; p4/rsync are mocks.
"""
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import time
import unittest

BIN = Path(__file__).resolve().parents[1] / "Server/Unix/p4/common/bin"
BASH = os.environ.get("SDP_TEST_BASH", shutil.which("bash"))


class MaintenanceLogging(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory(prefix="sdp-logging-")
        self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name)
        self.p4 = self.root / "p4"
        self.common = self.p4 / "common/bin"
        self.home = self.p4 / "1"
        self.logs = self.home / "logs"
        self.mockbin = self.root / "mockbin"
        for folder in [self.common, self.logs, self.home / "root", self.home / "bin",
                       self.home / "tmp", self.home / "cache/depot", self.mockbin,
                       self.p4 / "common/config", self.root / "metrics"]:
            folder.mkdir(parents=True, exist_ok=True)
        for source in BIN.iterdir():
            if source.is_file() and (source.suffix in (".sh", ".py") or source.name == "p4login_master"):
                target = self.common / source.name
                target.write_text(source.read_text().replace("/p4/", str(self.p4) + "/"))
                target.chmod(0o755)
        self.env = {"PATH": str(self.mockbin) + ":" + str(Path(BASH).parent) + ":/usr/bin:/bin",
                    "HOME": str(self.root), "HOSTNAME": "test-host", "LC_ALL": "C",
                    "SDP_TEST_ROOT": str(self.root), "SDP_LOG_FORMAT": "jsonl",
                    "SDP_METRICS_DIR": str(self.root / "metrics"),
                    "PYTHONDONTWRITEBYTECODE": "1"}
        self.write_executable(self.common / "p4_vars", r"""#!/bin/bash
export SDP_INSTANCE=$1
export P4HOME="$SDP_TEST_ROOT/p4/$1"
export LOGS="$P4HOME/logs" P4ROOT="$P4HOME/root" P4TMP="$P4HOME/tmp"
export P4CBIN="$SDP_TEST_ROOT/p4/common/bin" P4BIN="$P4HOME/bin/p4_$1"
export P4PCACHE="$P4HOME/cache" P4PORT=test:1666 P4USER=test-user P4SERVER=p4_1
export P4MASTERPORT=master.example:1666 RSYNCUSER=test COMMITSERVER=commit P4MASTERHOST=master
export P4MASTER_ID=master-id
""")
        self.write_executable(self.home / "bin/p4_1", "#!" + sys.executable + r'''
import json, os, sys
from pathlib import Path
args = sys.argv[1:]
root = Path(os.environ["SDP_TEST_ROOT"])
with (root / "calls.jsonl").open("a") as out:
    out.write(json.dumps(args) + "\n")
fail = os.environ.get("MOCK_FAIL", "")
if fail and fail in args:
    print("mock failure: " + fail, file=sys.stderr)
    sys.exit(7)
if "monitor" in args:
    if "show" in args:
        print(os.environ.get("MOCK_MONITOR", ""))
elif "info" in args:
    print("info: Server test:1666")
elif "changes" in args:
    print("info: Change 1 on 2026/10/16 by user@test")
elif "files" in args:
    print("//depot/a file.txt")
elif "fstat" in args:
    print(str(root / "archive file,d/1.1"))
elif "users" in args:
    print("user1\nuser2")
elif "group" in args:
    if "-o" in args:
        print("Group: limits\nUsers:")
    else:
        sys.stdin.read()
        print("Group saved.")
elif "server" in args:
    if "-o" in args:
        print("Services:\tstandard")
    else:
        sys.stdin.read()
        print("Server saved.")
elif "user" in args:
    print("Type:\tservice")
elif "protect" in args:
    print("super group service.g * //...")
elif "login" in args:
    print("3600" if "-s" in args else "User logged in.")
else:
    print("mock p4 command completed")
''')
        self.write_executable(self.common / "p4login", "#!" + BASH + '\necho "Login completed"\n')
        self.write_executable(self.common / "p4master_run", "#!" + BASH + '\nshift\nexec "$@"\n')
        # macOS lacks flock(1); use the same advisory lock through Python.
        self.write_executable(self.mockbin / "flock", "#!" + sys.executable +
                              '\nimport fcntl, sys\nfcntl.flock(int(sys.argv[-1]), fcntl.LOCK_EX)\n')
        self.write_executable(self.mockbin / "rsync", "#!" + BASH + r'''
echo "rsync output"
if [[ ${MOCK_RSYNC_FAIL:-0} == 1 ]]; then
    echo "mock rsync failure" >&2
    exit 23
fi
''')
        # Translate GNU find's successful-deletion counter for host macOS tests.
        self.write_executable(self.mockbin / "find", "#!" + sys.executable + r'''
import subprocess, sys
args = sys.argv[1:]
if len(args) >= 2 and args[-2:] == ["-printf", "."]:
    result = subprocess.run(["/usr/bin/find", *args[:-2], "-print0"], stdout=subprocess.PIPE)
    sys.stdout.buffer.write(b"." * result.stdout.count(b"\0"))
    sys.exit(result.returncode)
subprocess.run(["/usr/bin/find", *args], check=True)
''')
        # GNU sed's append syntax in the existing mk*.sh scripts differs on macOS.
        # Pass-through suffices here: these tests verify execution/logging, not
        # the existing server-spec transformation.
        self.write_executable(self.mockbin / "sed", "#!" + BASH + r'''
if [[ ${1:-} == '$ a ReplicatingFrom:'* ]]; then cat; else /usr/bin/sed "$@"; fi
''')

    @staticmethod
    def write_executable(path, content):
        path.write_text(content)
        path.chmod(0o755)

    def run_script(self, name, *args, extra=None):
        env = dict(self.env)
        env.update(extra or {})
        interpreter = sys.executable if name.endswith(".py") else BASH
        return subprocess.run([interpreter, str(self.common / name), *args],
                              env=env, cwd=self.root, capture_output=True, text=True, timeout=15)

    def event(self, name):
        records = (self.logs / (name + ".jsonl")).read_text().splitlines()
        self.assertEqual(len(records), 1, records)
        return json.loads(records[0])

    def log(self, name):
        return (self.logs / (name + ".log")).read_text()

    def test_sanity_initializes_logs_in_clean_environment(self):
        result = self.run_script("p4sanity_check.sh", "1")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(result.stdout, "")
        self.assertIn("PASS: Sanity checks", self.log("p4sanity_check"))
        self.assertEqual(self.event("p4sanity_check")["status"], "success")
        self.assertFalse(list(self.logs.glob("tmp.*.log")))

    def test_sanity_preserves_failed_command_diagnostics(self):
        result = self.run_script("p4sanity_check.sh", "1", extra={"MOCK_FAIL": "info"})
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("mock failure: info", self.log("p4sanity_check"))
        self.assertEqual(self.event("p4sanity_check")["status"], "failure")

    def test_sanity_rejects_failed_changes_even_with_valid_output(self):
        p4 = self.home / "bin/p4_1"
        p4.write_text(p4.read_text().replace(
            'print("info: Change 1 on 2026/10/16 by user@test")',
            'print("info: Change 1 on 2026/10/16 by user@test"); sys.exit(7)'))
        result = self.run_script("p4sanity_check.sh", "1")
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("Could not list recent changelists", self.log("p4sanity_check"))
        self.assertEqual(self.event("p4sanity_check")["status"], "failure")

    def test_proxy_skip_is_logged(self):
        (self.home / "bin/skip_cache_clean.txt").touch()
        result = self.run_script("proxyclean.sh", "1")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn("skipping", self.log("proxyclean"))
        self.assertEqual(self.event("proxyclean")["status"], "skipped")

    def test_proxy_waits_and_cleans_only_old_files(self):
        old = self.home / "cache/depot/old file"
        recent = old.with_name("recent")
        old.write_text("old")
        recent.write_text("new")
        age = time.time() - 20 * 86400
        os.utime(old, (age, age))
        monitor = self.home / "cache/pdb.monitor"
        monitor.write_text("monitor")
        os.utime(monitor, (age, age))
        result = self.run_script("proxyclean.sh", "1")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertFalse(old.exists())
        self.assertTrue(recent.exists())
        self.assertTrue(monitor.exists())
        self.assertIn("removed 1 file(s)", self.log("proxyclean"))
        self.assertEqual(self.event("proxyclean")["status"], "success")

    def test_proxy_worker_failure_is_logged(self):
        self.write_executable(self.mockbin / "find", "#!" + BASH + '\nsleep 0.1\necho "worker failed" >&2\nexit 8\n')
        result = self.run_script("proxyclean.sh", "1")
        self.assertEqual(result.returncode, 1)
        self.assertIn("worker failed", self.log("proxyclean"))
        self.assertEqual(self.event("proxyclean")["status"], "failure")

    def test_proxy_honors_configured_age(self):
        (self.home / "bin/proxy_clean_atime.txt").write_text("30")
        old = self.home / "cache/depot/20-day-old"
        old.write_text("preserve")
        age = time.time() - 20 * 86400
        os.utime(old, (age, age))
        result = self.run_script("proxyclean.sh", "1")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertTrue(old.exists())
        self.assertEqual(self.event("proxyclean")["atime_days"], 30)

    def test_idle_no_server_is_logged(self):
        result = self.run_script("kill_idle.sh", "1")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn("No server.pid", self.log("kill_idle"))
        self.assertEqual(self.event("kill_idle")["status"], "skipped")

    def test_idle_no_targets_does_not_invoke_terminate(self):
        (self.home / "root/server.pid").write_text("999999")
        result = self.run_script("kill_idle.sh", "1")
        self.assertEqual(result.returncode, 0, result.stderr)
        calls = [json.loads(x) for x in (self.root / "calls.jsonl").read_text().splitlines()]
        self.assertEqual(calls, [["monitor", "show", "-ale"]] * 4)
        self.assertEqual(self.event("kill_idle")["status"], "success")

    def test_idle_query_failure_is_logged(self):
        (self.home / "root/server.pid").write_text("999999")
        result = self.run_script("kill_idle.sh", "1", extra={"MOCK_FAIL": "monitor"})
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("mock failure: monitor", self.log("kill_idle"))
        self.assertEqual(self.event("kill_idle")["status"], "failure")

    def test_idle_termination_failure_is_not_masked(self):
        (self.home / "root/server.pid").write_text("999999")
        # Six-minute idle process qualifies for p4 terminate, but not OS kill.
        result = self.run_script("kill_idle.sh", "1", extra={
            "MOCK_MONITOR": "999998 user IDLE 00:06:30 command",
            "MOCK_FAIL": "terminate"})
        self.assertEqual(result.returncode, 1, result.stderr)
        self.assertIn("mock failure: terminate", self.log("kill_idle"))
        self.assertEqual(self.event("kill_idle")["status"], "failure")

    def test_repair_transfer_failure_survives_cleanup(self):
        (self.logs / "verify_output.log").write_text("//depot/file#1 - MISSING!\n")
        result = self.run_script("rsync_missing.sh", "1", extra={"MOCK_RSYNC_FAIL": "1"})
        self.assertEqual(result.returncode, 23, result.stderr)
        self.assertIn("mock rsync failure", self.log("rsync_missing"))
        self.assertFalse((self.home / "tmp/temp_rsync.sh").exists())
        self.assertEqual(self.event("rsync_missing")["status"], "failure")

    def test_repair_no_missing_files_is_success(self):
        (self.logs / "verify_output.log").write_text("All revisions verified.\n")
        result = self.run_script("rsync_missing.sh", "1")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(self.event("rsync_missing")["status"], "success")

    def test_shelf_copy_initializes_environment_and_logs(self):
        result = self.run_script("copy_missing_shelf_files.sh", "123", "1")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn("archive file", self.log("copy_missing_shelf_files"))
        self.assertEqual(self.event("copy_missing_shelf_files")["status"], "success")

    def test_shelf_copy_logs_transfer_failure(self):
        result = self.run_script("copy_missing_shelf_files.sh", "123", "1", extra={"MOCK_RSYNC_FAIL": "1"})
        self.assertNotEqual(result.returncode, 0)
        self.assertIn("mock rsync failure", self.log("copy_missing_shelf_files"))
        self.assertEqual(self.event("copy_missing_shelf_files")["status"], "failure")

    def test_creation_helpers_fail_on_early_configuration_errors(self):
        for name in ["mkedge", "mkedge-standby", "mkstandby", "mkstandby_shared"]:
            with self.subTest(name=name):
                result = self.run_script(name + ".sh", "1", "replica", "test-password-not-for-log", "master:1666", "upstream", extra={"MOCK_FAIL": "configure"})
                self.assertNotEqual(result.returncode, 0)
                log = self.log(name)
                self.assertIn("mock failure: configure", log)
                self.assertNotIn("test-password-not-for-log", log)
                self.assertEqual(self.event(name)["status"], "failure")

    def test_creation_helpers_do_not_write_after_failed_spec_reads(self):
        for name in ["mkedge", "mkedge-standby", "mkstandby", "mkstandby_shared"]:
            with self.subTest(name=name):
                result = self.run_script(name + ".sh", "1", "replica", "test-password-not-for-log",
                                         "master:1666", "upstream", extra={"MOCK_FAIL": "server"})
                self.assertEqual(result.returncode, 7)
                self.assertEqual(self.event(name)["status"], "failure")
        calls = [json.loads(line) for line in (self.root / "calls.jsonl").read_text().splitlines()]
        self.assertTrue(all(call == ["server", "-o", "replica"] for call in calls), calls)

    def test_missing_transfer_stops_before_later_success_can_mask_failure(self):
        (self.logs / "verify_output.log").write_text(
            "info: //depot/first#1 - MISSING!\ninfo: //depot/second#1 - MISSING!\n")
        self.write_executable(self.mockbin / "rsync", "#!" + sys.executable + '''
import os, sys
from pathlib import Path
p = Path(os.environ["SDP_TEST_ROOT"]) / "transfer-count"
count = int(p.read_text()) + 1 if p.exists() else 1
p.write_text(str(count))
sys.exit(23 if count == 1 else 0)
''')
        result = self.run_script("rsync_missing.sh", "1")
        self.assertEqual(result.returncode, 23)
        self.assertEqual((self.root / "transfer-count").read_text(), "1")

    def prepare_rsync_transport_probe(self):
        """Observe child rsync environment/arguments without making a connection."""
        self.write_executable(self.mockbin / "rsync", "#!" + sys.executable + r"""
import json, os, sys
from pathlib import Path
path = Path(os.environ["SDP_TEST_ROOT"]) / "rsync-probe.jsonl"
prior = [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else []
source = sys.argv[-2]
first_for_source = not any(row["args"][-2] == source for row in prior)
failed = (os.environ.get("MOCK_RSYNC_FAIL_ALWAYS") == "1" or
          (os.environ.get("MOCK_RSYNC_FAIL_FIRST_PER_SOURCE") == "1" and first_for_source))
with path.open("a") as output:
    output.write(json.dumps({"args": sys.argv[1:], "rsh": os.environ.get("RSYNC_RSH"),
                             "failed": failed}) + "\n")
if failed:
    print("mock SSH transport failed for " + source, file=sys.stderr)
    sys.exit(23)
print("mock transfer complete")
""")
        # The replica entrypoints execute their real transfer/retry sections,
        # then stop at the first recovery helper. No service/database operation
        # is permitted to run in this fixture.
        self.write_executable(self.common / "backup_functions.sh", r"""
check_vars () { :; }
set_vars () {
    CHECKPOINTS="$P4HOME/checkpoints with spaces"
    JOURNALS="$P4HOME/journals with spaces"
    P4REPLICA=TRUE
    SHAREDDATA=FALSE
}
get_journalnum () { JOURNALNUM=100; }
check_offline_db_usable () { :; }
ckp_running () { :; }
rotate_last_run_logs () { :; }
check_uid () { :; }
log_init () { :; }
sdp_log_init () { :; }
log () { printf '%s\n' "$*" >> "$LOGFILE"; }
die () { log "$*"; exit 1; }
recreate_offline_db_files () {
    echo 'test stopped before recovery' > "$SDP_TEST_ROOT/recovery-sentinel"
    exit 73
}
""")
        (self.logs / "verify_output.log").write_text("info: //depot/file#1 - MISSING!\n")

    def run_transport_probe(self, script, extra=None):
        probe = self.root / "rsync-probe.jsonl"
        sentinel = self.root / "recovery-sentinel"
        for path in [probe, sentinel]:
            if path.exists():
                path.unlink()
        args = ("123", "1") if script == "copy_missing_shelf_files.sh" else ("1",)
        result = self.run_script(script, *args, extra=extra)
        rows = [json.loads(line) for line in probe.read_text().splitlines()] if probe.exists() else []
        return result, rows

    def test_remote_rsync_default_is_exported_to_all_four_scripts(self):
        self.prepare_rsync_transport_probe()
        for script in ["sync_replica.sh", "recreate_db_sync_replica.sh",
                       "copy_missing_shelf_files.sh", "rsync_missing.sh"]:
            for environment in [{}, {"RSYNC_RSH": ""}]:
                with self.subTest(script=script, environment=environment):
                    result, rows = self.run_transport_probe(script, environment)
                    replica = script in ["sync_replica.sh", "recreate_db_sync_replica.sh"]
                    self.assertEqual(result.returncode, 73 if replica else 0, result.stderr)
                    self.assertEqual(len(rows), 2 if replica else 1)
                    self.assertTrue(all(row["rsh"] == "ssh -o StrictHostKeyChecking=accept-new"
                                        for row in rows), rows)
                    if script == "copy_missing_shelf_files.sh":
                        self.assertEqual(rows[0]["args"][-2:], [
                            str(self.root / "archive file,d/1.1"),
                            "test@commit:" + str(self.root / "archive file,d/1.1")])
                    if replica:
                        self.assertTrue((self.root / "recovery-sentinel").exists())

    def test_remote_rsync_preserves_explicit_transport_for_all_four_scripts(self):
        self.prepare_rsync_transport_probe()
        custom = 'ssh -p 2222 -i "/keys/key with spaces" -o StrictHostKeyChecking=yes'
        for script in ["sync_replica.sh", "recreate_db_sync_replica.sh",
                       "copy_missing_shelf_files.sh", "rsync_missing.sh"]:
            with self.subTest(script=script):
                result, rows = self.run_transport_probe(script, {"RSYNC_RSH": custom})
                replica = script in ["sync_replica.sh", "recreate_db_sync_replica.sh"]
                self.assertEqual(result.returncode, 73 if replica else 0, result.stderr)
                self.assertEqual(len(rows), 2 if replica else 1)
                self.assertTrue(all(row["rsh"] == custom for row in rows), rows)

    def test_replica_checkpoint_and_journal_retries_keep_transport_and_paths(self):
        self.prepare_rsync_transport_probe()
        for script in ["sync_replica.sh", "recreate_db_sync_replica.sh"]:
            for custom in [None, "ssh -p 2222 -o StrictHostKeyChecking=accept-new"]:
                with self.subTest(script=script, custom=custom):
                    extra = {"MOCK_RSYNC_FAIL_FIRST_PER_SOURCE": "1"}
                    if custom is not None:
                        extra["RSYNC_RSH"] = custom
                    result, rows = self.run_transport_probe(script, extra)
                    self.assertEqual(result.returncode, 73, result.stderr)
                    self.assertEqual(len(rows), 4, rows)
                    self.assertEqual([row["failed"] for row in rows], [True, False, True, False])
                    self.assertTrue(all(row["rsh"] == (custom or "ssh -o StrictHostKeyChecking=accept-new")
                                        for row in rows), rows)
                    for row, folder in zip(rows, ["checkpoints with spaces"] * 2 + ["journals with spaces"] * 2):
                        target = str(self.home / folder)
                        self.assertEqual(row["args"], ["-avz", "--delete", "test@master:" + target + "/", target])
                    self.assertTrue((self.root / "recovery-sentinel").exists())
                    self.assertIn("mock SSH transport failed", self.log("sync_replica"))

    def test_replica_failed_retry_stops_before_recovery_and_logs_error(self):
        self.prepare_rsync_transport_probe()
        for script in ["sync_replica.sh", "recreate_db_sync_replica.sh"]:
            with self.subTest(script=script):
                result, rows = self.run_transport_probe(script, {"MOCK_RSYNC_FAIL_ALWAYS": "1"})
                self.assertEqual(result.returncode, 1)
                self.assertEqual(len(rows), 2, rows)
                self.assertTrue(all(row["failed"] for row in rows))
                self.assertTrue(all(row["rsh"] == "ssh -o StrictHostKeyChecking=accept-new"
                                    for row in rows), rows)
                self.assertFalse((self.root / "recovery-sentinel").exists())
                self.assertIn("mock SSH transport failed", self.log("sync_replica"))
                self.assertIn("exit code was: 23", self.log("sync_replica"))

    def test_master_login_initializes_environment(self):
        (self.p4 / "common/config/.p4passwd.p4_1.admin").write_text("test-only\n")
        result = self.run_script("p4login_master", "1")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn("User logged in", self.log("p4login"))
        self.assertEqual(self.event("p4login")["status"], "success")

    def test_creation_helpers_complete_without_logging_passwords(self):
        for name in ["mkedge", "mkedge-standby", "mkstandby", "mkstandby_shared"]:
            with self.subTest(name=name):
                result = self.run_script(name + ".sh", "1", "replica", "test-password-not-for-log", "master:1666", "upstream")
                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertNotIn("test-password-not-for-log", self.log(name))
                self.assertEqual(self.event(name)["status"], "success")

    def test_limits_success_is_logged_without_inherited_environment(self):
        result = self.run_script("update_limits.py", "1")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertIn("1 groups updated, 2 users", self.log("update_limits"))
        self.assertEqual(self.event("update_limits")["status"], "success")

    def test_limits_command_failure_is_preserved(self):
        result = self.run_script("update_limits.py", "1", extra={"MOCK_FAIL": "group"})
        self.assertEqual(result.returncode, 1, result.stderr)
        self.assertIn("mock failure: group", self.log("update_limits"))
        self.assertEqual(self.event("update_limits")["status"], "failure")

    def test_logging_modes_preserve_text_log(self):
        (self.home / "bin/skip_cache_clean.txt").touch()
        for mode in ["off", "prom"]:
            result = self.run_script("proxyclean.sh", "1", extra={"SDP_LOG_FORMAT": mode})
            self.assertEqual(result.returncode, 0, result.stderr)
            self.assertIn("skipping", self.log("proxyclean"))
            self.assertFalse((self.logs / "proxyclean.jsonl").exists())
        self.assertTrue(list((self.root / "metrics").glob("*.prom")))

    def test_unwritable_log_stops_before_cleanup(self):
        (self.logs / "proxyclean.log").mkdir()
        old = self.home / "cache/depot/old"
        old.write_text("preserve")
        age = time.time() - 20 * 86400
        os.utime(old, (age, age))
        result = self.run_script("proxyclean.sh", "1")
        self.assertNotEqual(result.returncode, 0)
        self.assertTrue(old.exists())

    def test_existing_logger_keeps_stdout_behavior(self):
        script = self.common / "legacy.sh"
        script.write_text('source "' + str(self.common / "p4_vars") + '" 1\n'
                          'source "' + str(self.common / "sdp_structured_log.sh") + '"\n'
                          'sdp_log_init legacy.sh\necho original-output\n')
        result = self.run_script("legacy.sh")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(result.stdout, "original-output\n")
        self.assertFalse((self.logs / "legacy.log").exists())
        self.assertEqual(self.event("legacy")["status"], "success")


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