fixture_inventory.py #1

#!/usr/bin/env python3
"""Exact disposable file/spec/shelf fixture for real recovery assertions.

The caller owns the disposable-host guard. This helper requires an explicit
fixture name and uses only its dedicated depot, client, label and group.
"""
import hashlib
import io
import marshal
from pathlib import Path
import re
import subprocess


class FixtureInventory:
    def __init__(self, env, name):
        if not re.fullmatch(r"sdp-fixture-[a-z0-9-]+", name):
            raise ValueError("Refusing a non-fixture namespace")
        self.env = env
        self.name = name
        self.client = name + "-client"
        self.label = name + "-label"
        self.group = name + "-group"
        self.work = Path(env["P4HOME"]) / "tmp" / name
        self.expected = None
        self.shelf = None
        self.hashes = {}

    def p4(self, *args, data=None, port=None, raw=False):
        if args and args[0] == "login":
            raw = True  # Password prompts are plain text even with -G.
        argv = [self.env["P4BIN"], "-p", port or self.env["P4PORT"], "-u", self.env["P4USER"], "-c", self.client]
        if not raw:
            argv += ["-G"]
        result = subprocess.run(argv + list(args), env=self.env, input=data, capture_output=True, timeout=60)
        if raw:
            if result.returncode:
                raise RuntimeError("Fixture command failed: " + result.stderr.decode(errors="replace"))
            return result.stdout
        records, stream = [], io.BytesIO(result.stdout)
        while stream.tell() < len(result.stdout):
            row = marshal.load(stream)
            records.append({k.decode() if isinstance(k, bytes) else k: v.decode() if isinstance(v, bytes) else v for k, v in row.items()})
        errors = [r for r in records if r.get("code") == "error" and int(r.get("severity", 0)) >= 3]
        if result.returncode or errors:
            raise RuntimeError("Fixture command %r failed: %s" % (args, errors or records))
        return records

    def form(self, command, fields):
        return self.p4(command, "-i", data=marshal.dumps({k.encode(): v.encode() for k, v in fields.items()}, 0))

    def create(self):
        self.work.mkdir()
        self.form("depot", {"Depot": self.name, "Owner": self.env["P4USER"], "Type": "local", "Map": self.name + "/...", "Description": "Disposable exact recovery fixture\n"})
        self.form("client", {"Client": self.client, "Owner": self.env["P4USER"], "Root": str(self.work), "LineEnd": "unix", "View0": "//%s/... //%s/..." % (self.name, self.client)})
        for filename, payload, filetype in (("text.txt", b"text revision one\n", "text+C"), ("binary.dat", bytes(range(256)) * 16, "binary+F"), ("space file.txt", b"space filename payload\n", "text+C"), ("deleted.txt", b"historical deleted bytes\n", "text+C"), ("move source.txt", b"historical moved bytes\n", "text+C")):
            path = self.work / filename
            path.write_bytes(payload)
            self.p4("add", "-t", filetype, str(path))
            self.p4("submit", "-d", "fixture initial " + filename, str(path))
            self.hashes["//%s/%s#1" % (self.name, filename)] = hashlib.sha256(payload).hexdigest()
        path = self.work / "text.txt"
        self.p4("edit", str(path))
        path.write_bytes(b"text revision two\n")
        self.p4("submit", "-d", "fixture second revision", str(path))
        self.hashes["//%s/text.txt#2" % self.name] = hashlib.sha256(path.read_bytes()).hexdigest()
        self.p4("delete", str(self.work / "deleted.txt"))
        self.p4("submit", "-d", "fixture deletion", str(self.work / "deleted.txt"))
        self.p4("edit", str(self.work / "move source.txt"))
        self.p4("move", str(self.work / "move source.txt"), str(self.work / "move destination.txt"))
        self.p4("submit", "-d", "fixture move")
        self.hashes["//%s/move destination.txt#1" % self.name] = hashlib.sha256(b"historical moved bytes\n").hexdigest()
        self.form("label", {"Label": self.label, "Owner": self.env["P4USER"], "Description": "recovery fixture label\n", "Options": "unlocked noautoreload", "View0": "//" + self.name + "/..."})
        self.p4("tag", "-l", self.label, "//" + self.name + "/...#head")
        self.form("group", {"Group": self.group, "Users0": self.env["P4USER"], "MaxResults": "unlimited"})
        self.p4("edit", str(path))
        path.write_bytes(b"exact shelved content\n")
        response = self.form("change", {"Change": "new", "Client": self.client, "User": self.env["P4USER"], "Description": "recovery fixture shelf\n"})
        match = re.search(r"Change (\d+) created", str(response))
        if not match:
            raise RuntimeError("Fixture shelf changelist was not created")
        self.shelf = match.group(1)
        self.p4("reopen", "-c", self.shelf, str(path))
        self.p4("shelve", "-c", self.shelf)
        self.p4("revert", str(path))
        self.expected = self.snapshot()
        if self.expected["hashes"] != self.hashes or len(self.expected["revisions"]) != 9:
            raise RuntimeError("Initial fixture differs from independently expected revisions/bytes")
        return self.expected

    def snapshot(self, port=None):
        revisions = sorted([{k: r[k] for k in ("depotFile", "rev", "action", "type")} for r in self.p4("files", "-a", "//" + self.name + "/...", port=port) if "depotFile" in r], key=lambda r: (r["depotFile"], r["rev"]))
        hashes = {r["depotFile"] + "#" + r["rev"]: hashlib.sha256(self.p4("print", "-q", r["depotFile"] + "#" + r["rev"], port=port, raw=True)).hexdigest() for r in revisions if r["action"] not in ("delete", "move/delete")}
        specs = {}
        for command, keys in (("client", ("Client", "Owner", "Root", "LineEnd", "Options", "View0")), ("label", ("Label", "Owner", "Options", "Description", "View0")), ("group", ("Group", "Users0", "MaxResults"))):
            row = next(r for r in self.p4(command, "-o", getattr(self, command), port=port) if r.get("code") == "stat")
            specs[command] = {k: row.get(k) for k in keys}
        label_members = sorted([r["depotFile"], r["rev"]] for r in self.p4("files", "//%s/...@%s" % (self.name, self.label), port=port) if "depotFile" in r)
        payload = self.p4("print", "-q", "//%s/text.txt@=%s" % (self.name, self.shelf), port=port, raw=True)
        if payload != b"exact shelved content\n":
            raise RuntimeError("Shelved content changed")
        shelf = [{k: v for k, v in r.items() if k in ("desc", "client", "user", "status") or k.startswith(("depotFile", "rev", "type", "action"))} for r in self.p4("describe", "-S", "-s", self.shelf, port=port) if r.get("code") == "stat"]
        return {"revisions": revisions, "hashes": hashes, "specs": specs, "label_members": label_members, "shelf": shelf, "shelf_sha256": hashlib.sha256(payload).hexdigest()}

    def assert_unchanged(self, port=None):
        actual = self.snapshot(port)
        if actual != self.expected:
            raise RuntimeError("Recovered fixture differs in revisions, bytes, specs, label membership or shelf")

    def post_recovery_write(self):
        path = self.work / "post-recovery.txt"
        path.write_bytes(b"new write after recovery\n")
        self.p4("add", "-t", "text+C", str(path))
        self.p4("submit", "-d", "post recovery fixture write", str(path))
        if self.p4("print", "-q", "//%s/post-recovery.txt#1" % self.name, raw=True) != b"new write after recovery\n":
            raise RuntimeError("Post-recovery submitted bytes differ")