path: root/packages/meshbay-node/tests/test_e2e_windows_app.py

"""The installed Windows application, driven the way a person uses it.

Opt-in (MESHBAY_WIN_E2E=1): it needs an installed MeshBay, launches it, and
starts and stops the real node. Everything it asserts was broken at least once
on a real install while every other test passed:

- a fresh account signs in and, with nothing else done, has a node set up,
  started, linked and listed in the sidebar (it used to take `meshbay-node
  init` in a terminal and a key pasted on the profile page);
- Stop leaves no meshbay-node.exe behind and only then says "Stopped";
- Start and Restart answer with a running node, a new one for Restart.

Before running it:
- install MeshBay with the setup .exe ("background service" is the default;
  MESHBAY_E2E_MODE=signin|open when another was chosen);
- no %LOCALAPPDATA%\\meshbay and no %APPDATA%\\meshbay-client: it refuses to
  touch an existing node or profile -- move them aside first;
- run it from an ordinary terminal. Inside a packaged application (MSIX,
  such as an IDE or agent installed from the Store) %LOCALAPPDATA% and
  %APPDATA% are virtualised for every child process: what this test writes
  there, a node started by Task Scheduler never sees. Start it outside the
  package (Win32_Process.Create, a plain console window).

    $env:MESHBAY_WIN_E2E = 1
    pytest packages/meshbay-node/tests/test_e2e_windows_app.py -v -s

The hub is a throwaway one on loopback (SQLite, a fresh key), never a real one.
"""

import json
import os
import secrets
import socket
import sqlite3
import subprocess
import sys
import time
from pathlib import Path

import pytest

pytestmark = [
    pytest.mark.skipif(sys.platform != "win32", reason="the Windows application"),
    pytest.mark.skipif(os.environ.get("MESHBAY_WIN_E2E") != "1",
                       reason="opt-in: set MESHBAY_WIN_E2E=1 (see the module docstring)"),
    pytest.mark.timeout(900),
]

LOCALAPPDATA = Path(os.environ.get("LOCALAPPDATA", ""))
APPDATA = Path(os.environ.get("APPDATA", ""))
INSTALL = Path(os.environ.get("MESHBAY_E2E_INSTALL", LOCALAPPDATA / "Programs" / "MeshBay"))
NODE_HOME = LOCALAPPDATA / "meshbay"
APP_PROFILE = APPDATA / "meshbay-client"
MODE = os.environ.get("MESHBAY_E2E_MODE", "service")
CDP_PORT = 9333


def _free_port() -> int:
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def _launch(argv: list[str]) -> None:
    """Start a process detached: DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP."""
    subprocess.Popen(argv, creationflags=0x00000008 | 0x00000200, close_fds=True)


def _node_pids() -> list[int]:
    r = subprocess.run(["tasklist", "/FI", "IMAGENAME eq meshbay-node.exe", "/NH", "/FO", "CSV"],
                       capture_output=True, text=True)
    return [int(c[1]) for c in (line.split('","') for line in r.stdout.splitlines())
            if len(c) > 1 and c[1].isdigit()]


def _wait(what: str, cond, timeout: float, every: float = 1.0):
    deadline = time.monotonic() + timeout
    last = None
    while time.monotonic() < deadline:
        last = cond()
        if last:
            return last
        time.sleep(every)
    raise AssertionError(f"timed out after {timeout:.0f}s waiting for {what} (last: {last!r})")


SUBMIT = "[...document.querySelectorAll('main form button[type=submit]')].pop().click()"


