SOLIDWORKS Bridge

#!/usr/bin/env python3
"""live-smoke.py — exercise the swx bridge against the user's REAL SOLIDWORKS.

Mock-green is not green: v0.2 passed 49 mock tests while shipping four COM bugs.
This suite runs the v0.3 surface against a live SOLIDWORKS, checks EFFECTS
(bytes on disk, geometry read back in millimetres, a dimension that really
moved) rather than response flags, and puts the machine back as it found it.

THE BRIDGE IS NOW DENY-BY-DEFAULT. It refuses to write to any document it did
not create, and to any destination that already exists, unless the process was
started with SWX_ALLOW_USER_WRITES=1 AND the call asks explicitly. This suite
runs in the DEFAULT posture on purpose: it must pass without that opt-in, which
is the strongest single statement that the safe subset is genuinely useful.

SAFETY — the contract this file holds itself to, and why each rule exists
─────────────────────────────────────────────────────────────────────────
This suite runs against a CAD application somebody is working in. An earlier
draft of it could, if `swx_new_document` failed, have carried on and hit the
ACTIVE document — the user's — with set_parameter, props_set and a plain
save_as that RENAMES AND RELOCATES it, then closed it save:false and deleted
the folder it had been moved into. That is a data-loss path, so the rules are
now structural rather than aspirational:

  * OWN NOTHING BY ACCIDENT. `Doc` tracks the title of every document this
    suite created. `mutate()` REFUSES to send a mutating verb unless its target
    is one of them — there is no code path that reaches a mutating verb without
    passing that check.
  * BASELINE IS SACRED. The set of documents open at the start is captured, and
    `assert_not_users()` fails hard if the active document is ever one of them.
  * ABORT MEANS ABORT. Preflight failures (wrong version, mock mode, SOLIDWORKS
    down) and a failed `new_document` raise `Abort`, which skips straight to
    cleanup. The batch runner also stops sending on the first failed step
    instead of marching on through the rest of the list.
  * THE INVARIANT IS A GATE, NOT A REPORT. "SOLIDWORKS ended with exactly the
    documents it started with" runs even on an early abort, and a failure there
    fails the whole suite regardless of how many checks passed.
  * NOTHING IS DELETED THAT WE DID NOT WRITE. Cleanup removes named files it
    created, not a directory tree.
  * NEVER launches, stops or kills SOLIDWORKS, and never closes a window.

It also deliberately DOES exercise the refusals: the .wrl bypass, the
non-bridge-document refusal, the unattributed-window refusal. Those are checked
because they must never reach COM — a `.wrl` that got through would hang the
user's SOLIDWORKS for two minutes.

BEFORE RUNNING: restart the bridge process. Document ownership lives in the
server's memory and is scoped to a SOLIDWORKS session; starting from a clean
process means the suite cannot inherit a stale claim on a title that now
belongs to somebody else's document. Preflight checks the version, refuses mock
mode, and refuses to run at all if SOLIDWORKS is not already up (it will not
start it — this bridge cannot start the Connected edition, and will not try).

Usage:
    python3 tools/live-smoke.py                   # full run
    python3 tools/live-smoke.py --keep            # leave the scratch files behind
    python3 tools/live-smoke.py --touch-user-doc  # ALSO aim a refused mutating
                                                  # verb at the user's own open
                                                  # document (opt-in; see
                                                  # phase_ownership_refusal)

Exit 0 = every check passed AND the machine was left as found.
"""
import atexit
import importlib.util
import json
import os
import signal
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
_spec = importlib.util.spec_from_file_location("swx_batch", os.path.join(_HERE, "swx-batch.py"))
swx_batch = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(swx_batch)

DIR = r"C:\Users\Public\swx-smoke4"
KEEP = "--keep" in sys.argv
# Set by preflight from swx_status.abAvailable.
AB_AVAILABLE = True
# Set by preflight from swx_status.userWrites. The suite is written for the
# DEFAULT posture; when the bridge has been started with the write opt-in (to
# live-test the opt-in paths), the checks that assert a refusal have to assert
# the RIGHT refusal instead of the wrong one. Nothing in this suite becomes
# destructive in either posture — the difference is only which error is correct.
WRITES_ENABLED = False
# Opt-in: aim a (refused) mutating verb at the user's REAL document. Off by
# default — see phase_ownership_refusal.
TOUCH_USER_DOC = "--touch-user-doc" in sys.argv

# Verbs that CHANGE a document. Every one of these must be aimed at a document
# this suite created; `mutate()` is the only way to call them.
MUTATING = {
    "swx_run_script", "swx_set_parameter", "swx_props_set", "swx_save",
    "swx_save_as", "swx_export", "swx_close_doc",
}
for _f in ["step", "iges", "stl", "3mf", "glb", "sat", "parasolid", "pdf", "dxf", "dwg"]:
    MUTATING.add("swx_export_" + _f)
# rebuild_check calls ForceRebuild3: it re-solves every feature and dirties the
# document. Left out of this set, it was the one call in the suite that could
# still land on Drew's model if he clicked it mid-batch — and a large assembly
# can outrun the timeout or park on a rebuild-error dialog.
MUTATING.add("swx_rebuild_check")
MUTATING.add("swx_bom")
MUTATING.add("swx_pack_and_go")
# set_preference writes a document-scoped setting and dirties the document. It
# was called through `call()` with no targetDocument, so it aimed at whatever was
# ACTIVE — the same class of miss as rebuild_check.
MUTATING.add("swx_set_preference")

CHECKS = []
# Names this suite wrote. Recorded only AFTER a write reports success, and never
# for a name that already existed at start-up — `FILES_WRITTEN` used to record
# INTENT (appended before the export ran, and unconditionally after save_as),
# so a pre-existing file at one of the fixed names would have been deleted as
# though the suite had created it.
FILES_WRITTEN = []
# Whatever was in DIR before we started. Nothing in here is ever deleted.
PRE_EXISTING = set()
# Callables that put the user's machine back. Registered the moment something
# is changed, so cleanup restores it on EVERY exit path — not just the one
# where the run got as far as its own restore line.
RESTORE = []


class Abort(Exception):
    """Stop the run and go straight to cleanup + the invariant gate."""


