cpfgen/extension/test_chromium.py

"""Carrega a extensão no Brave/Chromium (headless) via DevTools Protocol e
confirma que o service worker MV3 sobe e registra os 8 itens de menu.
Usa apenas a stdlib: o WebSocket do CDP é falado na mão."""
import base64, json, os, shutil, socket, struct, subprocess, tempfile, time
import urllib.request

EXT = os.path.dirname(os.path.abspath(__file__))
PORT = 9333

BROWSER = next((b for b in ("brave", "chromium", "google-chrome-stable", "chrome")
                if shutil.which(b)), None)
if not BROWSER:
    print("nenhum navegador Chromium encontrado, pulando")
    raise SystemExit(0)

profile = tempfile.mkdtemp(prefix="chromium-test-")
proc = subprocess.Popen([
    BROWSER,
    "--headless=new",
    f"--remote-debugging-port={PORT}",
    f"--user-data-dir={profile}",
    f"--load-extension={EXT}",
    f"--disable-extensions-except={EXT}",
    "--no-first-run", "--no-default-browser-check",
    "--disable-gpu",
], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)


def cleanup():
    proc.terminate()
    try: proc.wait(timeout=10)
    except Exception: proc.kill()
    shutil.rmtree(profile, ignore_errors=True)


def http(path):
    with urllib.request.urlopen(f"http://127.0.0.1:{PORT}{path}", timeout=10) as r:
        return json.load(r)


class WS:
    """Cliente WebSocket minimalista, suficiente para falar CDP."""
    def __init__(self, url):
        _, _, rest = url.partition("://")
        hostport, _, path = rest.partition("/")
        host, _, port = hostport.partition(":")
        self.s = socket.create_connection((host, int(port)), timeout=20)
        self.s.settimeout(20)
        key = base64.b64encode(os.urandom(16)).decode()
        self.s.sendall((
            f"GET /{path} HTTP/1.1\r\nHost: {hostport}\r\n"
            "Upgrade: websocket\r\nConnection: Upgrade\r\n"
            f"Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n"
        ).encode())
        buf = b""
        while b"\r\n\r\n" not in buf:
            buf += self.s.recv(4096)
        self.buf = buf.split(b"\r\n\r\n", 1)[1]
        self.mid = 0

    def _frame(self, payload):
        data = payload.encode()
        hdr = bytearray([0x81])
        mask = os.urandom(4)
        n = len(data)
        if n < 126:
            hdr.append(0x80 | n)
        elif n < 65536:
            hdr.append(0x80 | 126); hdr += struct.pack(">H", n)
        else:
            hdr.append(0x80 | 127); hdr += struct.pack(">Q", n)
        hdr += mask
        hdr += bytes(b ^ mask[i % 4] for i, b in enumerate(data))
        return bytes(hdr)

    def _recv_exact(self, n):
        while len(self.buf) < n:
            self.buf += self.s.recv(65536)
        out, self.buf = self.buf[:n], self.buf[n:]
        return out

    def _read_frame(self):
        h = self._recv_exact(2)
        ln = h[1] & 0x7F
        if ln == 126:
            ln = struct.unpack(">H", self._recv_exact(2))[0]
        elif ln == 127:
            ln = struct.unpack(">Q", self._recv_exact(8))[0]
        return self._recv_exact(ln)

    def call(self, method, params=None, session=None):
        self.mid += 1
        msg = {"id": self.mid, "method": method, "params": params or {}}
        if session:
            msg["sessionId"] = session
        self.s.sendall(self._frame(json.dumps(msg)))
        while True:
            m = json.loads(self._read_frame())
            if m.get("id") == self.mid:
                if "error" in m:
                    raise RuntimeError(f"{method}: {m['error']}")
                return m.get("result", {})


try:
    for _ in range(60):
        try:
            ver = http("/json/version"); break
        except Exception:
            time.sleep(1)
    else:
        raise SystemExit(f"{BROWSER} não subiu")
    print("browser:", ver.get("Product"))

    ws = WS(ver["webSocketDebuggerUrl"])

    # Espera o service worker da extensão aparecer como alvo.
    target = None
    for _ in range(40):
        for t in ws.call("Target.getTargets")["targetInfos"]:
            if t["type"] == "service_worker" and t["url"].startswith("chrome-extension://"):
                target = t; break
        if target: break
        time.sleep(0.5)
    if not target:
        raise SystemExit("FALHA: service worker da extensão não iniciou")
    print("service worker:", target["url"])

    sess = ws.call("Target.attachToTarget",
                   {"targetId": target["targetId"], "flatten": True})["sessionId"]

    # Os menus só existem se background.js executou até o fim. Recria um item
    # com um id já usado: o erro "duplicate id" prova que o original está lá.
    check = """
      (async () => {
        const probe = (id) => new Promise(res => {
          try {
            chrome.contextMenus.create({id, title: 't', contexts: ['editable']},
              () => res(chrome.runtime.lastError
                ? chrome.runtime.lastError.message : 'CRIADO-NOVO'));
          } catch (e) { res('THROW: ' + e.message); }
        });
        const ids = ['cpf','cnpj','cpf-formatted','cpf-raw','cnpj-formatted',
                     'cnpj-raw','cnpj-alpha-formatted','cnpj-alpha-raw'];
        const out = {};
        for (const id of ids) out[id] = await probe(id);
        return JSON.stringify({
          menusApi: typeof chrome.contextMenus.create === 'function',
          scripting: typeof chrome.scripting.executeScript === 'function',
          probes: out,
        });
      })().catch(e => "ERRO: " + (e && e.stack || e))
    """
    r = ws.call("Runtime.evaluate",
                {"expression": check, "awaitPromise": True, "returnByValue": True},
                session=sess)
    val = r["result"].get("value")
    if not isinstance(val, str):
        raise SystemExit(f"FALHA ao avaliar no SW: {json.dumps(r)[:900]}")
    if val.startswith("ERRO"):
        raise SystemExit(val)

    data = json.loads(val)
    assert data["menusApi"], "chrome.contextMenus indisponível"
    assert data["scripting"], "chrome.scripting indisponível"
    faltando = [i for i, r in data["probes"].items() if "duplicate" not in r.lower()]
    assert not faltando, f"itens não criados pelo background: {faltando}"
    print("8/8 itens de menu confirmados no Chromium")

    # Checa erros do service worker no console.
    errs = ws.call("Log.enable", session=sess)
    print("\nOK: SW ativo, APIs disponíveis, geradores funcionando no Chromium")
finally:
    cleanup()