Newer
Older
openstack-caracal-ipv4 / scripts / phase03_accept_walk.py
#!/usr/bin/env python3
# scripts/phase03_accept_walk.py
#
# Read `juju status --format json` on stdin; report every unit (subordinates included)
# that is NOT workload=active / juju=idle, and classify each as EXPECTED or UNEXPECTED.
#
# This is the phase-03 Step 3.1 acceptance gate, hardened from the do-doc's bare COUNT
# (1 or 2) to an IDENTITY check: a different app sitting blocked would also produce
# count==2 yet must FAIL. The only post-deploy exceptions allowed here are:
#   - octavia/*                    : workload=blocked, msg mentions 'configure-resources'
#                                    (D-021; cleared in phase-05)
#   - glance-simplestreams-sync/*  : workload in {unknown, waiting} (image-sync state)
#
# Exit 0 if every non-active/idle unit is EXPECTED (gate clear); 1 if any UNEXPECTED;
# 2 if stdin is not parseable juju-status JSON. Read-only. ASCII + LF.
import json
import sys


def expected(name, ws, msg):
    if name.startswith("octavia/") and ws == "blocked" and "configure-resources" in msg:
        return True
    if name.startswith("glance-simplestreams-sync/") and ws in ("unknown", "waiting"):
        return True
    return False


def walk(units, out):
    for name, u in (units or {}).items():
        ws = (u.get("workload-status") or {}).get("current") or ""
        js = (u.get("juju-status") or {}).get("current") or ""
        msg = (u.get("workload-status") or {}).get("message") or ""
        if ws != "active" or js != "idle":
            out.append((name, ws, js, msg))
        walk(u.get("subordinates"), out)


def main():
    try:
        d = json.load(sys.stdin)
    except Exception as e:  # noqa: BLE001 - any parse failure is a precondition fail
        print("FATAL: cannot parse juju status JSON: %s" % e, file=sys.stderr)
        return 2
    bad = []
    for app in (d.get("applications") or {}).values():
        walk(app.get("units"), bad)
    unexpected = [b for b in bad if not expected(b[0], b[1], b[3])]
    print("Non-active/idle units: %d (expected: %d, UNEXPECTED: %d)"
          % (len(bad), len(bad) - len(unexpected), len(unexpected)))
    for name, ws, js, msg in bad:
        tag = "ok" if expected(name, ws, msg) else "XX"
        print("  [%s] %s: workload=%s juju=%s msg=%s" % (tag, name, ws, js, msg))
    return 1 if unexpected else 0


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