def check(name, ok, detail=""):
    CHECKS.append((name, bool(ok), str(detail)[:400]))
    print(("  PASS  " if ok else "  FAIL  ") + name + ("" if ok else "\n          <- " + str(detail)[:400]))
    return bool(ok)


def require(name, ok, detail=""):
    """A check that ABORTS the run when it fails."""
    if not check(name, ok, detail):
        raise Abort(name)
    return True


class Doc:
    """The documents this suite created — the only ones it may touch.

    Entries are (title, path). The PATH is the identity wherever there is one:
    two folders can each hold a `thing.sldprt`, and a bridge that treated the
    title as the identity once renamed the user's document instead of its own.
    """

    def __init__(self):
        self.mine = []
        self.baseline = []
        # Separate from `baseline` being empty: a machine with nothing open has
        # an empty baseline and MUST still be checked.
        self.baseline_captured = False

    def own(self, title, path=None):
        if not title:
            return
        self.mine = [(t, p) for (t, p) in self.mine if t != title]
        self.mine.append((title, path))

    def disown(self, title):
        self.mine = [(t, p) for (t, p) in self.mine if t != title]

    def owns(self, title):
        return bool(title) and any(t == title for (t, _) in self.mine)

    def path_of(self, title):
        for t, p in self.mine:
            if t == title:
                return p
        return None

    def titles(self):
        return [t for (t, _) in self.mine]


DOCS = Doc()


def call(verb, args=None, label=None):
    rows = swx_batch.run([{"verb": verb, "args": args or {}, "label": label or verb}], 300, 150, quiet=True)
    if not rows:
        return {"success": False, "error": "no response from the desktop"}
    return rows[0]["result"]


def batch(calls, total=600, per=150):
    """Run a list of calls, STOPPING at the first one that fails.

    A batch is ONE round trip, so it is only ever used for calls that are
    read-only or already aimed at a document `mutate()` has cleared. Mutating
    work goes one call at a time, so the active document is re-checked in
    between.
    """
    rows = swx_batch.run(calls, total, per, quiet=True, stop_on_failure=True)
    if rows is None:
        raise Abort("the desktop did not answer the batch — nothing further will be sent")
    return {r["label"]: r["result"] for r in rows}


def mutate(verb, args=None, label=None, where=""):
    """The door to a mutating verb. Refuses anything not ours.

    NOT the only door, and saying otherwise was a false claim: `batch()` also
    sends `swx_run_script` and `swx_save_as` (phase_build, phase_unproven).
    Both carry an explicit `targetDocument` that the caller has already vetted
    against `DOCS`, and both are the FIRST call in their batch, so nothing can
    have moved the active document under them. Every OTHER mutating call comes
    through here.

    `targetDocument` is now REAL — the bridge resolves it server-side and the
    com host acts on that document — so checking it here checks the same thing
    the bridge will do. When it is absent the target is the ACTIVE document,
    which can change under us at any moment (SOLIDWORKS' ActiveDoc follows
    window focus, and the user may click their own part mid-run), so that case
    is re-read immediately before every call rather than once per phase.
    """
    args = dict(args or {})
    if verb in MUTATING:
        named = args.get("name") if verb == "swx_close_doc" else args.get("targetDocument")
        target = named if named is not None else active_title()
        if not DOCS.owns(target):
            raise Abort(
                "REFUSED to send %s at '%s' — this suite only touches documents it created (%s). "
                "%sThis is the guard that stops a failed setup step, or the user clicking their "
                "own document mid-run, from turning into a write to it."
                % (verb, target, DOCS.titles(), (where + ": ") if where else "")
            )
        # Belt and braces: whatever we are about to touch must also not be one
        # of the documents that were open before this suite started.
        assert_not_users("before %s" % verb)
    return call(verb, args, label)


def active_title():
    d = call("swx_docs")
    a = d.get("active") or {}
    return a.get("title")


def doc_key(d):
    return (d.get("title") or "") + "|" + (d.get("path") or "")


def assert_not_users(where):
    """The active document must never be one that was open before we started."""
    t = active_title()
    if t and any((d.get("title") or "") == t for d in DOCS.baseline):
        raise Abort("%s: the ACTIVE document is '%s', which was open before this run started" % (where, t))
    return t


def shell(command, reason):
    r = swx_batch.ab("shell_execute", {"command": command, "timeoutSeconds": 90, "reason": reason})
    return (r.get("stdout") or "").strip()


def listing():
    out = shell(
        'powershell -NoProfile -Command "Get-ChildItem \'%s\' -ErrorAction SilentlyContinue | '
        'ForEach-Object { $_.Name + \'=\' + $_.Length }"' % DIR,
        "Listing the files the SOLIDWORKS live smoke test wrote, to check the bytes really landed")
    sizes = {}
    for line in out.splitlines():
        if "=" in line:
            n, _, v = line.partition("=")
            try:
                sizes[n.strip().lower()] = int(v.strip())
            except ValueError:
                pass
    return sizes


def record_written(name):
    """Record a file this suite actually created. Never a pre-existing one."""
    if name.lower() in PRE_EXISTING:
        check("refusing to claim authorship of the pre-existing %s" % name, False,
              "it was in %s before this run started; it will NOT be deleted" % DIR)
        return
    if name not in FILES_WRITTEN:
        FILES_WRITTEN.append(name)


def near(v, want, tol=0.05):
    try:
        return abs(float(v) - float(want)) <= tol
    except (TypeError, ValueError):
        return False


# ── the run ─────────────────────────────────────────────────────────────────