class Page:
    """The application's window over the DevTools protocol."""

    HELPERS = """
      var fill = (el, v) => {
        const proto = el.tagName === 'SELECT' ? HTMLSelectElement.prototype
          : HTMLInputElement.prototype;
        Object.getOwnPropertyDescriptor(proto, 'value').set.call(el, v);
        el.dispatchEvent(new Event('input', { bubbles: true }));
        el.dispatchEvent(new Event('change', { bubbles: true }));
      };
      var clickText = (t) => {
        const els = [...document.querySelectorAll('button, a')].filter(e => !e.disabled);
        const el = els.find(e => e.innerText.trim() === t)
          || els.find(e => e.innerText.trim().startsWith(t));
        if (!el) throw new Error('nothing to click named ' + t);
        el.click();
      };
    """

    def __init__(self):
        from websockets.sync.client import connect
        targets = json.loads(_http_get(f"http://127.0.0.1:{CDP_PORT}/json/list"))
        page = next(t for t in targets if t["type"] == "page" and t["url"].startswith("app://"))
        self._conn = connect(page["webSocketDebuggerUrl"], max_size=None)
        self._ws = self._conn.__enter__()
        self._id = 0

    def eval(self, expr: str):
        self._id += 1
        self._ws.send(json.dumps({"id": self._id, "method": "Runtime.evaluate", "params": {
            "expression": f"{self.HELPERS}\n(async () => ({expr}))()",
            "awaitPromise": True, "returnByValue": True, "userGesture": True}}))
        while True:
            msg = json.loads(self._ws.recv(timeout=120))
            if msg.get("id") == self._id:
                break
        res = msg["result"]
        if "exceptionDetails" in res:
            raise AssertionError(res["exceptionDetails"].get("exception", {}).get("description"))
        return res["result"].get("value")

    def text(self) -> str:
        return self.eval("document.body.innerText")

    def sidebar(self) -> str:
        return self.eval("(document.querySelector('aside') || {}).innerText || ''")

    def click(self, label: str, timeout: float = 30) -> None:
        """Click a button or link by its label once it is there and enabled."""
        time.sleep(0.3)   # after a fill, let the page render what it was given
        _wait(f"a control named {label!r}",
              lambda: _try(lambda: self.eval(f"(clickText({json.dumps(label)}), 1)")), timeout)

    def close(self):
        self._conn.__exit__(None, None, None)


def _http_get(url: str, token: str | None = None) -> str:
    import urllib.request
    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"} if token else {})
    with urllib.request.urlopen(req, timeout=5) as r:
        return r.read().decode()


def _node_status() -> dict | None:
    import urllib.request
    try:
        token = (NODE_HOME / "data" / "ui-token").read_text(encoding="utf-8").strip()
        with urllib.request.urlopen(f"http://127.0.0.1:18000/api/status?t={token}",
                                    timeout=3) as r:
            return json.loads(r.read())
    except (OSError, ValueError):
        return None


@pytest.fixture(scope="module")
def hub(tmp_path_factory):
    pytest.importorskip("meshbay_hub")
    pytest.importorskip("websockets.sync.client")
    work = tmp_path_factory.mktemp("hub")
    port = _free_port()
    env = {**os.environ,
           "MESHBAY_DATABASE_URL": f"sqlite+aiosqlite:///{(work / 'hub.db').as_posix()}",
           "MESHBAY_HUB_ID": "127.0.0.1", "MESHBAY_HUB_KEY": str(work / "hub_private.pem"),
           "MESHBAY_HUB_HOST": "127.0.0.1", "MESHBAY_HUB_PORT": str(port)}
    proc = subprocess.Popen([sys.executable, "-m", "uvicorn", "meshbay_hub.app:create_app",
                             "--factory", "--host", "127.0.0.1", "--port", str(port)],
                            env=env, cwd=work, stdout=(work / "hub.log").open("w"),
                            stderr=subprocess.STDOUT)
    url = f"http://127.0.0.1:{port}"
    try:
        _wait("the local hub", lambda: _try(lambda: _http_get(f"{url}/v1/hub/version")), 30)
        yield {"url": url, "db": work / "hub.db"}
    finally:
        proc.kill()


def _try(fn):
    try:
        return fn()
    except Exception:
        return None


@pytest.fixture(scope="module")
def app(hub):
    exe = INSTALL / "MeshBay.exe"
    if not exe.exists():
        pytest.skip(f"no installed MeshBay at {INSTALL}")
    for there in (NODE_HOME / "node.toml", NODE_HOME / "keystore.enc", APP_PROFILE):
        if there.exists():
            pytest.fail(f"{there} exists: this test only runs on a machine with no node "
                        "and no app profile -- move them aside, it never deletes them")
    if MODE == "service":
        r = subprocess.run(["schtasks", "/query", "/tn", "MeshBay Node"], capture_output=True)
        assert r.returncode == 0, "service mode chosen at install, but there is no boot task"
    APP_PROFILE.mkdir(parents=True)
    (APP_PROFILE / "config.json").write_text(json.dumps({"hubBase": hub["url"]}),
                                             encoding="utf-8")
    _launch([str(exe), f"--remote-debugging-port={CDP_PORT}"])
    page = _wait("the application window", lambda: _try(Page), 60)
    try:
        yield page
    finally:
        # What the window showed last: the first thing to read when it failed.
        last = _try(lambda: page.eval("location.href + ' ' + document.body.innerText"))
        print(f"\n--- the application's page at the end ---\n{last}")
        page.close()
        subprocess.run(["taskkill", "/F", "/IM", "MeshBay.exe"], capture_output=True)
        subprocess.run([str(INSTALL / "resources" / "node-runtime" / "meshbay-node.exe"),
                        "autostart", "stop"], capture_output=True, timeout=120)


