Project-SDE-WP-Suite/tests/files_check.py

#!/usr/bin/env python3
"""Do drawings travel with the package, and open offline? — CR-007 / D8, T7.7.

The field wants the specific sheet attached, not a link to a Bluebeam session.
D8 set the numbers: 5MB a file, PDFs and images, stored in the SAME database
(a backup that excludes the drawings cannot be restored from), 2GB a project
with a warning at 80%. Offline caching follows ASSIGNMENT, not project: a
package assigned to someone else is deliberately not cached.

The ceiling is driven with WP_FILE_PROJECT_CEILING so this probe can hit 80%
and 100% without writing two gigabytes; the shipped default (asserted here by
reading the source) is the 2GB decision.

Self-contained: throwaway SQLite, its own uvicorn, headless Edge or Chrome
with a real service worker and CDP network-offline emulation.
Exit 0 all passed, 1 a failure, 2 could not run.
"""
import base64
import json
import os
import re
import sys
import tempfile
import time
import urllib.error
import urllib.request

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

# The ceiling override MUST be in the environment before the server process
# starts - it is read at import time, which is the point: it is a deploy-time
# number, not a runtime mutable.
CEILING = 200_000
os.environ["WP_FILE_PROJECT_CEILING"] = str(CEILING)

import cdp                                                           # noqa: E402
from browser_check import seed, start_server, chk, _PASS, _FAIL      # noqa: E402
from sections_check import set_sop                                   # noqa: E402
from stepper_check import dismiss_dialogs                            # noqa: E402

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

PDF_BYTES = b"%PDF-1.4\n1 0 obj<</Type/Catalog>>endobj\ntrailer<<>>\n%%EOF\n" * 20
PNG_BYTES = base64.b64decode(
    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk"
    "+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==")


def ascii_(v, n=300):
    return re.sub(r"\s+", " ", str(v)).encode("ascii", "replace").decode()[:n]


def settle(seconds=0.6):
    time.sleep(seconds)


def api(base, path, token, method="GET", body=None, raw=False):
    req = urllib.request.Request(base + path, method=method)
    req.add_header("Cookie", "wp_session=" + token)
    data = None
    if body is not None:
        data = json.dumps(body).encode()
        req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, data, timeout=30) as r:
            payload = r.read()
            return r.status, (payload if raw else json.loads(payload.decode() or "null"))
    except urllib.error.HTTPError as e:
        try:
            return e.code, json.loads(e.read().decode() or "null")
        except Exception:
            return e.code, None


def upload(base, tok, wp_id, name, mime, blob, description=""):
    return api(base, "/api/wps/%s/files" % wp_id, tok, "POST", {
        "name": name, "mime": mime, "description": description,
        "data_base64": base64.b64encode(blob).decode()})


def mkwp(base, tok, wp_id, assignee=None):
    return api(base, "/api/wps", tok, "POST", {
        "id": wp_id, "project_id": "projA", "number": "F-" + wp_id[-2:],
        "subject": "drawings host", "status": "In Progress", "assignee_id": assignee,
        "data": {"constraints": [{"name": "Materials", "status": "cleared", "comment": ""}],
                 "attachments": [{"doc": "E-101", "rev": "2", "link": "https://example.test/e101"}]}})


def wait_for(fn, timeout=15.0):
    end = time.time() + timeout
    while time.time() < end:
        try:
            if fn():
                return True
        except Exception:
            pass
        time.sleep(0.4)
    return False