def preflight():
    st = call("swx_status")
    require("bridge answers /command", st.get("success") is True, st.get("error") or st)
    require("bridge is v0.3.0 (redeploy the exe if not)", st.get("bridgeVersion") == "0.3.0",
            st.get("bridgeVersion"))
    require("NOT in mock mode — this must be the real thing", st.get("mock") is False, st.get("mock"))
    sw = st.get("solidworks") or {}
    require("SOLIDWORKS is running", sw.get("running") is True,
            "Start it from the 3DEXPERIENCE Platform. This bridge cannot launch the Connected "
            "edition and this suite will not try.")
    require("COM is attached", (st.get("com") or {}).get("attached") is True, st.get("com"))
    check("detection reports the edition", sw.get("edition") in ("connected", "desktop", "unknown"),
          sw.get("edition"))
    # The suite runs in the DEFAULT posture on purpose: passing without the
    # opt-in is the strongest statement that the safe subset is genuinely useful.
    # The ab-delegated verbs (window info, dialogs, badge, screenshot) need
    # ADOM_DIRECT_API_URL, which only a bridge SPAWNED BY Adom Bridge inherits.
    # A hand-launched dev bridge fails all of them for a reason that has nothing
    # to do with SOLIDWORKS, so those checks are SKIPPED rather than failed.
    global AB_AVAILABLE, WRITES_ENABLED
    AB_AVAILABLE = bool(st.get("abAvailable"))
    WRITES_ENABLED = st.get("userWrites") == "enabled"
    if WRITES_ENABLED:
        print("  NOTE: this bridge has the WRITE OPT-IN enabled (SWX_ALLOW_USER_WRITES=1).\n"
              "        Refusal checks assert the opt-in-enabled refusal; nothing here writes to\n"
              "        a document it did not create in either posture.")
    if AB_AVAILABLE:
        check("ab's direct API is reachable (window/dialog/badge checks need it)", True, "")
    else:
        # A DEPLOYMENT fact, not a bridge defect: a hand-launched exe never
        # inherits ADOM_DIRECT_API_URL. Reporting it as a failure makes a clean
        # bridge look broken, which is exactly the confusion this suite exists
        # to prevent.
        check("(skipped) ab's direct API — cannot test without an ab-spawned bridge", True,
              "abAvailable=false: this bridge was started by hand. Every ab-delegated verb "
              "(window info, dialogs, taskbar badge, screenshot, kill) is UNTESTED in this run, "
              "not broken. Spawn it through Adom Bridge (bridge_install) to cover them.")
    # NOT a hard requirement any more. It was, because a refusal check that
    # asserts the wrong error in the wrong posture is worse than no check — but
    # the opt-in PATHS now need live testing too (three live-proven cases where
    # the human authorized an action in full and the bridge refused anyway), and
    # a suite that aborts on the posture those paths require cannot test them.
    # Every posture-sensitive check is now explicitly two-sided instead.
    check("the bridge's write posture is reported and understood",
          st.get("userWrites") in ("enabled", "disabled"),
          "userWrites=%s — with writes ENABLED, refusal checks assert the opt-in-enabled "
          "refusal. Nothing in this suite writes to a document it did not create in either "
          "posture; the write opt-in changes which error is correct, not what is touched."
          % st.get("userWrites"))

    # C:\Users\Public is a SHARED, all-users location. If anything is already
    # in our scratch folder it is somebody else's, and nine hard-coded filenames
    # could collide with it — so refuse rather than write into it.
    existing = listing()
    PRE_EXISTING.update(existing)
    require("the scratch folder %s is empty (or absent)" % DIR, not existing,
            "it already contains %s. This is a shared all-users path and this suite writes fixed "
            "filenames into it — move or remove those first. Nothing has been touched."
            % sorted(existing))
    return sw


def phase_start_attaches_and_never_relaunches(sw):
    """SOLIDWORKS is already running, so swx_start must ATTACH — and launch
    NOTHING, with a process count to prove it.

    Attach-first is not just an optimisation: on the Connected editions a second
    launch means a second 3DEXPERIENCE sign-in window on the user's desktop. It
    also has to attach WITHOUT a `reason`, which the launch road requires — a
    reason demanded for a no-op attach would be noise in Drew's activity log
    every time a script checked whether SOLIDWORKS was up."""
    before = shell('powershell -NoProfile -Command "(Get-Process SLDWORKS -ErrorAction '
                   'SilentlyContinue | Measure-Object).Count"',
                   "Counting SOLIDWORKS processes before testing that the bridge attaches rather than launches")
    r = call("swx_start", {})
    after = shell('powershell -NoProfile -Command "(Get-Process SLDWORKS -ErrorAction '
                  'SilentlyContinue | Measure-Object).Count"',
                  "Counting SOLIDWORKS processes after, to prove the bridge started nothing")
    check("swx_start ATTACHES to the running SOLIDWORKS", r.get("success") is True, r.get("error") or r)
    check("swx_start launched nothing (wasLaunched:false)", r.get("wasLaunched") is False, r)
    check("...and the process count is unchanged", before == after, "%s -> %s" % (before, after))
    check("swx_start reports the edition", r.get("edition") in ("connected", "desktop", "unknown"),
          r.get("edition"))
    check("swx_start attaches with NO reason argument (the launch road needs one, attach must not)",
          r.get("errorCode") != "reason_required", r)
    # THE LAUNCH ANSWER, reported but NOT exercised: firing the launcher while
    # SOLIDWORKS is already up is precisely the thing that must never happen, so
    # this suite only proves the bridge has FOUND the sanctioned route. The
    # launch itself is a separate, deliberate test with SOLIDWORKS closed.
    route, shortcut = sw.get("launchRoute"), sw.get("shortcut")
    check("swx_status reports which launch route this machine has",
          route in ("com", "platformShortcut", "blocked"), sw)
    if sw.get("edition") == "connected":
        check("a Connected machine resolves to the PLATFORM SHORTCUT route, not 'blocked'",
              route == "platformShortcut",
              "route=%s shortcut=%s — if this says 'blocked', the shortcut glob missed it; "
              "SWX_SOLIDWORKS_SHORTCUT=<path to the .lnk> is the override" % (route, shortcut))
        check("...and names the shortcut it would invoke",
              isinstance(shortcut, str) and shortcut.lower().endswith(".lnk"), shortcut)
        check("...which exists on disk",
              bool(shortcut) and shell('powershell -NoProfile -Command "if (Test-Path -LiteralPath '
                                       '\'%s\') { \'yes\' } else { \'no\' }"' % shortcut,
                                       "Confirming the SOLIDWORKS launcher shortcut the bridge found "
                                       "really exists (read-only Test-Path)") == "yes",
              shortcut)


