Yuvomi-App/scripts/test-integration.py

#!/usr/bin/env python3
"""Real upstream HTTPS/reverse-proxy tests; optionally drive the Android emulator.

Keys and CA trust exist only in a private temporary directory and the .integration
APK. Production and ordinary debug builds never include this trust configuration.
"""
import argparse
import http.cookiejar
import json
import os
from pathlib import Path
import secrets
import shutil
import ssl
import subprocess
import tempfile
import time
import urllib.error
import urllib.request

ROOT = Path(__file__).resolve().parents[1]
REVISION = "10bfef7686ed7303b4c74c7fcc55a5eb16356c9d"


def prepare_android_screenshots(app_output):
    additional = app_output / "connected_android_test_additional_output/integrationAndroidTest"
    screenshots = app_output / "native-screenshots"
    # Never accept a previous run's screenshot as current evidence.
    for stale in additional.rglob("native-dashboard.png"):
        stale.unlink()
    (screenshots / "native-dashboard.png").unlink(missing_ok=True)
    return additional, screenshots


def collect_native_screenshot(additional, screenshots):
    captures = list(additional.rglob("native-dashboard.png"))
    assert len(captures) == 1, "Missing or ambiguous native fixture screenshot"
    capture = captures[0].read_bytes()
    assert capture.startswith(b"\x89PNG\r\n\x1a\n"), "Invalid native fixture screenshot"
    screenshots.mkdir(parents=True, exist_ok=True)
    (screenshots / "native-dashboard.png").write_bytes(capture)


