browser-cli/tests/test_performance_integration.py

"""Integration tests for browser performance controls and background jobs."""
import time
import uuid

import pytest

def _wait_for_job(browser, job_id, timeout=10):
    deadline = time.time() + timeout
    last = None
    while time.time() < deadline:
        last = browser("jobs.status", {"jobId": job_id})
        if last.get("status") in {"done", "error", "cancelled"}:
            return last
        time.sleep(0.1)
    return last or {}

def _close_tabs(browser, tab_ids):
    for tab_id in tab_ids:
        try:
            browser("tabs.close", {"tabId": tab_id})
        except Exception:
            pass

def _require_perf_features(browser):
    try:
        return browser("perf.status", {})
    except RuntimeError as exc:
        if "Unknown command: perf.status" in str(exc):
            pytest.skip("Running browser has not reloaded the v0.10.0 extension background worker yet")
        raise

def test_perf_status_and_profile_roundtrip_real_browser(browser):
    initial = _require_perf_features(browser)
    assert "performanceProfile" in initial
    assert "audible" in initial
    assert "throttle" in initial
    assert "jobs" in initial

    original = initial.get("performanceProfile", "auto")
    try:
        changed = browser("perf.set_profile", {"profile": "gentle"})
        assert changed["performanceProfile"] == "gentle"
        status = browser("perf.status", {})
        assert status["performanceProfile"] == "gentle"
        assert status["throttle"]["mode"] == "gentle"
    finally:
        browser("perf.set_profile", {"profile": original})

def test_background_session_load_job_reports_progress_real_browser(browser):
    _require_perf_features(browser)
    name = f"_pytest_perf_job_{uuid.uuid4().hex}"
    marker_url = f"https://example.com/?browser-cli-job={uuid.uuid4().hex}"
    marker_tab = browser("navigate.open", {"url": marker_url, "background": True})
    loaded_ids = set()

    try:
        browser("session.save", {"name": name})
        baseline_ids = {tab["id"] for tab in browser("tabs.list")}

        started = browser("session.load", {
            "name": name,
            "__background": True,
            "lazy": True,
            "eagerTabs": 0,
            "gentleMode": "ultra",
        })
        assert started["status"] == "running"
        assert started["jobId"]

        status = _wait_for_job(browser, started["jobId"])
        assert status["status"] == "done"
        assert status["command"] == "session.load"
        assert status["percent"] == 100
        assert status["phase"] == "done"
        assert status["total"] is None or status["total"] >= 0
        assert status.get("result", {}).get("lazy") is True

        loaded_ids = {tab["id"] for tab in browser("tabs.list")} - baseline_ids
        assert loaded_ids, "Expected lazy session load to create tabs"
    finally:
        _close_tabs(browser, loaded_ids)
        _close_tabs(browser, [marker_tab["id"]])
        try:
            browser("session.remove", {"name": name})
        except Exception:
            pass

def test_lazy_session_load_creates_lightweight_placeholders_real_browser(browser):
    _require_perf_features(browser)
    name = f"_pytest_lazy_{uuid.uuid4().hex}"
    marker_url = f"https://example.com/?browser-cli-lazy={uuid.uuid4().hex}"
    marker_tab = browser("navigate.open", {"url": marker_url, "background": True})
    loaded_ids = set()

    try:
        browser("session.save", {"name": name})
        baseline_ids = {tab["id"] for tab in browser("tabs.list")}
        result = browser("session.load", {"name": name, "lazy": True, "eagerTabs": 0, "gentleMode": "ultra"})
        assert result["lazy"] is True
        assert result["eagerTabs"] == 0

        tabs_after = browser("tabs.list")
        loaded_tabs = [tab for tab in tabs_after if tab["id"] not in baseline_ids]
        loaded_ids = {tab["id"] for tab in loaded_tabs}
        assert loaded_tabs
        assert any((tab.get("url") or "").startswith("data:text/html") for tab in loaded_tabs)
    finally:
        _close_tabs(browser, loaded_ids)
        _close_tabs(browser, [marker_tab["id"]])
        try:
            browser("session.remove", {"name": name})
        except Exception:
            pass

def test_session_load_restores_pinned_tabs_real_browser(browser):
    _require_perf_features(browser)
    name = f"_pytest_pinned_{uuid.uuid4().hex}"
    marker_url = f"https://example.com/?browser-cli-pinned={uuid.uuid4().hex}"
    marker_tab = browser("navigate.open", {"url": marker_url, "background": True})
    loaded_ids = set()

    try:
        browser("tabs.pin", {"tabId": marker_tab["id"]})
        browser("session.save", {"name": name})
        baseline_ids = {tab["id"] for tab in browser("tabs.list")}

        result = browser("session.load", {"name": name, "gentleMode": "ultra", "discardBackgroundTabs": True})
        assert result["tabs"] >= 1

        loaded_tabs = [tab for tab in browser("tabs.list") if tab["id"] not in baseline_ids]
        loaded_ids = {tab["id"] for tab in loaded_tabs}
        matching = [tab for tab in loaded_tabs if tab.get("url") == marker_url]
        assert matching, "Expected session load to restore marker tab"
        assert matching[0].get("pinned") is True
    finally:
        _close_tabs(browser, loaded_ids)
        _close_tabs(browser, [marker_tab["id"]])
        try:
            browser("session.remove", {"name": name})
        except Exception:
            pass

def test_job_cancel_command_real_browser(browser):
    _require_perf_features(browser)
    started = browser("tabs.sort", {"by": "domain", "__background": True, "gentleMode": "ultra"})
    job_id = started["jobId"]
    try:
        cancelled = browser("jobs.cancel", {"jobId": job_id})
        assert cancelled["cancelled"] is True
    except RuntimeError:
        # Tiny real-browser sorts can finish before the cancel request arrives.
        status = browser("jobs.status", {"jobId": job_id})
        assert status["status"] in {"done", "cancelled"}
        return

    status = _wait_for_job(browser, job_id)
    assert status["status"] in {"cancelled", "done"}