def phase_build():
    """Create our own document and build a real solid in it."""
    r = call("swx_new_document", {"type": "part"})
    require("swx_new_document created a part", r.get("success") is True, r.get("error") or r)
    title = (r.get("created") or {}).get("title")
    require("...and it has a title we can target", bool(title), r.get("created"))
    DOCS.own(title)
    check("the new document has NO path until it is saved", (r.get("created") or {}).get("path") == "",
          r.get("created"))
    check("swx_new_document names it the way SOLIDWORKS does", str(title).startswith("Part"), title)
    check("...and reports the template it resolved",
          isinstance(r.get("template"), str) and r["template"].lower().endswith(".prtdot"), r.get("template"))
    require("the active document is now OURS, not the user's", assert_not_users("after new_document") == title,
            title)

    res = batch([
        {"label": "design_before", "verb": "swx_design_info", "args": {}},
        {"label": "build", "verb": "swx_run_script", "args": {"script": [
            {"op": "sketch_rect", "plane": "Front Plane", "widthMm": 50, "heightMm": 30},
            {"op": "extrude", "depthMm": 20},
            {"op": "set_material", "name": "6061 Alloy"},
            {"op": "set_prop", "name": "PartNo", "value": "SMOKE-001"},
        ], "targetDocument": title}},
        {"label": "bodies", "verb": "swx_inspect_bodies", "args": {}},
        {"label": "design_after", "verb": "swx_design_info", "args": {}},
        {"label": "mass", "verb": "swx_mass_properties", "args": {}},
        {"label": "params", "verb": "swx_get_parameters", "args": {}},
        {"label": "docinfo", "verb": "swx_document_info", "args": {}},
        {"label": "appstate", "verb": "swx_get_app_state", "args": {}},
    ])

    check("swx_design_info on the empty part sees 0 bodies", res.get("design_before", {}).get("bodies") == 0,
          res.get("design_before"))
    build = res.get("build", {})
    require("swx_run_script built the solid", build.get("success") is True and build.get("stepCount") == 4,
            build.get("error") or build)
    check("run_script pinned the document it was given", build.get("targetDocument") == title, build.get("targetDocument"))
    check("...and reports it as a bridge-created document", build.get("bridgeCreatedDocument") is True, build)

    bodies = res.get("bodies", {})
    check("swx_inspect_bodies sees exactly one solid body", bodies.get("bodyCount") == 1,
          bodies.get("error") or bodies)
    check("swx_inspect_bodies reports millimetres", bodies.get("units") == "mm", bodies.get("units"))
    size = ((bodies.get("bodies") or [{}])[0]).get("sizeMm") or {}
    check("the body is the 50x30x20 mm box the script asked for",
          near(size.get("x"), 50) and near(size.get("y"), 30) and near(size.get("z"), 20), size)
    check("its volume is 30000 mm3", near(((bodies.get("bodies") or [{}])[0]).get("volumeMm3"), 30000, 1.0),
          ((bodies.get("bodies") or [{}])[0]).get("volumeMm3"))

    check("swx_design_info now sees the body", res.get("design_after", {}).get("bodies") == 1,
          res.get("design_after"))
    check("swx_design_info reports the material that was set",
          ((res.get("design_after", {}).get("material") or {}) or {}).get("name") == "6061 Alloy",
          res.get("design_after", {}).get("material"))
    mass = res.get("mass", {})
    check("swx_mass_properties agrees with the geometry (aluminium, 81 g)",
          near(mass.get("massKg"), 0.081, 0.002), mass.get("massKg"))
    check("swx_mass_properties reports the density it used", near(mass.get("density"), 2700, 5),
          mass.get("density"))
    # ForceRebuild3 goes through `mutate()` like every other write: it is the one
    # call here that could otherwise land on Drew's model if he clicked it
    # between batched calls, and a large assembly can outrun the timeout or park
    # on a rebuild-error dialog.
    rb = mutate("swx_rebuild_check", {"targetDocument": title}, where="rebuild")
    check("swx_rebuild_check reports a clean model", rb.get("errorCount") == 0, rb.get("featureErrors"))
    check("...and no longer marks every feature unreadable (report-bug #4)",
          rb.get("unreadableFeatures") == 0, rb.get("unreadableFeatures"))
    check("swx_document_info says our scratch document is unsaved",
          res.get("docinfo", {}).get("isSaved") is False, res.get("docinfo", {}).get("isSaved"))
    check("swx_get_app_state names the active document", res.get("appstate", {}).get("activeDocument") == title,
          res.get("appstate", {}).get("activeDocument"))

    names = [p.get("name") for p in (res.get("params", {}).get("parameters") or [])]
    check("swx_get_parameters finds the extrusion dimension",
          any(str(n).startswith("D1@Boss-Extrude") for n in names), names[:8])
    return title, next((n for n in names if str(n).startswith("D1@Boss-Extrude")), None)


def phase_drive(title, dim):
    if not dim:
        check("a named dimension to drive", False, "swx_get_parameters found none")
        return
    r = mutate("swx_set_parameter", {"name": dim, "valueMm": 25, "targetDocument": title})
    check("swx_set_parameter drove the dimension to 25 mm",
          r.get("success") is True and near(r.get("valueMm"), 25, 0.01), r.get("error") or r)
    check("...and reports what it changed FROM", near(r.get("before"), 0.02, 0.001), r.get("before"))
    b = call("swx_inspect_bodies")
    z = (((b.get("bodies") or [{}])[0]).get("sizeMm") or {}).get("z")
    check("the MODEL actually moved (the body is now 25 mm deep)", near(z, 25, 0.01), z)


