"""Chromium layout regression for the B-44 admin balance directory. Measures 15-digit period-end amounts against the unresolved chip using getBoundingClientRect — not static text presence. """ from __future__ import annotations import base64 import hashlib import json import os from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import socket import subprocess import tempfile import threading import time from urllib.parse import urlparse from urllib.request import urlopen import unittest import sys ROOT = Path(__file__).resolve().parents[1] if str(Path(__file__).resolve().parent) not in sys.path: sys.path.insert(0, str(Path(__file__).resolve().parent)) FIXTURE = "/tests/fixtures/b44-balance-directory.html" CHROME_CANDIDATES = [ Path(os.environ.get("PROGRAMFILES", r"C:\Program Files")) / "Google/Chrome/Application/chrome.exe", Path(os.environ.get("PROGRAMFILES(X86)", r"C:\Program Files (x86)")) / "Microsoft/Edge/Application/msedge.exe", ] VIEWPORTS = (720, 768, 800, 900, 375, 1024, 1440) def chrome_bin() -> Path: for path in CHROME_CANDIDATES: if path.exists(): return path raise FileNotFoundError("Chrome/Edge not found for layout regression") class QuietHandler(SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs) -> None: super().__init__(*args, directory=str(ROOT), **kwargs) def log_message(self, *_args) -> None: pass def handle(self) -> None: try: super().handle() except (ConnectionResetError, BrokenPipeError, TimeoutError): pass class _Cdp: def __init__(self, ws_url: str) -> None: parsed = urlparse(ws_url) self._sock = socket.create_connection((parsed.hostname, parsed.port or 80), timeout=10) self._sock.settimeout(10) key = base64.b64encode(os.urandom(16)).decode() path = parsed.path + (f"?{parsed.query}" if parsed.query else "") self._sock.sendall( ( f"GET {path} HTTP/1.1\r\n" f"Host: {parsed.netloc}\r\n" "Upgrade: websocket\r\n" "Connection: Upgrade\r\n" f"Sec-WebSocket-Key: {key}\r\n" "Sec-WebSocket-Version: 13\r\n\r\n" ).encode() ) header = b"" while b"\r\n\r\n" not in header: chunk = self._sock.recv(4096) if not chunk: raise ConnectionError("CDP websocket handshake failed") header += chunk expected = base64.b64encode( hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()).digest() ).decode() if expected not in header.decode("latin1"): raise ConnectionError("CDP websocket accept mismatch") leftover = header.split(b"\r\n\r\n", 1)[1] self._buf = leftover self._next_id = 1 def call(self, method: str, params: dict | None = None, timeout: float = 15) -> dict: msg_id = self._next_id self._next_id += 1 self._send({"id": msg_id, "method": method, "params": params or {}}) deadline = time.time() + timeout while time.time() < deadline: payload = self._recv() if payload.get("id") == msg_id: if "error" in payload: raise RuntimeError(f"{method}: {payload['error']}") return payload.get("result") or {} raise TimeoutError(method) def wait_event(self, name: str, timeout: float = 15) -> dict: deadline = time.time() + timeout while time.time() < deadline: payload = self._recv() if payload.get("method") == name: return payload.get("params") or {} raise TimeoutError(name) def close(self) -> None: try: self._sock.close() except OSError: pass def _send(self, obj: dict) -> None: data = json.dumps(obj, separators=(",", ":")).encode() mask = os.urandom(4) header = bytearray([0x81]) length = len(data) if length < 126: header.append(0x80 | length) elif length < 65536: header.append(0x80 | 126) header.extend(length.to_bytes(2, "big")) else: header.append(0x80 | 127) header.extend(length.to_bytes(8, "big")) header.extend(mask) masked = bytes(b ^ mask[i % 4] for i, b in enumerate(data)) self._sock.sendall(header + masked) def _recv(self) -> dict: while True: opcode, payload = self._read_frame() if opcode == 0x9: self._send_pong(payload) continue if opcode == 0xA: continue if opcode == 0x8: raise ConnectionError("CDP websocket closed") return json.loads(payload.decode()) def _send_pong(self, payload: bytes) -> None: mask = os.urandom(4) header = bytearray([0x8A, 0x80 | len(payload)]) header.extend(mask) masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) self._sock.sendall(header + masked) def _read_frame(self) -> tuple[int, bytes]: header = self._read_exact(2) opcode = header[0] & 0x0F length = header[1] & 0x7F masked = bool(header[1] & 0x80) if length == 126: length = int.from_bytes(self._read_exact(2), "big") elif length == 127: length = int.from_bytes(self._read_exact(8), "big") mask = self._read_exact(4) if masked else b"" payload = self._read_exact(length) if masked: payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) return opcode, payload def _read_exact(self, size: int) -> bytes: while len(self._buf) < size: chunk = self._sock.recv(4096) if not chunk: raise ConnectionError("CDP websocket closed") self._buf += chunk data, self._buf = self._buf[:size], self._buf[size:] return data class B44LayoutRegressionTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.chrome = chrome_bin() cls.httpd = ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler) cls.http_thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True) cls.http_thread.start() cls.port = cls.httpd.server_address[1] cls.url = f"http://127.0.0.1:{cls.port}{FIXTURE}" cls.tmp = tempfile.TemporaryDirectory(prefix="b44-layout-", ignore_cleanup_errors=True) user_dir = cls.tmp.name cls.proc = subprocess.Popen( [ str(cls.chrome), "--headless=new", "--disable-gpu", "--no-first-run", "--disable-extensions", "--remote-debugging-port=0", f"--user-data-dir={user_dir}", "about:blank", ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) port_file = Path(user_dir) / "DevToolsActivePort" deadline = time.time() + 15 listing = None while time.time() < deadline: if port_file.exists() and port_file.stat().st_size: text = port_file.read_text(encoding="utf-8").strip().splitlines() if text: cls.debug_port = int(text[0]) try: listing = json.loads( urlopen( f"http://127.0.0.1:{cls.debug_port}/json/list", timeout=5, ).read() ) except Exception: listing = None if listing and any(item.get("type") == "page" for item in listing): break time.sleep(0.05) else: raise RuntimeError("Chrome DevTools port not ready") page = next(item for item in listing if item.get("type") == "page") cls.cdp = _Cdp(page["webSocketDebuggerUrl"]) cls.cdp.call("Runtime.enable") cls.cdp.call("Page.enable") @classmethod def tearDownClass(cls) -> None: if getattr(cls, "cdp", None): cls.cdp.close() if getattr(cls, "proc", None): cls.proc.terminate() try: cls.proc.wait(timeout=5) except subprocess.TimeoutExpired: cls.proc.kill() cls.proc.wait(timeout=5) time.sleep(0.2) if getattr(cls, "httpd", None): cls.httpd.shutdown() cls.httpd.server_close() if getattr(cls, "tmp", None): cls.tmp.cleanup() def _metrics(self, width: int) -> dict: self.cdp.call("Emulation.setDeviceMetricsOverride", { "width": width, "height": 1100, "deviceScaleFactor": 1, "mobile": False, }) self.cdp.call("Page.navigate", {"url": f"{self.url}?w={width}"}) deadline = time.time() + 10 last = None while time.time() < deadline: result = self.cdp.call( "Runtime.evaluate", { "expression": ( "typeof window.measureB175 === 'function' " "? (window.measureB175(), document.getElementById('b175-metrics').textContent) " ": null" ), "returnByValue": True, }, ) last = result.get("result", {}).get("value") if last: return json.loads(last) time.sleep(0.1) raise TimeoutError(f"layout metrics not ready: {last}") def test_fifteen_digit_amount_clear_of_unresolved_chip(self) -> None: for width in VIEWPORTS: with self.subTest(width=width): metrics = self._metrics(width) self.assertEqual(width, metrics["viewport"], metrics) self.assertIn("23,456,789,012,345.00", metrics["amountText"]) self.assertIn("未决", metrics["unresolvedText"]) self.assertIn("2026.07.31", metrics["cutoffText"]) self.assertTrue(metrics["amountVisible"]) self.assertTrue(metrics["unresolvedVisible"]) self.assertTrue(metrics["cutoffVisible"]) self.assertFalse(metrics["amountClipped"], metrics) self.assertFalse(metrics["unresolvedClipped"], metrics) self.assertFalse(metrics["cutoffClipped"], metrics) self.assertEqual(0, metrics["amountUnresolvedOverlap"], metrics) self.assertEqual(0, metrics["amountCutoffOverlap"], metrics) self.assertEqual(0, metrics["unresolvedCutoffOverlap"], metrics) self.assertEqual(0, metrics["directionUnresolvedOverlap"], metrics) self.assertEqual(0, metrics["cutoffChevronOverlap"], metrics) self.assertFalse(metrics["drawerAmountClipped"], metrics) if 376 <= width <= 1179: self.assertTrue(metrics["fiveColVisible"], metrics) if 376 <= width <= 900: self.assertTrue(metrics["stackedResult"], metrics) if width <= 375: self.assertFalse(metrics["fiveColVisible"], metrics) LIVE_MEASURE = r""" (() => { const pair = document.querySelector('[data-view="pair"]'); if (pair) pair.click(); const rows = [...document.querySelectorAll("#balanceLedgers .company-ledger.is-balances")]; if (!rows.length) return null; const row = rows.find((item) => (item.textContent || "").includes("123,456,789,012,345")) || rows[0]; const amount = row.querySelector(".ledger-result .amount-with-currency"); const direction = row.querySelector(".ledger-result .status"); const unresolved = row.querySelector(".ledger-unresolved .status") || row.querySelector(".ledger-unresolved"); const cutoff = row.querySelector(".ledger-cutoff"); const chevron = row.querySelector("summary > svg"); const box = (el) => { const r = el.getBoundingClientRect(); return { left: r.left, right: r.right, top: r.top, bottom: r.bottom }; }; const overlap = (a, b) => { const dx = Math.min(a.right, b.right) - Math.max(a.left, b.left); const dy = Math.min(a.bottom, b.bottom) - Math.max(a.top, b.top); if (dx <= 0 || dy <= 0) return 0; return Math.round(dx); }; const clipped = (el) => el.scrollWidth > el.clientWidth + 1; const amountBox = box(amount); const unresolvedBox = box(unresolved); return { viewport: window.innerWidth, amountText: (amount.textContent || "").replace(/\s+/g, " ").trim(), unresolvedText: (unresolved.textContent || "").replace(/\s+/g, " ").trim(), cutoffText: (cutoff.textContent || "").replace(/\s+/g, " ").trim(), amountUnresolvedOverlap: overlap(amountBox, unresolvedBox), directionUnresolvedOverlap: overlap(box(direction), unresolvedBox), amountCutoffOverlap: overlap(amountBox, box(cutoff)), unresolvedCutoffOverlap: overlap(unresolvedBox, box(cutoff)), cutoffChevronOverlap: overlap(box(cutoff), box(chevron)), amountClipped: clipped(amount), unresolvedClipped: clipped(unresolved), cutoffClipped: clipped(cutoff), fiveColVisible: getComputedStyle(document.querySelector(".ledger-head.is-balances")).display !== "none", stackedResult: getComputedStyle(row.querySelector(".ledger-result")).flexDirection === "column", rowCount: rows.length, }; })() """ COMPANY_MEASURE = r""" (() => { const balances = document.querySelector('[data-view="balances"]'); if (balances) balances.click(); const row = document.querySelector(".company-row.is-balance-counterparty"); if (!row) return null; const amount = row.querySelector(".amount-with-currency") || row.querySelector(".company-row-figure strong"); const direction = row.querySelector(".status"); const cutoff = row.querySelector(".company-row-figure small"); const box = (el) => { const r = el.getBoundingClientRect(); return { left: r.left, right: r.right, top: r.top, bottom: r.bottom, width: r.width, height: r.height }; }; const overlap = (a, b) => { const dx = Math.min(a.right, b.right) - Math.max(a.left, b.left); const dy = Math.min(a.bottom, b.bottom) - Math.max(a.top, b.top); if (dx <= 0 || dy <= 0) return 0; return Math.round(dx); }; return { viewport: window.innerWidth, amountText: (amount?.textContent || "").replace(/\s+/g, " ").trim(), directionText: (direction?.textContent || "").replace(/\s+/g, " ").trim(), cutoffText: (cutoff?.textContent || "").replace(/\s+/g, " ").trim(), amountDirectionOverlap: amount && direction ? overlap(box(amount), box(direction)) : 0, visible: Boolean(amount && amount.getBoundingClientRect().width > 4), }; })() """ class LiveAdminDirectoryLayoutTests(unittest.TestCase): """Chromium against a real login/API page with B-168-scale 15-digit amounts.""" @classmethod def setUpClass(cls) -> None: from test_positions_api import IntercompanyApiTests cls.api = IntercompanyApiTests("test_admin_balances_directory") cls.api.setUp() confirmed = cls.api._fresh_pair("123456789012345.00") cls.api._confirm(confirmed["ledger_event_id"]) cls.api._fresh_pair("2150.00") cls.tmp = tempfile.TemporaryDirectory(prefix="b44-live-", ignore_cleanup_errors=True) cls.proc = subprocess.Popen( [ str(chrome_bin()), "--headless=new", "--disable-gpu", "--no-first-run", "--disable-extensions", "--remote-debugging-port=0", f"--user-data-dir={cls.tmp.name}", "about:blank", ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) port_file = Path(cls.tmp.name) / "DevToolsActivePort" deadline = time.time() + 15 listing = None while time.time() < deadline: if port_file.exists() and port_file.stat().st_size: text = port_file.read_text(encoding="utf-8").strip().splitlines() if text: cls.debug_port = int(text[0]) try: listing = json.loads( urlopen( f"http://127.0.0.1:{cls.debug_port}/json/list", timeout=5, ).read() ) except Exception: listing = None if listing and any(item.get("type") == "page" for item in listing): break time.sleep(0.05) else: raise RuntimeError("Chrome DevTools port not ready") page = next(item for item in listing if item.get("type") == "page") cls.cdp = _Cdp(page["webSocketDebuggerUrl"]) cls.cdp.call("Runtime.enable") cls.cdp.call("Page.enable") cls.cdp.call("Network.enable") token = cls.api.admin.cookies["cw_session"] cls.cdp.call( "Network.setCookie", { "name": "cw_session", "value": token, "url": f"http://127.0.0.1:{cls.api.port}/", }, ) @classmethod def tearDownClass(cls) -> None: if getattr(cls, "cdp", None): cls.cdp.close() if getattr(cls, "proc", None): cls.proc.terminate() try: cls.proc.wait(timeout=5) except subprocess.TimeoutExpired: cls.proc.kill() cls.proc.wait(timeout=5) time.sleep(0.2) if getattr(cls, "api", None): cls.api.tearDown() cls.api.doCleanups() if getattr(cls, "tmp", None): cls.tmp.cleanup() def _eval(self, expression: str): deadline = time.time() + 12 last = None while time.time() < deadline: result = self.cdp.call( "Runtime.evaluate", {"expression": expression, "returnByValue": True}, ) last = result.get("result", {}).get("value") if last: return last time.sleep(0.15) raise TimeoutError(f"live page metrics not ready: {last}") def _open(self, path: str, width: int) -> None: self.cdp.call("Emulation.setDeviceMetricsOverride", { "width": width, "height": 1100, "deviceScaleFactor": 1, "mobile": False, }) self.cdp.call("Page.navigate", {"url": f"http://127.0.0.1:{self.api.port}/{path}"}) time.sleep(0.4) def test_live_admin_directory_viewports(self) -> None: for width in (720, 768, 800, 900, 375, 1024, 1440): with self.subTest(width=width): self._open("admin.html", width) metrics = self._eval(LIVE_MEASURE) self.assertEqual(width, metrics["viewport"], metrics) self.assertGreaterEqual(metrics["rowCount"], 1, metrics) self.assertIn("123,456,789,012,345", metrics["amountText"], metrics) self.assertIn("未决", metrics["unresolvedText"], metrics) self.assertIn("2026.07.31", metrics["cutoffText"], metrics) self.assertFalse(metrics["amountClipped"], metrics) self.assertFalse(metrics["unresolvedClipped"], metrics) self.assertEqual(0, metrics["amountUnresolvedOverlap"], metrics) self.assertEqual(0, metrics["directionUnresolvedOverlap"], metrics) self.assertEqual(0, metrics["amountCutoffOverlap"], metrics) if 376 <= width <= 900: self.assertTrue(metrics["stackedResult"], metrics) if width <= 375: self.assertFalse(metrics["fiveColVisible"], metrics) def test_live_company_balance_rows(self) -> None: token = self.api.cashier_a.cookies["cw_session"] self.cdp.call( "Network.setCookie", { "name": "cw_session", "value": token, "url": f"http://127.0.0.1:{self.api.port}/", }, ) for width in (375, 1440): with self.subTest(width=width): self._open("company.html", width) metrics = self._eval(COMPANY_MEASURE) self.assertTrue(metrics["visible"], metrics) self.assertTrue(metrics["amountText"], metrics) self.assertTrue(metrics["directionText"], metrics) self.assertIn("截止", metrics["cutoffText"], metrics) self.assertEqual(0, metrics["amountDirectionOverlap"], metrics) if __name__ == "__main__": unittest.main()