def test_a_fresh_account_gets_its_node_set_up_and_can_stop_it(app, hub):
    page = app
    username = "e2e" + secrets.token_hex(3)
    passphrase = "e2e-" + secrets.token_urlsafe(18)

    # Register.
    _wait("the sign-in page", lambda: "Register" in page.text(), 30)
    page.eval("(location.hash = '#/register', 1)")
    _wait("the registration form", lambda: page.eval(
        "document.querySelectorAll('main input').length >= 4"), 15)
    page.eval(f"""(() => {{ const i = document.querySelectorAll('main input');
        fill(i[0], {json.dumps(username)}); fill(i[1], {json.dumps(username + '@example.invalid')});
        fill(i[2], {json.dumps(passphrase)}); fill(i[3], {json.dumps(passphrase)});
        return 1; }})()""")
    time.sleep(0.5)   # the form's state follows the input events a render later
    page.eval(f"({SUBMIT}, 1)")
    # Not "recovery key" alone: the registration form says it too.
    _wait("the recovery key", lambda: "I have saved my recovery key" in page.text(), 90)
    page.eval("(document.querySelector('input[type=checkbox]').click(), 1)")
    page.click("Continue")
    _wait("the verification form", lambda: "Verify" in page.text(), 30)
    code = sqlite3.connect(hub["db"]).execute(
        "select code from email_verifications order by rowid desc limit 1").fetchone()[0]
    page.eval(f"(fill(document.querySelector('main input'), {json.dumps(code)}), 1)")
    page.click("Verify")
    _wait("the verified page", lambda: "Go to login" in page.text(), 30)
    page.click("Go to login")
    _wait("the login form", lambda: page.eval(
        "document.querySelectorAll('main input').length >= 2"), 15)
    page.eval(f"""(() => {{ const i = document.querySelectorAll('main input');
        fill(i[0], {json.dumps(username)}); fill(i[1], {json.dumps(passphrase)});
        return 1; }})()""")
    time.sleep(0.5)   # the form's state follows the input events a render later
    page.eval(f"({SUBMIT}, 1)")

    # Nothing else done: the node is set up, running, linked, and in the sidebar.
    _wait("the Node section in the sidebar", lambda: "Create group" in page.sidebar(), 120)
    status = _wait("a running node", lambda: (s := _node_status()) and
                   s.get("status") == "running" and s, 120)
    toml = (NODE_HOME / "node.toml").read_text(encoding="utf-8")
    assert f'username = "{username}"' in toml and hub["url"] in toml
    token = page.eval("JSON.parse(localStorage.getItem('mb_auth')).token")
    keys = json.loads(_http_get(f"{hub['url']}/v1/users/{username}/pubkeys", token))
    assert keys.get("pk_node_ed25519") == status["pk_node_ed25519"]

    # Stop: no process left, and only then "Stopped".
    page.eval("(location.hash = '#/node', 1)")
    _wait("the Node page", lambda: "Restart" in page.text(), 30)
    page.click("Stop")
    _wait("no meshbay-node.exe", lambda: not _node_pids(), 60)
    _wait("the page saying Stopped", lambda: "Stopped" in page.text(), 30)
    assert _node_status() is None

    # Start, then Restart: a running node each time, a new one for Restart.
    page.click("Start")
    first = _wait("a node after Start", lambda: (s := _node_status()) and
                  s.get("status") == "running" and s, 120)
    _wait("the page saying Running", lambda: "Running" in page.text(), 30)
    page.click("Restart")
    second = _wait("a new node after Restart", lambda: (s := _node_status()) and
                   s.get("status") == "running" and s.get("pid") != first.get("pid") and s, 120)
    assert len(_node_pids()) == 1, _node_pids()
    assert second["pk_node_ed25519"] == first["pk_node_ed25519"]