def phase_export(title):
    """Every proven format, in Silent|Copy mode — the fix that was never run."""
    r = mutate("swx_props_set", {"properties": {"Description": "swx live smoke"}, "targetDocument": title},
               where="props")
    check("swx_props_set succeeded", r.get("success") is True, r.get("error") or r)
    g = call("swx_props_get")
    check("swx_props_get round-trips the property",
          ((g.get("properties") or {}).get("Description") or {}).get("resolved") == "swx live smoke",
          g.get("properties"))

    formats = [("step", "import4.step"), ("iges", "run4.igs"), ("stl", "run4.stl"),
               ("3mf", "run4.3mf"), ("glb", "run4.glb"), ("sat", "run4.sat"),
               ("parasolid", "run4.x_t"), ("pdf", "run4.pdf")]
    # ONE AT A TIME, not a single batch. A batch is one round trip: if the user
    # clicks their own document halfway through it, every remaining export in
    # the list would follow the focus with nothing re-checking in between.
    # `mutate()` re-reads the active document before each call.
    for fmt, name in formats:
        e = mutate("swx_export_" + fmt,
                   {"outputPath": DIR + "\\" + name, "targetDocument": title},
                   label=fmt, where="export loop")
        ok = e.get("success") is True
        check("swx_export_%s succeeded" % fmt, ok, e.get("error") or e)
        if ok:
            record_written(name)
            # THE fix that has never been proven: Silent|Copy must leave the
            # open document's identity alone. Every live byte count quoted in
            # the parity report came from a probe using Silent ALONE.
            check("swx_export_%s reported a clean saveErrorCode" % fmt, e.get("saveErrorCode") == 0,
                  e.get("saveErrorCode"))
            check("swx_export_%s did NOT rename the open document" % fmt,
                  e.get("documentRenamed") is False, e.get("documentRenamed"))
            check("swx_export_%s left the title unchanged" % fmt,
                  (e.get("doc") or {}).get("title") == title, (e.get("doc") or {}).get("title"))

    docs = call("swx_docs").get("docs") or []
    check("the exports left our document open under its own name",
          any((d.get("title") or "") == title for d in docs), [d.get("title") for d in docs])

    sizes = listing()
    for _, name in formats:
        check("%s has real bytes on disk" % name, sizes.get(name.lower(), 0) > 0,
              "%s bytes (dir: %s)" % (sizes.get(name.lower(), 0), sorted(sizes)))
    return sizes


def phase_refusals(title):
    """The guards must refuse BEFORE reaching COM. A .wrl that got through
    would hang the user's SOLIDWORKS for two minutes."""
    for verb, args in [
        ("swx_export_step", {"outputPath": DIR + "\\bypass.wrl", "targetDocument": title}),
        ("swx_export_glb", {"outputPath": DIR + "\\bypass.wrl", "targetDocument": title}),
        ("swx_export", {"format": "step", "outputPath": DIR + "\\bypass.wrl", "targetDocument": title}),
        ("swx_save_as", {"outputPath": DIR + "\\bypass.wrl", "targetDocument": title}),
    ]:
        r = mutate(verb, args, label=verb + "-wrl")
        check("%s refuses a .wrl DESTINATION before reaching COM" % verb,
              r.get("errorCode") == "format_blocked", r)
    r = mutate("swx_run_script", {"script": [{"op": "save_as", "path": DIR + "\\bypass.wrl"}],
                                  "targetDocument": title})
    check("the modelling DSL refuses a .wrl destination too", r.get("errorCode") == "bad_script", r)
    sizes = listing()
    check("...and nothing named bypass.wrl was written", "bypass.wrl" not in sizes, sorted(sizes))
    check("SOLIDWORKS is still responsive after the refusals",
          (call("swx_status").get("com") or {}).get("attached") is True, "check swx_status")

    # MACROS. A macro is arbitrary code inside SOLIDWORKS, so this suite never
    # RUNS one — it proves the gate holds. The refusal must come from the
    # opt-in check and NOT from the filesystem: an unauthorized caller does not
    # get to learn what exists on Drew's disk either.
    # The path is deliberately one this suite has NEVER written, so that on an
    # opted-in bridge the call still cannot run anything: the gate opens and the
    # existence check stops it. There is no posture in which this suite executes
    # a macro.
    macro_probe = DIR + "\\no-such-macro.swp"
    r = call("swx_run_macro", {"path": macro_probe,
                               "allowUserWrites": True, "confirm": True,
                               "reason": "live smoke: proving the macro gate holds"})
    check("swx_run_macro never runs a macro this suite did not put there",
          r.get("success") is False, r)
    if WRITES_ENABLED:
        check("(writes ENABLED) the gate opens and the existence check stops it",
              r.get("errorCode") == "macro_not_found", r)
    else:
        check("swx_run_macro refuses in the default posture — even fully 'confirmed' by the caller",
              r.get("errorCode") == "user_writes_disabled", r)
        check("...and the refusal is about arbitrary code, not a document write",
              "arbitrary code" in json.dumps(r), r)
        check("...and it names the process-level opt-in it would need",
              (r.get("requires") or {}).get("env") == "SWX_ALLOW_USER_WRITES=1", r)

    # The unattributed-window refusal — a real hwnd from the desktop, never ours.
    r = call("swx_close_window", {"hwnd": 999999, "reason": "live smoke: proving the refusal",
                                  "confirm": True})
    check("swx_close_window refuses an hwnd it cannot attribute to SOLIDWORKS",
          r.get("errorCode") == "window_not_attributed", r)
    if AB_AVAILABLE:
        r = call("swx_dismiss_blocking_dialogs", {})
        check("swx_dismiss_blocking_dialogs reports and does not act",
              r.get("success") is True and r.get("closed") == [], r)
    else:
        check("(skipped) swx_dismiss_blocking_dialogs — cannot test without an ab-spawned bridge",
              True, "abAvailable=false")


def phase_ownership_refusal():
    """Prove the ownership refusal WITHOUT aiming a mutating verb at the user's
    document.

    Two earlier versions of this got it wrong. The first fired three mutating
    verbs at whatever Drew had open. The second claimed to be opt-in but only
    gated the run_script variant — the default still sent a real `swx_save_as`
    at `DOCS.baseline[0]["title"]`, through `call()` rather than `mutate()`, on
    the premise that "a refusal is a read". That premise was false: a title
    collision could make the gate approve the bridge's document and the com host
    rename the user's.

    So the default proves the refusal against documents the suite CREATED, plus
    error paths that name nothing real. `--touch-user-doc` opts in to naming the
    user's document, and even then only in a verb whose refusal is the assertion.
    """
    # A second scratch document, saved, so it has a path we can name precisely.
    r = call("swx_new_document", {"type": "part"})
    if not r.get("success"):
        check("a scratch document to prove the ownership refusal against", False, r.get("error"))
        return
    other = (r.get("created") or {}).get("title")
    DOCS.own(other)

    # An error path that names nothing real: no document can be harmed.
    r = call("swx_save_as", {"outputPath": DIR + "\\notopen.sldprt",
                             "targetDocument": "NoSuchDocument.sldprt"})
    check("a targetDocument that names nothing open is an ERROR, not a fall back to the active doc",
          r.get("errorCode") == "document_not_open", r)
    r = call("swx_save_as", {"outputPath": DIR + "\\amb.sldprt",
                             "targetDocument": other, "path": "C:\\nowhere\\else.sldprt"})
    check("naming a document two ways that disagree is refused, not guessed",
          r.get("errorCode") in ("ambiguous_target", "document_not_open", "no_target_document"), r)

    if not (TOUCH_USER_DOC and DOCS.baseline):
        check("(default) the ownership refusal was proved without naming the user's document", True,
              "pass --touch-user-doc to also assert the refusal against their open document")
        return

    # OPT-IN ONLY. Even here the assertion IS the refusal: if it does not
    # refuse, that is the finding, and nothing else is sent.
    victim = DOCS.baseline[0]
    target = victim.get("path") or victim.get("title")
    key = "path" if victim.get("path") else "targetDocument"
    r = call("swx_save_as", {"outputPath": DIR + "\\stolen.sldprt", key: target})
    check("swx_save_as REFUSES to rename a document the bridge did not create (--touch-user-doc)",
          r.get("errorCode") in ("user_writes_disabled", "not_bridge_document"), r)
    after = call("swx_docs").get("docs") or []
    check("...and the user's document is IDENTICAL — same title AND same path",
          any(doc_key(d) == doc_key(victim) for d in after),
          "wanted %s, got %s" % (doc_key(victim), [doc_key(d) for d in after]))


