#!/usr/bin/env python3 """Exercise Maintenance against a guarded, disposable local p4d. Creates uniquely named fixtures and keeps their submitted files for inspection. Never sends mail or contacts a cloud, directory, or external authentication service. The inherited guard requires the disposable marker, matching account, loopback endpoint, server ID and disabled mail before any Perforce mutation. """ import argparse from contextlib import contextmanager, redirect_stdout, redirect_stderr import importlib.util import io import json import os from pathlib import Path import subprocess import sys import tempfile import time import uuid from unittest.mock import patch from runtime_integration import RuntimeIntegration class MaintenanceIntegration(RuntimeIntegration): def __init__(self, args): super().__init__(args) from P4 import P4 self.source = Path("/p4/sdp/Maintenance") self.prefix = "sdp_m_" + uuid.uuid4().hex[:12] self.work = Path(tempfile.mkdtemp(prefix=self.prefix + "-", dir=self.home / "tmp")) self.env["SDP_INSTANCE"] = args.instance self.env["P4CONFIG"] = ".sdp-no-test-config" self.env["PATH"] = str(Path(self.env["P4BIN"]).parent) + ":" + self.env["PATH"] self.api = P4() self.api.port = self.env["P4PORT"] self.api.user = self.env["P4USER"] self.api.connect() self.api.password = Path("/p4/common/config/.p4passwd.p4_" + args.instance + ".admin").read_text().strip() self.api.run_login() self.users = [self.prefix + "_u" + str(index) for index in range(3)] self.groups = [self.prefix + "_g" + str(index) for index in range(2)] self.depot = self.prefix self.client = self.prefix + "_work" def scenario(self, name, scripts, detail, operation): # P4Exception is not a RuntimeError; include API failures in the JSON. started = time.monotonic() row = {"name": name, "scripts": scripts, "detail": detail, "kind": "live"} try: operation() row["status"] = "passed" except Exception as error: row.update(status="failed", error=str(error)[-6000:]) row["duration_seconds"] = round(time.monotonic() - started, 3) self.records.append(row) def check(self, condition, message): if not condition: raise RuntimeError(message) def script(self, name, *args, check=True): argv = [sys.executable, self.source / name, *map(str, args)] if name.endswith(".sh"): argv[0] = "bash" result = subprocess.run(argv, cwd=self.work, env=self.env, capture_output=True, text=True, timeout=120) (self.work / (name + ".output")).write_text(result.stdout + result.stderr) if check and result.returncode: raise RuntimeError(name + " failed (%d): " % result.returncode + (result.stdout + result.stderr)[-5000:]) return result @contextmanager def python_script(self, name): """Load installed code with the same guarded environment as CLI calls.""" previous = Path.cwd() sys.path.insert(0, str(self.source)) try: with patch.dict(os.environ, self.env, clear=True), patch.object(sys, "argv", [name, self.args.instance]): os.chdir(self.work) spec = importlib.util.spec_from_file_location("maintenance_live_" + name.removesuffix(".py"), self.source / name) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) yield module finally: os.chdir(previous) sys.path.remove(str(self.source)) def ensure_unload(self): if not any(row["type"] == "unload" for row in self.api.run_depots()): spec = self.api.fetch_depot(self.prefix + "_unload") spec["Type"] = "unload" self.api.save_depot(spec) def fixture_files(self): spec = self.api.fetch_depot(self.depot) self.api.save_depot(spec) spec = self.api.fetch_client(self.client) spec["Root"] = str(self.work / "workspace") spec["View"] = ["//%s/... //%s/..." % (self.depot, self.client)] self.api.save_client(spec) self.api.client = self.client self.env["P4CLIENT"] = self.client workspace = Path(spec["Root"]) for name in ("root.txt", "space dir/a file.txt", "deleted only/old.txt"): path = workspace / name path.parent.mkdir(parents=True, exist_ok=True) path.write_text("disposable maintenance fixture\n") self.api.run_add(str(path)) result = self.api.run_submit("-d", "Disposable maintenance integration fixture") self.change = next(str(row["submittedChange"]) for row in result if isinstance(row, dict) and "submittedChange" in row) self.api.run_delete(str(workspace / "deleted only/old.txt")) self.api.run_submit("-d", "Retain a deleted-only directory in inventory") def users_groups(self): (self.work / "users.csv").write_text("".join( "%s,%s@example.invalid,Disposable User %d\n" % (user, user, index) for index, user in enumerate(self.users))) self.script("createusers.py", "users.csv", self.args.instance) for user in self.users: self.check(self.api.fetch_user(user)["Email"] == user + "@example.invalid", "Bulk user creation lost its email field") (self.work / "groups.txt").write_text("group,%s\n%s\ngroup,%s\n%s\n" % (self.groups[0], self.users[0], self.groups[1], self.users[0])) self.script("creategroups.py", self.args.instance) group_spec = self.api.fetch_group(self.groups[0]) group_spec["Owners"] = [self.env["P4USER"]] group_spec["Subgroups"] = [self.groups[1]] group_spec["MaxResults"] = "100000" self.api.save_group(group_spec) protected_fields = {key: group_spec[key] for key in ("Owners", "Subgroups", "MaxResults")} self.script("addusertogroup.py", self.args.instance, self.users[1], self.groups[0]) self.check(set(self.api.fetch_group(self.groups[0])["Users"]) == set(self.users[:2]), "Adding a member replaced the existing group member") self.script("removeusersfromgroup.py", self.args.instance, self.users[1], self.groups[0]) self.check(self.api.fetch_group(self.groups[0])["Users"] == self.users[:1], "Single-group removal did not preserve the other member") # Keep one member so saving the groups does not delete empty groups. for group in self.groups: spec = self.api.fetch_group(group) spec["Users"] = self.users[:2] self.api.save_group(spec) (self.work / "selected-groups.txt").write_text("\n".join(self.groups) + "\n") self.script("removeuserfromgroups_file.py", self.args.instance, self.users[0], "selected-groups.txt") for group in self.groups: self.check(self.api.fetch_group(group)["Users"] == self.users[1:2], "Group-list removal lost an unrelated member") self.check(not list(self.work.glob(".sdp-group-removal-*")), "Successful group edits left recovery scratch files") edited = self.api.fetch_group(self.groups[0]) self.check(all(edited[key] == value for key, value in protected_fields.items()), "Membership edits changed owners, subgroups or resource limits") def user_reports(self): ghost = self.prefix + "_missing" spec = self.api.fetch_group(self.groups[0]) spec["Users"] = [self.users[1], ghost] self.api.save_group(spec) self.script("checkusers.py", self.args.instance) missing = (self.work / "removeusersfromgroups.txt").read_text().splitlines() self.check(ghost in missing and self.users[1] not in missing, "Missing-user report failed to distinguish real and stale group members") result = self.script("checkusers_not_in_group.py", self.args.instance) reported = result.stdout.splitlines() self.check(self.users[2] in reported and self.users[1] not in reported, "Ungrouped-user report disagrees with server memberships") self.script("make_email_list.py", self.args.instance) emails = (self.work / "emaillist.txt").read_text().splitlines() self.check(all(user + "@example.invalid" in emails for user in self.users), "Email-address export omitted fixture users") def inventory(self): self.script("create_p4_filelist.sh", self.args.instance) paths = (self.work / "p4_file_list.txt").read_text().splitlines() expected = {"//" + self.depot + "/" + name for name in ("root.txt", "space dir/a file.txt", "deleted only/old.txt")} actual = [path for path in paths if path.startswith("//" + self.depot + "/")] self.check(set(actual) == expected and len(actual) == len(expected), "Inventory omitted/duplicated a root file, spaced path or deleted-only directory") def labels(self): self.ensure_unload() label = self.prefix + "_label" spec = self.api.fetch_label(label) spec["View"] = ["//" + self.depot + "/..."] spec["Options"] = "unlocked noautoreload" spec["Description"] = "Preserve this label description\nincluding the second line.\n" self.api.save_label(spec) self.api.run_labelsync("-l", label, "//" + self.depot + "/...") spec = self.api.fetch_label(label) spec["Owner"] = self.users[1] self.api.save_label(spec, "-f") self.script("isitalabel.py", label) before = {(row["depotFile"], row["rev"]) for row in self.api.run_files("@" + label)} self.script("p4unlock.py", label) self.check((self.work / "owner.txt").read_text() == self.users[1], "Unlock did not retain the original label owner") self.script("p4lock.py", label) locked = self.api.fetch_label(label) self.check(locked["Owner"] == self.users[1] and "locked" in locked["Options"].split(), "Lock did not restore the owner and lock option") self.check(not (self.work / "owner.txt").exists(), "Successful relock left owner.txt") self.script("convert_label_to_autoreload.py", label) converted = self.api.fetch_label(label) after = {(row["depotFile"], row["rev"]) for row in self.api.run_files("@" + label)} self.check("autoreload" in converted["Options"].split() and before == after, "Autoreload conversion changed the label revision set") self.check(converted["Owner"] == self.users[1], "Conversion changed label ownership") self.check(converted["Description"] == locked["Description"] and converted["View"] == locked["View"], "Conversion changed the label description or view") self.check(not list(self.work.glob(".sdp-label-recovery-*")), "Successful conversion left recovery scratch files") def label_failure_recovery(self): self.ensure_unload() label = self.prefix + "_recover_label" spec = self.api.fetch_label(label) spec["View"] = ["//" + self.depot + "/..."] spec["Description"] = "Recover this tagged label after an interrupted conversion.\n" spec["Options"] = "unlocked noautoreload" self.api.save_label(spec) self.api.run_labelsync("-l", label, "//" + self.depot + "/...") spec = self.api.fetch_label(label) spec["Owner"] = self.users[1] spec["Options"] = "locked noautoreload" self.api.save_label(spec, "-f") expected = {(row["depotFile"], row["rev"]) for row in self.api.run_files("@" + label)} prior = set(self.work.glob(".sdp-label-recovery-*")) output = io.StringIO() with self.python_script("convert_label_to_autoreload.py") as module: original = module.sdputils.run_checked def fail_labelsync(argv, **kwargs): if argv[1] == "-x" and "labelsync" in argv: raise subprocess.CalledProcessError(7, argv, stderr="injected labelsync failure after real label replacement") return original(argv, **kwargs) with patch.object(module.sdputils, "run_checked", side_effect=fail_labelsync), redirect_stdout(output), redirect_stderr(output): try: module.convertlabel(label) except subprocess.CalledProcessError as error: self.check(error.returncode == 7, "Unexpected conversion failure") else: raise RuntimeError("Injected labelsync failure was not reached") recovery, = set(self.work.glob(".sdp-label-recovery-*")) - prior self.check((recovery / "label-name.txt").read_text().strip() == label, "Recovery directory does not identify its label") self.check((recovery / "owner.txt").read_text() == self.users[1], "Failure lost the original label owner") # The real failed conversion has replaced the spec but not its tags. exception_level = self.api.exception_level try: self.api.exception_level = 1 # An empty label produces a warning. remaining = self.api.run_files("@" + label) finally: self.api.exception_level = exception_level self.check(not remaining, "Injected failure did not exercise an empty replaced label") with redirect_stdout(output), redirect_stderr(output): original([self.env["P4BIN"], "-x", str(recovery / "label.labelfiles"), "labelsync", "-l", label]) module.p4lock.lock(label, owner_file=str(recovery / "owner.txt")) (self.work / "label-recovery.output").write_text(output.getvalue()) # The long-lived API connection queried the deliberately empty label. # Reconnect after external CLI restoration to discard that cached view. self.api.disconnect() self.api.connect() self.api.password = Path(self.env["SDP_ADMIN_PASSWORD_FILE"]).read_text().strip() self.api.run_login() restored = self.api.fetch_label(label) actual = {(row["depotFile"], row["rev"]) for row in self.api.run_files("@" + label)} self.check(actual == expected, "Recovery from retained files changed tagged revisions") self.check(restored["Owner"] == self.users[1] and "locked" in restored["Options"].split(), "Recovery did not restore ownership and lock status") self.check(restored["Description"] == spec["Description"] and restored["View"] == spec["View"], "Recovery changed label metadata") # Keep the fixture's recovery evidence available in the scenario output. (self.work / "recovered-label-directory.txt").write_text(str(recovery) + "\n") def inactive_clients(self): self.ensure_unload() names = [self.prefix + "_old_" + str(index) for index in range(2)] names.append(self.prefix + "_Swarm_old") for name in names: spec = self.api.fetch_client(name) spec["Root"] = str(self.work / name) spec["View"] = ["//%s/... //%s/..." % (self.depot, name)] self.api.save_client(spec, "-f") # Access is server-maintained. Use actual records and a controlled # selection clock, without changing the server clock or its database. self.script("accessdates.py", self.args.instance, "--client-snapshot", "clients.json") sys.path.insert(0, str(self.source)) import accessdates import sdputils import unload_clients records = [row for row in self.api.run_clients() if row["client"] in names] weeks = int(sdputils.SDPUtils(self.args.instance).get("weeks")) future = int(time.time()) + (weeks + 1) * 7 * 86400 report = accessdates.get_inactive_specs(self.api, "clients", "Access", weeks, records, future) self.check(set(report) == set(names), "Age report omitted fixture records") snapshot = self.work / "selected-clients.json" sdputils.write_client_snapshot(snapshot, self.args.instance, self.env["P4PORT"], weeks, records, future) selected = [entry["client"] for entry in json.loads(snapshot.read_text())["clients"]] self.check(set(selected) == set(names[:2]), "Shared selection failed its Swarm exclusion") output = io.StringIO() with patch.dict(os.environ, self.env, clear=True), patch("time.time", return_value=future), \ redirect_stdout(output), redirect_stderr(output): status = unload_clients.main([self.args.instance, "--clients-file", str(snapshot), "--limit", "1"]) (self.work / "selected-clients.output").write_text(output.getvalue()) self.check(status == 0, "Real snapshot unload failed: " + output.getvalue()) unloaded = {row["client"] for row in self.api.run_clients("-U")} active = {row["client"] for row in self.api.run_clients()} self.check(selected[0] in unloaded and selected[1] in active and names[2] in active, "Snapshot unload ignored its limit or Swarm exclusion") # Model the saved report having an older Access value while the real # server has a recent one. The recheck and rejected mutation are real. real_now = int(time.time()) stale = [{"client": selected[1], "Access": "1"}] recent_snapshot = self.work / "became-active.json" sdputils.write_client_snapshot(recent_snapshot, self.args.instance, self.env["P4PORT"], weeks, stale, real_now) with patch.dict(os.environ, self.env, clear=True), redirect_stdout(output), redirect_stderr(output): status = unload_clients.main([self.args.instance, "--clients-file", str(recent_snapshot), "--limit", "1"]) self.check(status == 0 and selected[1] in {row["client"] for row in self.api.run_clients()}, "A stale report unloaded a currently active client") (self.work / "selected-clients.output").write_text(output.getvalue()) def protections(self): original = self.api.fetch_protect() rows = list(original["Protections"]) added = "list group %s 192.0.2.* //%s/..." % (self.groups[0], self.depot) preserved = "list group %s 198.51.100.* //%s/..." % (self.groups[1], self.depot) try: spec = self.api.fetch_protect() spec["Protections"] = rows + [added, preserved] self.api.save_protect(spec) result = self.command([self.env["P4BIN"], "protect", "-o"]) (self.work / "p4.protect").write_text(result.stdout) (self.work / "remove-groups.txt").write_text(self.groups[0] + "\n") filtered = self.script("clean_protect.py", "remove-groups.txt", "p4.protect").stdout expected = "".join(line for line in result.stdout.splitlines(keepends=True) if added not in line) self.check(filtered == expected and preserved in filtered, "Protection filtering changed unselected lines or missed the restricted-host row") finally: restored = self.api.fetch_protect() restored["Protections"] = rows self.api.save_protect(restored) def unload_label(self): self.ensure_unload() name = self.prefix + "_old_label" spec = self.api.fetch_label(name) spec["View"] = ["//" + self.depot + "/..."] self.api.save_label(spec, "-f") self.check(all(int(row["Access"]) > time.time() - 86400 for row in self.api.run_labels()), "Label unload requires a fresh fixture server with no old labels") self.script("unload_labels.py", self.args.instance) active = {row["label"] for row in self.api.run_labels()} self.check(name in active, "Age-based label unload removed a recent fixture label") def delete_shelf_client(self): name = self.prefix + "_shelved_client" spec = self.api.fetch_client(name) spec["Root"] = str(self.work / name) spec["View"] = ["//%s/... //%s/..." % (self.depot, name)] self.api.save_client(spec) self.api.client = name try: target = "//" + self.depot + "/root.txt" self.api.run_sync(target) self.api.run_edit(target) change = self.api.fetch_change() change["Description"] = "Disposable shelf cleanup fixture\n" result = self.api.save_change(change) number = str(result[0]).split()[1] self.api.run_shelve("-c", number) self.api.run_revert("//" + name + "/...") self.check(all(int(row["Access"]) > time.time() - 86400 for row in self.api.run_clients()), "Shelf cleanup requires a fresh fixture server with no old clients") self.script("del_shelve.py", self.args.instance) active = {row["client"] for row in self.api.run_clients()} shelved = {str(row["change"]) for row in self.api.run_changes("-s", "shelved")} self.check(name in active and number in shelved, "Shelf cleanup removed a recent client or its shelved change") # Limit the controlled age-selection clock to this exact fixture. # Actual shelf/client deletion commands and unrelated preservation # are observed against the real server. with self.python_script("del_shelve.py") as module: original_select = module.accessdates.get_inactive_specs def select_fixture(api, command, field, weeks): records = [row for row in api.run("clients", "-e", name) if row["client"] == name] return original_select(api, command, field, weeks, records, int(time.time()) + (weeks + 1) * 7 * 86400) output = io.StringIO() with patch.object(module.accessdates, "get_inactive_specs", side_effect=select_fixture), redirect_stdout(output), redirect_stderr(output): module.main() (self.work / "aged-shelf-cleanup.output").write_text(output.getvalue()) after_clients = {row["client"] for row in self.api.run_clients()} after_shelves = {str(row["change"]) for row in self.api.run_changes("-s", "shelved")} self.check(after_clients == active - {name} and after_shelves == shelved - {number}, "Controlled aged-client cleanup removed unexpected clients/shelves or retained its target") finally: self.api.client = self.client def remove_job(self): name = self.prefix + "_job" spec = self.api.fetch_job(name) spec["Description"] = "Disposable job for maintenance integration\n" self.api.save_job(spec) self.api.run_fix("-c", self.change, name) self.check(bool(self.api.run_fixes("-j", name)), "Fixture job has no fix to remove") (self.work / "jobs.txt").write_text(name + " on 2026/01/01 by fixture *open* 'fixture'\n") self.script("remove_jobs.py", self.args.instance, "jobs.txt") self.check(not self.api.run_jobs("-e", "Job=" + name), "Job cleanup retained its fixture job") self.check(not self.api.run_fixes("-j", name), "Job cleanup retained its fixture fix") def run(self): cases = [ ("maintenance_fixture", [], "Create and submit isolated files, including a root file, spaced path and deleted-only directory.", self.fixture_files), ("maintenance_groups", ["createusers.py", "creategroups.py", "addusertogroup.py", "removeusersfromgroup.py", "removeuserfromgroups_file.py", "sdputils.py"], "Real bulk users/groups, member addition and both removal entry points preserve unrelated members and clean scratch.", self.users_groups), ("maintenance_user_reports", ["checkusers.py", "checkusers_not_in_group.py", "make_email_list.py"], "Query real group membership and distinguish existing/missing/ungrouped users; export addresses without sending mail.", self.user_reports), ("maintenance_inventory", ["create_p4_filelist.sh", "file_inventory.py"], "Real marshalled p4 output inventories root, whitespace and deleted-only paths exactly once.", self.inventory), ("maintenance_labels", ["isitalabel.py", "p4unlock.py", "p4lock.py", "convert_label_to_autoreload.py"], "Real label lookup, unlock/relock owner restoration and autoreload conversion preserve tagged revisions.", self.labels), ("maintenance_label_failure_recovery", ["convert_label_to_autoreload.py", "p4lock.py"], "Inject failure after real label replacement, then restore tags/owner/lock from retained recovery files and compare metadata.", self.label_failure_recovery), ("maintenance_client_unload", ["accessdates.py", "unload_clients.py", "sdputils.py"], "Real client records with a controlled selection clock enforce cap/Swarm exclusion; current server access prevents a stale report from unloading a recent client.", self.inactive_clients), ("maintenance_label_unload", ["unload_labels.py"], "Execute the real age-filtered command and verify a recent label is preserved; old-label mutation is not exercised.", self.unload_label), ("maintenance_shelf_cleanup", ["del_shelve.py"], "Preserve a recent real shelf/client, then use a controlled age selection for that exact fixture and verify only its shelf/client are deleted.", self.delete_shelf_client), ("maintenance_protections", ["clean_protect.py"], "Filter real protection output with restricted hosts, preserve unselected text, and restore the original server table.", self.protections), ("maintenance_jobs", ["remove_jobs.py"], "Delete only the uniquely named fixture job after removing its real submitted-change fix.", self.remove_job), ] try: for args in cases: self.scenario(*args) finally: self.api.disconnect() print(json.dumps({"scenarios": self.records, "fixture_directory": str(self.work)}, indent=2)) return int(any(row["status"] != "passed" for row in self.records)) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--root", required=True) parser.add_argument("--instance", required=True) parser.add_argument("--port", required=True, type=int) parser.add_argument("--os-user", required=True) parser.add_argument("--confirm-disposable", action="store_true") args = parser.parse_args() try: return MaintenanceIntegration(args).run() except Exception as error: print(json.dumps({"guard_error": str(error), "scenarios": []}, indent=2)) return 2 if __name__ == "__main__": sys.exit(main())