def main():
    exe = cdp.find_browser()
    if not exe:
        print("no headless-capable browser found; set WP_BROWSER.")
        return 2

    tmpdir = tempfile.mkdtemp(prefix="wpsuite-files-")
    db_path = os.path.join(tmpdir, "check.db")
    server = None
    browser = None
    try:
        tok = seed(db_path)
        set_sop(db_path, {})
        port = cdp.free_port()
        base = "http://127.0.0.1:%d" % port
        server = start_server(port, db_path)
        root = tok["root"]

        # ── 1. upload and retrieval ───────────────────────────────────────────
        print("\n1. upload, retrieve, describe, delete")
        mkwp(base, root, "wpF1", assignee="user_root")
        code, out = upload(base, root, "wpF1", "tray-L3.pdf", "application/pdf",
                           PDF_BYTES, "Tray section, Level 3 east only")
        chk("a PDF uploads", code == 200, ascii_((code, out)))
        pdf_id = (out or {}).get("file", {}).get("id", "")
        code, blob = api(base, "/api/files/" + pdf_id, root, raw=True)
        chk("...and comes back byte-for-byte", code == 200 and blob == PDF_BYTES,
            (code, len(blob or b"")))
        code, out = upload(base, root, "wpF1", "sector-p.png", "image/png", PNG_BYTES,
                           "Highlighted sector P")
        chk("an image uploads", code == 200, code)
        png_id = (out or {}).get("file", {}).get("id", "")
        code, blob = api(base, "/api/files/" + png_id, root, raw=True)
        chk("...and comes back byte-for-byte", code == 200 and blob == PNG_BYTES, code)

        _, wp = api(base, "/api/wps/wpF1", root)
        files = (wp.get("data") or {}).get("files") or []
        chk("the package record mirrors the file list, descriptions included",
            len(files) == 2 and files[0].get("description") == "Tray section, Level 3 east only",
            ascii_(files))
        chk("the link attachments are still on the package, beside the uploads",
            (wp.get("data") or {}).get("attachments", [{}])[0].get("link")
            == "https://example.test/e101")

        code, _ = api(base, "/api/files/" + png_id, root, "PATCH",
                      {"description": "Sector P, north wall only"})
        _, listing = api(base, "/api/wps/wpF1/files", root)
        chk("a description edit persists",
            code == 200 and any(f.get("description") == "Sector P, north wall only"
                                for f in listing.get("files", [])), ascii_(listing))

        # The upsert cannot clobber the server-owned list with a stale client copy.
        api(base, "/api/wps", root, "POST", {
            "id": "wpF1", "project_id": "projA", "number": wp["number"],
            "subject": wp["subject"], "status": wp["status"],
            "data": {k: v for k, v in (wp.get("data") or {}).items() if k != "files"}})
        _, wp2 = api(base, "/api/wps/wpF1", root)
        chk("a client save WITHOUT the file list does not erase it (server-owned key)",
            len((wp2.get("data") or {}).get("files") or []) == 2,
            ascii_((wp2.get("data") or {}).get("files")))

        # ── 2. refusals, server-side ──────────────────────────────────────────
        print("\n2. the server refuses what the browser refuses")
        code, out = upload(base, root, "wpF1", "notes.txt", "text/plain", b"hello")
        chk("a type outside PDF/image is refused, naming what IS accepted",
            code == 400 and "PDF and image" in str(out), ascii_((code, out)))
        big = b"x" * (5 * 1024 * 1024 + 1)
        code, out = upload(base, root, "wpF1", "big.pdf", "application/pdf", big)
        chk("a file over 5MB is refused, naming the limit",
            code == 413 and "5MB" in str(out), ascii_((code, out)))
        code, _ = upload(base, tok["bob"], "wpF1", "x.pdf", "application/pdf", PDF_BYTES)
        chk("someone outside the project cannot upload", code == 403, code)
        code, _ = api(base, "/api/files/" + pdf_id, tok["bob"])
        chk("...or fetch", code == 403, code)

        # ── 3. the ceiling ────────────────────────────────────────────────────
        print("\n3. the 2GB ceiling (driven at %d bytes)" % CEILING)
        src = open(os.path.join(ROOT, "server", "app.py"), encoding="utf-8").read()
        chk("the shipped default IS the decision: 2GB, in the source",
            "2 * 1024 * 1024 * 1024" in src)
        filler = b"f" * 165_000            # past 80% of the 200KB ceiling
        code, out = upload(base, root, "wpF1", "fill.pdf", "application/pdf", filler)
        chk("a large upload under the ceiling lands", code == 200, ascii_((code, out)))
        chk("...and the response flags the 80% warning",
            (out or {}).get("warn") is True, ascii_(out))
        used_before = (out or {}).get("used", 0)
        _, st = api(base, "/api/projects/projA/storage", root)
        chk("the storage endpoint reports the running total",
            st.get("used", 0) == used_before and st.get("ceiling") == CEILING, ascii_(st))
        code, out = upload(base, root, "wpF1", "over.pdf", "application/pdf", b"y" * 100_000)
        chk("at the ceiling the upload is refused, naming it",
            code == 413 and str(CEILING) in str(out) and "full" in str(out),
            ascii_((code, out)))
        code, out = api(base, "/api/files/" + pdf_id, root, "DELETE")
        chk("deleting a drawing frees its bytes from the total",
            code == 200 and out.get("used", 10**9) < used_before, ascii_(out))

        # a second package, assigned to someone ELSE, with its own drawing - the
        # offline check needs a file that must NOT be cached.
        mkwp(base, root, "wpF2", assignee="user_sue")
        code, out = upload(base, root, "wpF2", "other.png", "image/png", PNG_BYTES,
                           "someone else's sheet")
        other_id = (out or {}).get("file", {}).get("id", "")

        # ── 4. the creator: limits first, meter always, refusal costs nothing ─
        print("\n4. the creator at 1440px")
        browser = cdp.Browser(exe)
        page = browser.page()
        page.clear_cookies()
        page.set_cookie("wp_session", root)
        page.viewport(1440, 900)
        page.goto(base + "/wp-creation-index.html?project=projA&wp=wpF1")
        dismiss_dialogs(page)
        chk("the creator boots on the package",
            wait_for(lambda: page.eval("!!window.wpCreatorReady")), )
        settle(1.6)
        page.eval("window.prompt=()=>null; window.alert=()=>{}; window.confirm=()=>true;")
        rules = page.eval("(document.getElementById('file-rules')||{textContent:''}).textContent")
        chk("the limits are stated BEFORE upload: 5MB and PDF/image, in the card",
            "5MB" in rules and "PDF" in rules, ascii_(rules))
        chk("the running project total is visible where uploads happen",
            "Project storage:" in rules, ascii_(rules))
        chk("the uploaded drawings are listed with their descriptions",
            page.eval("document.querySelectorAll('#wp-file-list .wp-file-item').length") >= 1)

        page.eval("""window.__fetchCalls=[]; window.__origFetch=window.fetch;
          window.fetch=function(u,o){ window.__fetchCalls.push(String(u)); return window.__origFetch(u,o); };""")
        page.eval("""wpFileUpload({target:{files:[
          new File([new Uint8Array(6*1024*1024)], 'big.pdf', {type:'application/pdf'})], value:''}})""")
        settle(0.8)
        toast_txt = page.eval("(document.getElementById('toast')||{textContent:''}).textContent")
        chk("an oversize file is refused BEFORE upload, naming the limit",
            "5MB" in toast_txt, ascii_(toast_txt))
        chk("...and no upload request ever left the browser",
            page.eval("!window.__fetchCalls.some(u=>u.includes('/files'))"))
        page.eval("""wpFileUpload({target:{files:[
          new File(['zzz'], 'macro.docx', {type:'application/vnd.ms-word'})], value:''}})""")
        settle(0.8)
        toast_txt = page.eval("(document.getElementById('toast')||{textContent:''}).textContent")
        chk("a type outside PDF/image is refused before upload, naming the accepted types",
            "PDF and image" in toast_txt, ascii_(toast_txt))

        page.eval("""wpFileUpload({target:{files:[
          new File([new Uint8Array([137,80,78,71])], 'site.png', {type:'image/png'})], value:''}})""")
        chk("a real upload through the form lands and appears in the list",
            wait_for(lambda: page.eval(
                "[...document.querySelectorAll('#wp-file-list a')].some(a=>a.textContent.includes('site.png'))")))

        page.eval("renderPackage(collectPackage())")
        settle(0.8)
        doc = page.eval("(document.getElementById('pkg-doc')||{innerHTML:''}).innerHTML")
        chk("the export carries the uploads: name, description, and the image inline",
            "sector-p.png" in doc and "Sector P, north wall only" in doc
            and "/api/files/" in doc and "<img" in doc, ascii_(doc, 200))
        chk("...and still carries the plain link attachments",
            "E-101" in doc and "example.test/e101" in doc)

        page.close()

        # ── 5. the field view at 390px, then the network goes away ───────────
        print("\n5. offline on the tablet")
        page = browser.page()
        page.clear_cookies()
        page.set_cookie("wp_session", root)
        page.viewport(390, 844, mobile=True)
        page.goto(base + "/field.html?project=projA")
        dismiss_dialogs(page)
        settle(2.0)
        chk("the service worker takes the page",
            wait_for(lambda: page.eval("!!(navigator.serviceWorker&&navigator.serviceWorker.controller)"), 20))
        # the prefetch runs after the pull; give it a beat, then check the cache
        chk("the drawings of MY assigned package are cached",
            wait_for(lambda: page.eval(
                "caches.open('wp-suite-drawings-v1').then(c=>c.keys()).then(ks=>ks.some(k=>k.url.includes('%s')))" % png_id), 20))
        chk("a package assigned to someone ELSE is not cached (D8)",
            page.eval("caches.open('wp-suite-drawings-v1').then(c=>c.keys())"
                      ".then(ks=>!ks.some(k=>k.url.includes('%s')))" % other_id))

        # the drawing row itself, on the phone-size detail
        page.eval("openWP('wpF1')")
        settle(0.8)
        row = json.loads(page.eval("""JSON.stringify((() => {
          const a = document.querySelector('.fld-drawing');
          if (!a) return null;
          const r = a.getBoundingClientRect();
          return {text: a.textContent, h: r.height, w: r.width, within: r.right <= 390};
        })())"""))
        chk("390px: the drawing row renders on the package, description included",
            row is not None and "Sector P" in (row["text"] or ""), ascii_(row))
        chk("390px: it is a 44px touch target that stays inside the screen",
            row and row["h"] >= 44 and row["within"], ascii_(row))

        js_errors = [e for e in page.js_errors()]
        chk("no JavaScript errors anywhere in this run", not js_errors,
            ascii_(js_errors[:2]))

        # ── 6. the network actually goes away ────────────────────────────────
        # CDP's emulateNetworkConditions only throttles the PAGE's session; the
        # service worker fetches on its own target and sails straight past it -
        # which made the first version of this check pass for the wrong reason.
        # Killing the server is offline nobody can argue with.
        print(chr(10) + "6. the server is gone")
        server.terminate()
        server = None
        settle(1.5)
        chk("offline: my assigned package's drawing still opens (from the SW cache)",
            page.eval("fetch('/api/files/%s').then(r=>r.ok).catch(()=>false)" % png_id) is True)
        chk("offline: the other package's drawing does not (assignment-scoped, D8)",
            page.eval("fetch('/api/files/%s').then(r=>r.ok).catch(()=>false)" % other_id) is False)

    finally:
        if browser is not None:
            try:
                browser.close()
            except Exception:
                pass
        if server is not None:
            try:
                server.terminate()
            except Exception:
                pass

    print("\n" + "-" * 54)
    print("%d/%d checks passed." % (len(_PASS), len(_PASS) + len(_FAIL)))
    for f in _FAIL:
        print("  - " + f)
    return 1 if _FAIL else 0


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