def run(upstream, android):
    revision = subprocess.check_output(["git", "-C", str(upstream), "rev-parse", "HEAD"], text=True).strip()
    assert revision == REVISION, f"Expected pinned upstream {REVISION}, got {revision}"
    base_temp = Path("/tmp/opencode") if Path("/tmp/opencode").exists() else Path(tempfile.gettempdir())
    with tempfile.TemporaryDirectory(prefix="yuvomi-https-", dir=base_temp) as temporary:
        directory = Path(temporary)
        subprocess.run([
            "openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "1",
            "-keyout", str(directory / "key.pem"), "-out", str(directory / "cert.pem"),
            "-subj", "/CN=Yuvomi temporary test CA",
            "-addext", "subjectAltName=IP:127.0.0.1,IP:10.0.2.2,DNS:localhost",
        ], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        with (directory / "server.log").open("w") as log:
            process = subprocess.Popen(["node", str(ROOT / "scripts/integration-server.mjs"), str(upstream), temporary],
                                       cwd=upstream, stdout=log, stderr=log,
                                       env={**os.environ, "NODE_EXTRA_CA_CERTS": str(directory / "cert.pem")})
            try:
                for _ in range(200):
                    if (directory / "ready.json").exists():
                        break
                    if process.poll() is not None:
                        raise RuntimeError((directory / "server.log").read_text())
                    time.sleep(0.1)
                ready = json.loads((directory / "ready.json").read_text())
                port = ready["port"]
                base = f"https://127.0.0.1:{port}"
                context = ssl.create_default_context(cafile=str(directory / "cert.pem"))
                cookies = http.cookiejar.CookieJar()
                client = urllib.request.build_opener(urllib.request.HTTPSHandler(context=context),
                                                     urllib.request.HTTPCookieProcessor(cookies))

                def request(route, payload=None, headers=None):
                    body = None if payload is None else json.dumps(payload).encode()
                    req = urllib.request.Request(base + route, body, headers={
                        "Accept": "application/json", "Content-Type": "application/json", **(headers or {})})
                    try:
                        with client.open(req, timeout=15) as response:
                            return response.status, response.read(), response.headers
                    except urllib.error.HTTPError as error:
                        return error.code, error.read(), error.headers

                # A certificate unknown to the system MUST be rejected.
                try:
                    urllib.request.urlopen(base + "/api/v1/version", timeout=5)
                    raise AssertionError("Untrusted certificate was accepted")
                except urllib.error.URLError as error:
                    assert isinstance(error.reason, ssl.SSLCertVerificationError)
                print("PASS: untrusted TLS certificate rejected")
                status, body, _ = request("/api/v1/version")
                metadata = json.loads(body)
                assert status == 200 and isinstance(metadata["app_name"], str)
                assert metadata["setup_required"] is True and "version" not in metadata
                assert isinstance(metadata["password_reset_enabled"], bool)
                status, body, _ = request("/api/v1/auth/oidc/config")
                assert status == 200 and json.loads(body)["enabled"] is True
                print("PASS: actual upstream compatibility metadata over HTTPS reverse proxy")
                password = secrets.token_urlsafe(24)
                assert request("/api/v1/auth/setup", {"username": "mobiletest", "display_name": "Mobile Test", "password": password})[0] == 201
                status, body, headers = request("/api/v1/auth/login", {"username": "mobiletest", "password": password})
                assert status == 200
                session = [cookie for cookie in cookies if cookie.name == "yuvomi.sid"][0]
                assert session.secure and session.has_nonstandard_attr("HttpOnly")
                assert "SameSite=Lax" in " ".join(headers.get_all("Set-Cookie"))
                csrf = json.loads(body)["csrfToken"]
                assert request("/api/v1/auth/me")[0] == 200
                assert json.loads(request("/api/v1/version")[1])["version"] == "2.78.0"
                assert request("/api/v1/auth/logout", {})[0] == 403
                assert request("/api/v1/auth/logout", {}, {"X-CSRF-Token": csrf})[0] == 200
                assert request("/api/v1/auth/me")[0] == 401
                print("PASS: login, Secure/HttpOnly/SameSite cookies, persistence, CSRF rejection and logout/revocation")
                assert request("/api/v1/auth/oidc/start")[0] == 200
                status, body, _ = request("/api/v1/auth/me")
                identity = json.loads(body)
                assert status == 200 and identity.get("user", {}).get("username") == "mobile-oidc", (
                    f"OIDC session rejected (HTTP {status}); " + "\n".join(
                        (directory / "server.log").read_text().splitlines()[-20:]))
                assert request("/api/v1/auth/logout", {}, {"X-CSRF-Token": identity["csrfToken"]})[0] == 200
                print("PASS: upstream OIDC discovery, RS256/nonce validation, S256 PKCE exchange and test-IdP session")
                if android:
                    resources = directory / "res"
                    (resources / "xml").mkdir(parents=True)
                    (resources / "raw").mkdir()
                    shutil.copyfile(directory / "cert.pem", resources / "raw/integration_ca.pem")
                    (resources / "xml/network_security_config.xml").write_text('''<network-security-config>
<base-config cleartextTrafficPermitted="false"><trust-anchors>
<certificates src="system"/><certificates src="@raw/integration_ca"/>
</trust-anchors></base-config></network-security-config>''')
                    env = {**os.environ, "YUVOMI_TEST_RES": str(resources),
                           "YUVOMI_TEST_URL": base, "YUVOMI_TEST_PASSWORD": password}
                    sdk = Path(env.get("ANDROID_HOME", env.get("ANDROID_SDK_ROOT", "")))
                    adb = sdk / "platform-tools/adb"
                    user_state = subprocess.check_output(
                        [str(adb), "shell", "am", "get-started-user-state", "0"], text=True).strip()
                    if user_state != "RUNNING_UNLOCKED":
                        raise RuntimeError(f"Unlock the emulator before testing (user state: {user_state}).")
                    subprocess.run([str(adb), "reverse", f"tcp:{port}", f"tcp:{port}"], check=True)
                    subprocess.run([str(adb), "reverse", f"tcp:{ready['idpPort']}", f"tcp:{ready['idpPort']}"], check=True)
                    try:
                        app_output = (Path(env["YUVOMI_BUILD_DIR"]) / "app" if "YUVOMI_BUILD_DIR" in env
                                      else ROOT / "android/app/build") / "outputs"
                        additional, screenshots = prepare_android_screenshots(app_output)
                        # The ephemeral password is passed through the local test runner only,
                        # never written into a source file or printed by this harness.
                        result = subprocess.run(["bash", str(ROOT / "android/gradlew"), "-p", str(ROOT / "android"),
                            "-Pintegration", "connectedIntegrationAndroidTest"], env=env, check=False)
                        if result.returncode:
                            raise RuntimeError("Android integration tests failed. Inspect the Gradle test report.")
                        reports = app_output / "androidTest-results/connected/integration"
                        subprocess.run(["python3", str(ROOT / "scripts/verify-test-results.py"),
                                        str(reports), "--minimum", "7"], check=True)
                        collect_native_screenshot(additional, screenshots)
                    finally:
                        subprocess.run([str(adb), "reverse", "--remove", f"tcp:{port}"], check=False)
                        subprocess.run([str(adb), "reverse", "--remove", f"tcp:{ready['idpPort']}"], check=False)
            finally:
                process.terminate()
                try:
                    process.wait(timeout=10)
                except subprocess.TimeoutExpired:
                    process.kill()
                    process.wait()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("upstream", type=Path)
    parser.add_argument("--android", action="store_true")
    arguments = parser.parse_args()
    run(arguments.upstream.resolve(), arguments.android)