def phase_unproven(title):
    """The signatures that have never executed. Failures here are FINDINGS."""
    res = batch([
        {"label": "saveas", "verb": "swx_save_as",
         "args": {"outputPath": DIR + "\\run4.sldprt", "targetDocument": title}},
        {"label": "docs", "verb": "swx_docs", "args": {}},
    ], 600, 200)
    sa = res.get("saveas", {})
    check("swx_save_as wrote the part (SaveAs3 plain-Silent path)", sa.get("success") is True,
          sa.get("error") or sa)
    if sa.get("success"):
        record_written("run4.sldprt")
    saved_title = (sa.get("doc") or {}).get("title")
    if sa.get("success") and saved_title:
        # A save-as RETITLES the document AND gives it a path: track both, or
        # cleanup misses it — and the path is what makes the close unambiguous.
        DOCS.disown(title)
        DOCS.own(saved_title, (sa.get("doc") or {}).get("path") or (DIR + "\\run4.sldprt"))
        check("save_as retitled our document, as a real Save As does", saved_title != title, saved_title)

    activate_target = saved_title or title
    r = call("swx_activate_document", {"name": activate_target})
    check("ActivateDoc3 (never executed before) activates by title", r.get("success") is True,
          r.get("error") or r)

    n = call("swx_new_document", {"type": "assembly"})
    if n.get("success"):
        atitle = (n.get("created") or {}).get("title")
        DOCS.own(atitle)
        check("new_document makes an assembly, and reports its REAL untitled name",
              str(atitle).startswith("Assem"), atitle)
    else:
        check("new_document can make an assembly", False, n.get("error"))

    n = call("swx_new_document", {"type": "drawing"})
    if n.get("success"):
        dtitle = (n.get("created") or {}).get("title")
        DOCS.own(dtitle)
        check("new_document makes a drawing", bool(dtitle), dtitle)
        # PINNED: the one call in the suite still aimed at whatever was ACTIVE.
        # Worst case a zoom-to-fit on Drew's own drawing, but there is no reason
        # for it to be the exception.
        assert_not_users("before show_drawing")
        r = call("swx_show_drawing", {"targetDocument": dtitle})
        check("swx_show_drawing activates and fits a drawing", r.get("success") is True, r.get("error") or r)
    else:
        check("new_document can make a drawing", False, n.get("error"))

    r = call("swx_activate_document", {"name": activate_target})
    check("re-activating our part before the DSL probes", r.get("success") is True, r.get("error") or r)

    # FeatureCut4 and CreateCircleByRadius — both shipped never having run.
    r = mutate("swx_run_script", {"targetDocument": activate_target, "script": [
        {"op": "sketch_circle", "plane": "Top Plane", "radiusMm": 5},
        {"op": "cut", "depthMm": 5},
    ]})
    check("CreateCircleByRadius + FeatureCut4 (both unproven) run", r.get("success") is True,
          r.get("error") or r)

    # A DELIBERATE mid-script failure must not leave SOLIDWORKS in sketch mode.
    r = mutate("swx_run_script", {"targetDocument": activate_target, "script": [
        {"op": "sketch_rect", "plane": "No Such Plane", "widthMm": 10, "heightMm": 10},
        {"op": "extrude", "depthMm": 5},
    ]})
    check("a script naming a plane that does not exist fails cleanly", r.get("success") is False, r)
    probe = call("swx_inspect_bodies", {})
    check("...and SOLIDWORKS is NOT left in sketch-edit mode (the next verb works)",
          probe.get("success") is True, probe.get("error") or probe)

    r = call("swx_get_preferences", {"targetDocument": activate_target})
    check("swx_get_preferences answers", r.get("success") is True, r.get("error") or r)
    cur = (r.get("current") or {}).get("lengthUnitRaw") or {}
    before = cur.get("value")
    if isinstance(before, int):
        # PINNED to our own document, and through `mutate()`: this writes a
        # document-scoped setting and dirties the document, so aiming it at
        # whatever happened to be ACTIVE was the same defect as rebuild_check.
        # The restore is registered BEFORE the change and is pinned too, so an
        # abort or a dropped relay cannot leave the user's units flipped.
        RESTORE.append(lambda: call("swx_set_preference", {
            "name": "lengthUnitRaw", "value": before, "targetDocument": activate_target}))
        w = mutate("swx_set_preference",
                   {"name": "lengthUnitRaw", "value": 0 if before != 0 else 1,
                    "targetDocument": activate_target},
                   where="preferences")
        check("SetUserPreferenceInteger (unproven) writes AND verifies by reading back",
              w.get("verified") is True, w)
    else:
        check("a preference value to round-trip", False, r.get("current"))

    if AB_AVAILABLE:
        badge_before = call("swx_set_driving_badge", {}).get("showDrivingBadge")
        RESTORE.append(lambda: call("swx_set_driving_badge", {"enabled": bool(badge_before)}))
        r = call("swx_set_driving_badge", {"enabled": True})
        check("desktop_taskbar badge (unproven) applies to the SOLIDWORKS window",
              r.get("applied") is True, r.get("ab"))
    else:
        check("(skipped) desktop_taskbar badge — cannot test without an ab-spawned bridge", True,
              "abAvailable=false")

    # LoadFile4/LoadFile2 must produce EXACTLY ONE document.
    d0 = len(call("swx_docs").get("docs") or [])
    # Import a STEP under a DIFFERENT stem: SOLIDWORKS titles an imported
    # document `<stem>.SLDPRT`, so importing `smoke.step` beside our own
    # `smoke.SLDPRT` made the bridge manufacture a title collision and then
    # (correctly) refuse to close either by name.
    imp = call("swx_import", {"filePath": DIR + "\\import4.step"})
    d1 = len(call("swx_docs").get("docs") or [])
    if imp.get("success"):
        DOCS.own((imp.get("imported") or {}).get("title"))
        check("swx_import opened EXACTLY ONE document (not two)", d1 - d0 == 1, "%s -> %s" % (d0, d1))
        check("...and says which route it took", isinstance(imp.get("via"), str), imp.get("via"))
    else:
        check("swx_import imports a STEP file", False, imp.get("error"))


