#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""test_tenantsync_prune.py — felhom-tenantsync v1.3.0 gives every household namespace its prune job (R-1165).

Runs the REAL script against stubbed `proxmox-backup-manager` / `proxmox-backup-client` / `df` on PATH (real bash, real
jq). The stubs keep the prune jobs in a JSON file and LOG every call, so each case asserts what the script DID:

  1. provision of a new household → exactly one `prune-job create prune-<id> --store felhom-offsite --ns <id>
     --max-depth 0 --keep-weekly 2 --schedule 03:30`, and the provision still succeeds;
  2. a second run (reissue) with the job present → no create (idempotent); a job under ANOTHER id that already names
     the namespace counts too;
  3. a 31-character customer id → a job id of at most 32 characters (PBS's limit);
  4. a create that fails → logged, the provision still succeeds (the alarm names the namespace);
  5. deprovision → the namespace's job removed, another household's kept;
  6. usage → `unpruned` = the namespaces no job covers; a root job without max-depth 0 covers all; an unreadable job
     list → `unpruned_checked: false`, never an empty list that reads as „all covered".

Exit 0 all passed · 1 a case failed. Without bash or jq it says NOT CHECKED out loud and exits 0 (the CI runner may
lack jq; the workspace pre-push run has both).
"""
import json
import os
import shutil
import subprocess
import sys
import tempfile

HERE = os.path.dirname(os.path.abspath(__file__))
SCRIPT = os.path.join(HERE, "felhom-tenantsync.sh")

PBM = r'''#!/usr/bin/env bash
echo "pbm $*" >> "$CALLLOG"
case "$1 $2" in
  "cert info")            echo "Fingerprint (sha256): aa:bb:cc:dd" ;;
  "datastore list")       printf '[{"name":"felhom-offsite","path":"%s"}]\n' "$DS_PATH" ;;
  "user generate-token")  echo '{"value": "FAKE-TOKEN-VALUE"}' ;;
  "user list")            echo '[{"userid":"felhom@pbs"}]' ;;
  "user list-tokens")     echo '[]' ;;
  "prune-job list")       [ -n "${JOBS_FAIL:-}" ] && exit 1; cat "$JOBS" ;;
  "prune-job create")     [ -n "${CREATE_FAIL:-}" ] && exit 1
                          id="$3"; shift 3; ns=""
                          while [ $# -gt 0 ]; do [ "$1" = "--ns" ] && ns="$2"; shift; done
                          jq -c --arg id "$id" --arg ns "$ns" '. + [{"id":$id,"store":"felhom-offsite","ns":$ns,"keep-weekly":2,"schedule":"03:30"}]' "$JOBS" > "$JOBS.n" && mv "$JOBS.n" "$JOBS" ;;
  "prune-job remove")     jq -c --arg id "$3" 'map(select(.id != $id))' "$JOBS" > "$JOBS.n" && mv "$JOBS.n" "$JOBS" ;;
  *)                      : ;;
esac
'''
PBC = r'''#!/usr/bin/env bash
echo "pbc $*" >> "$CALLLOG"
case "$1 $2" in
  "namespace list")   echo '{"data":[]}' ;;
  *)                  : ;;
esac
'''
DF = '#!/usr/bin/env bash\necho "1B-blocks Used Avail"\necho "41943040 8388608 33554432"\n'

fails = []
ran = 0


class Box(object):
    def __init__(self):
        self.dir = tempfile.mkdtemp(prefix="tenantsync-prune-")
        self.bin = os.path.join(self.dir, "bin")
        os.makedirs(self.bin)
        for name, body in (("proxmox-backup-manager", PBM), ("proxmox-backup-client", PBC), ("df", DF)):
            p = os.path.join(self.bin, name)
            open(p, "w").write(body)
            os.chmod(p, 0o755)
        self.ds = os.path.join(self.dir, "ds")
        os.makedirs(os.path.join(self.ds, "ns"))
        self.jobs = os.path.join(self.dir, "jobs.json")
        self.calls = os.path.join(self.dir, "calls.log")
        self.set_jobs([])

    def set_jobs(self, jobs):
        open(self.jobs, "w").write(json.dumps(jobs))

    def get_jobs(self):
        return json.load(open(self.jobs))

    def run(self, payload, **env):
        open(self.calls, "w").close()
        e = dict(os.environ, PATH=self.bin + os.pathsep + os.environ["PATH"], DS_PATH=self.ds, JOBS=self.jobs,
                 CALLLOG=self.calls)
        e.update(env)
        p = subprocess.run(["bash", SCRIPT], input=json.dumps(payload), env=e, capture_output=True, text=True)
        return p.returncode, p.stdout, p.stderr, open(self.calls).read().splitlines()

    def close(self):
        shutil.rmtree(self.dir, ignore_errors=True)


def check(name, cond, detail=""):
    global ran
    ran += 1
    if cond:
        print("ok   %s" % name)
    else:
        fails.append("%s %s" % (name, detail))


def creates(calls):
    return [c for c in calls if c.startswith("pbm prune-job create")]


def main():
    if not shutil.which("bash") or not shutil.which("jq"):
        print("test_tenantsync_prune: NOT CHECKED — bash or jq is missing here (the workspace run has both)")
        return 0

    b = Box()
    try:
        rc, out, err, calls = b.run({"op": "provision", "customer_id": "house-1"})
        cr = creates(calls)
        check("provision/creates-the-job", rc == 0 and len(cr) == 1 and cr[0] == (
            "pbm prune-job create prune-house-1 --store felhom-offsite --ns house-1 --max-depth 0 --keep-weekly 2 "
            "--schedule 03:30 --comment R-1165: one copy per ISO week, 2 weeks, server-side (box tokens are write-only)"),
            "rc=%d creates=%s err=%s" % (rc, cr, err[-300:]))
        check("provision/still-answers-ok", '"status":"ok"' in out and "token_secret" in out, out[-200:])

        rc, out, err, calls = b.run({"op": "reissue", "customer_id": "house-1"})
        check("reissue/idempotent", rc == 0 and not creates(calls) and len(b.get_jobs()) == 1,
              "rc=%d creates=%s jobs=%s" % (rc, creates(calls), b.get_jobs()))

        b.set_jobs([{"id": "hand-made", "store": "felhom-offsite", "ns": "house-2", "keep-weekly": 2}])
        rc, out, err, calls = b.run({"op": "provision", "customer_id": "house-2"})
        check("provision/a-job-under-another-id-counts", rc == 0 and not creates(calls), "creates=%s" % creates(calls))

        long_id = "a" * 31
        rc, out, err, calls = b.run({"op": "provision", "customer_id": long_id})
        cr = creates(calls)
        jid = cr[0].split()[3] if cr else ""
        check("provision/long-id-fits-32", rc == 0 and 0 < len(jid) <= 32 and jid.startswith("prune-"), "id=%r" % jid)

        rc, out, err, calls = b.run({"op": "provision", "customer_id": "house-3"}, CREATE_FAIL="1")
        check("provision/create-failure-is-not-fatal", rc == 0 and '"status":"ok"' in out and "could not create prune job" in err,
              "rc=%d out=%s err=%s" % (rc, out[-120:], err[-200:]))

        b.set_jobs([{"id": "prune-house-1", "store": "felhom-offsite", "ns": "house-1"},
                    {"id": "prune-house-9", "store": "felhom-offsite", "ns": "house-9"}])
        rc, out, err, calls = b.run({"op": "deprovision", "customer_id": "house-1"})
        ids = [j["id"] for j in b.get_jobs()]
        check("deprovision/removes-only-its-job", rc == 0 and ids == ["prune-house-9"], "rc=%d ids=%s" % (rc, ids))

        for n in ("demo-hp", "tester-1", "Tester-2"):
            os.makedirs(os.path.join(b.ds, "ns", n))
        b.set_jobs([{"id": "prune-demo-hp", "store": "felhom-offsite", "ns": "demo-hp", "max-depth": 0},
                    {"id": "other-store", "store": "elsewhere", "ns": "tester-1"}])
        rc, out, err, calls = b.run({"op": "usage"})
        u = json.loads(out) if out.strip().startswith("{") else {}
        check("usage/names-the-unpruned", rc == 0 and u.get("unpruned_checked") is True and u.get("unpruned") == ["Tester-2", "tester-1"]
              and u.get("total") == 41943040, out)
        check("usage/no-mutation", not [c for c in calls if any(k in c for k in ("create", "remove", "generate-token", "acl"))], calls)

        b.set_jobs([{"id": "root", "store": "felhom-offsite", "keep-weekly": 2}])
        rc, out, err, calls = b.run({"op": "usage"})
        u = json.loads(out) if out.strip().startswith("{") else {}
        check("usage/a-root-job-covers-all", u.get("unpruned") == [] and u.get("unpruned_checked") is True, out)

        b.set_jobs([{"id": "root0", "store": "felhom-offsite", "max-depth": 0}])
        rc, out, err, calls = b.run({"op": "usage"})
        u = json.loads(out) if out.strip().startswith("{") else {}
        check("usage/a-root-job-with-depth-0-covers-none", u.get("unpruned") == ["Tester-2", "demo-hp", "tester-1"], out)

        rc, out, err, calls = b.run({"op": "usage"}, JOBS_FAIL="1")
        u = json.loads(out) if out.strip().startswith("{") else {}
        check("usage/unreadable-jobs-say-so", rc == 0 and u.get("unpruned_checked") is False and "unpruned" not in u, out)
    finally:
        b.close()

    if fails:
        for f in fails:
            print("FAIL: %s" % f)
        return 1
    print("test_tenantsync_prune: all %d cases passed" % ran)
    return 0


if __name__ == "__main__":
    sys.exit(main())