def phase_readonly(sw):
    res = batch([
        {"label": "describe", "verb": "swx_describe", "args": {}},
        {"label": "readiness", "verb": "swx_readiness", "args": {}},
        {"label": "state", "verb": "swx_state", "args": {"screenshot": False}},
        {"label": "window_info", "verb": "swx_window_info", "args": {}},
        {"label": "recovery", "verb": "swx_check_recovery", "args": {}},
        {"label": "cloud", "verb": "swx_cloud_status", "args": {}},
        {"label": "times", "verb": "swx_verb_times", "args": {}},
        {"label": "ledger", "verb": "swx_action_ledger", "args": {}},
        {"label": "progress", "verb": "swx_progress", "args": {}},
    ], 600, 150)
    check("swx_describe lists the whole v0.3 surface",
          len(res.get("describe", {}).get("verbs") or []) >= 60, len(res.get("describe", {}).get("verbs") or []))
    check("swx_readiness is ready", res.get("readiness", {}).get("ready") is True, res.get("readiness"))
    check("swx_state reports SOLIDWORKS' own window handle",
          isinstance(res.get("state", {}).get("mainHwnd"), int), res.get("state", {}).get("mainHwnd"))
    wi = res.get("window_info", {})
    if AB_AVAILABLE:
        check("swx_window_info finds the SOLIDWORKS main window",
              any(w.get("isMain") for w in (wi.get("windows") or [])),
              [w.get("title") for w in (wi.get("windows") or [])][:5])
        check("...and marks unattributable dialogs as such",
              all(isinstance(w.get("ownerConfirmed"), bool) for w in (wi.get("windows") or [])),
              wi.get("windows"))
    else:
        check("(skipped) swx_window_info — cannot test without an ab-spawned bridge", True,
              "abAvailable=false")
        check("swx_window_info reports the list as UNKNOWN, not empty",
              wi.get("windows") is None and wi.get("abAvailable") is False, wi)
    check("swx_check_recovery answers", res.get("recovery", {}).get("success") is True, res.get("recovery"))
    check("swx_cloud_status runs the probe that proves there is no COM cloud route",
          res.get("cloud", {}).get("comApiExposesCloud") is False, res.get("cloud"))
    times = res.get("times", {}).get("stages") or {}
    check("swx_verb_times has measured this machine", len(times) > 0, list(times)[:6])
    budgets = {"swx_run_script": 420, "swx_new_document": 180, "swx_import": 420, "swx_save_as": 420}
    for verb, budget in budgets.items():
        p90 = (times.get(verb) or {}).get("p90")
        if p90 is not None:
            check("%s p90 (%.1fs) fits its bridge.json budget of %ds" % (verb, p90, budget),
                  p90 < budget, p90)
    check("swx_action_ledger recorded the document the bridge created",
          any(a.get("action") == "document_created" for a in (res.get("ledger", {}).get("actions") or [])),
          len(res.get("ledger", {}).get("actions") or []))
    check("swx_progress is idle", res.get("progress", {}).get("active") == [], res.get("progress"))


def invariant_gate():
    """THE gate: SOLIDWORKS must end with exactly the documents it started with.

    Run FIRST in cleanup and in its own try, because it used to be the LAST
    statement after `listing()` and a dozen `shell()` round trips — any one of
    which could raise and have the whole thing swallowed by the outer handler,
    so the runs that most needed the answer were the runs that never got it.

    SKIPS (does not fail) when no baseline was ever captured: a preflight abort
    fires before `DOCS.baseline` is set, and a spurious failure there teaches
    the reader to ignore the one check that matters.
    """
    if not DOCS.baseline_captured:
        check("(skipped) document-set invariant — the run aborted before a baseline was taken",
              True, "nothing was touched, so there is nothing to compare")
        return
    final = call("swx_docs").get("docs") or []
    before = sorted(doc_key(d) for d in DOCS.baseline)
    after = sorted(doc_key(d) for d in final)
    check("SOLIDWORKS was left with EXACTLY the documents it started with", before == after,
          "before=%s after=%s" % (before, after))


def cleanup():
    """Close what we made, delete only what we wrote, and gate on the invariant.

    Runs on EVERY exit path, including an abort.
    """
    print("-- cleanup --")
    # Close ours first so the invariant has a fair chance, but the gate runs
    # whether or not this succeeds — and it runs in its OWN try, immediately
    # after, not at the end behind a dozen round trips that can throw past it.
    # Iterating DOCS.mine means every title here is one the suite created, so
    # this goes through `call` rather than `mutate`: `mutate` would raise on
    # failure, and cleanup must keep going and reach the gate.
    orphans = []
    for title, path in list(DOCS.mine):
        args = {"save": False}
        if path:
            args["path"] = path
        else:
            args["name"] = title
        r = call("swx_close_doc", args)
        if not r.get("success") and not path:
            # RETRY BY THE TITLE SOLIDWORKS IS ACTUALLY REPORTING. The live run
            # hit this: a document the suite had created could not be closed by
            # the handle it was created with, and the loop simply gave up —
            # leaving an orphan on Drew's machine with nothing said about it.
            live = [d.get("title") for d in (call("swx_docs").get("docs") or [])]
            for candidate in live:
                if candidate and candidate == title:
                    r = call("swx_close_doc", {"name": candidate, "save": False})
                    break
        if r.get("success"):
            DOCS.disown(title)
        else:
            orphans.append((title, path, (r.get("error") or "")[:160]))
            check("closed our scratch document '%s'" % title, False, r.get("error"))
    r = call("swx_reset_workspace", {"force": True})
    check("swx_reset_workspace ran", r.get("success") is True, r.get("error") or r)
    # Anything still ours after the sweep is an ORPHAN, and saying so plainly is
    # the whole point: an unclosable scratch document left on the user's machine
    # is a fact they need, not a line buried in a failure list.
    still_ours = [t for (t, _) in DOCS.mine]
    if orphans or still_ours:
        names = sorted({t for (t, _, _) in orphans} | set(still_ours))
        check("NO orphan documents were left on the machine", False,
              "THIS SUITE LEFT %d DOCUMENT(S) OPEN IN SOLIDWORKS THAT IT COULD NOT CLOSE: %s. "
              "They are the suite's own scratch documents, not the user's, and they are unsaved "
              "— closing them by hand in SOLIDWORKS discards nothing of theirs. Reasons: %s"
              % (len(names), ", ".join(names),
                 "; ".join("%s: %s" % (t, e) for (t, _, e) in orphans) or "closed but still tracked"))


    # THE GATE, in its own try, before anything else that can throw.
    try:
        invariant_gate()
    except BaseException as e:
        check("the document-set invariant was evaluated", False, "%s: %s" % (type(e).__name__, e))

    # Restore anything the run changed on the user's machine, whatever happened.
    for restore in list(RESTORE):
        try:
            restore()
        except BaseException as e:
            check("restored a setting the suite changed", False, "%s: %s" % (type(e).__name__, e))

    # LANDMINES. These names are only ever used by probes that must be REFUSED,
    # so their existence means a guard failed — and `stolen.sldprt` would then
    # BE the user's relocated document. The previous version merely printed a
    # failure and carried on into the deletion eight lines later.
    left = listing()
    landmines = [n for n in ("stolen.sldprt", "bypass.wrl", "notopen.sldprt", "amb.sldprt",
                             "renamed.sldprt", "x.sldprt") if n in left]
    if landmines:
        for n in landmines:
            check("NO %s was written — its existence means a guard failed" % n, False,
                  "%s exists (%s bytes)." % (n, left[n]))
        check("DELETION HALTED — nothing in %s will be removed" % DIR, False,
              "One of these is written only when a refusal did not happen, and stolen.sldprt "
              "would be the USER'S relocated document. Inspect %s by hand before touching it."
              % DIR)
        return
    if not KEEP:
        for name in sorted(set(FILES_WRITTEN) - PRE_EXISTING):
            shell('powershell -NoProfile -Command "Remove-Item -LiteralPath \'%s\\%s\' -Force '
                  '-ErrorAction SilentlyContinue"' % (DIR, name),
                  "Deleting one file the SOLIDWORKS live smoke test wrote")
        # Only remove the directory once we have CONFIRMED it is empty — rather
        # than relying on an untested assumption about how -Force behaves on a
        # non-empty directory.
        remaining = listing()
        check("every file this suite wrote was removed", not remaining, sorted(remaining))
        if not remaining:
            shell('powershell -NoProfile -Command "Remove-Item -LiteralPath \'%s\' -Force '
                  '-ErrorAction SilentlyContinue"' % DIR,
                  "Removing the now-empty scratch folder the SOLIDWORKS live smoke test created")

    st = call("swx_status")
    check("SOLIDWORKS is STILL RUNNING after the suite",
          (st.get("solidworks") or {}).get("running") is True, st.get("solidworks"))


_CLEANED = []


def cleanup_once():
    """Cleanup runs exactly once, however the process is ending."""
    if _CLEANED:
        return
    _CLEANED.append(True)
    try:
        cleanup()
    except BaseException as e:
        check("cleanup completed", False, "%s: %s" % (type(e).__name__, e))


def _on_signal(signum, _frame):
    print("\n  signal %s — running cleanup before exit\n" % signum)
    cleanup_once()
    finish()
    os._exit(1)


def main():
    print("== swx live smoke ==")
    # A Ctrl-C or a SIGTERM must not skip the invariant and the restores; the
    # `finally` alone does not cover being killed.
    atexit.register(cleanup_once)
    for sig in (signal.SIGTERM, signal.SIGINT):
        try:
            signal.signal(sig, _on_signal)
        except (ValueError, OSError):
            pass
    aborted = None
    try:
        sw = preflight()
        baseline = call("swx_docs").get("docs") or []
        DOCS.baseline = baseline
        DOCS.baseline_captured = True
        print("  ..  baseline open documents: %s" % ([d.get("title") for d in baseline] or "none"))
        phase_start_attaches_and_never_relaunches(sw)
        title, dim = phase_build()
        phase_drive(title, dim)
        phase_export(title)
        phase_refusals(title)
        phase_ownership_refusal()
        phase_unproven(title)
        phase_readonly(sw)
    except Abort as e:
        aborted = str(e)
        print("\n  ABORTED: %s\n" % aborted)
    except BaseException as e:  # SystemExit and KeyboardInterrupt INCLUDED
        # The invariant gate is the whole point of this file, and it lives in
        # the `finally` below. A SystemExit raised anywhere in the run — the
        # batch runner used to raise one — would otherwise sail past it, so the
        # only run where "did we leave SOLIDWORKS as we found it?" goes
        # unanswered is the run that went wrong.
        aborted = "%s: %s" % (type(e).__name__, e)
        print("\n  ABORTED (%s)\n" % aborted)
    finally:
        cleanup_once()

    passed = sum(1 for _, ok, _ in CHECKS if ok)
    failed = [c for c in CHECKS if not c[1]]
    print("")
    print("=" * 66)
    print("LIVE SMOKE: %d passed, %d failed, %d total" % (passed, len(failed), len(CHECKS)))
    if aborted:
        print("RUN ABORTED: %s" % aborted)
    for name, _, detail in failed:
        print("  FAIL  %s\n          <- %s" % (name, detail))
    print("=" * 66)
    return 1 if (failed or aborted) else 0


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