vrr/ouroboros
2
0
Fork 0
mirror of https://github.com/razzant/ouroboros.git synced 2026-10-03 04:07:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7
ouroboros/tests/test_subscription_setup_browser.py
Ouroboros 227b99b4f3 ui: establish a coherent shared design system 
Centralize shared UI primitives and repair cross-surface chat, form, and interaction presentation while preserving existing task and Project behavior.
2026-09-09 20:24:35 +03:00

330 lines
17 KiB
Python
Raw Permalink Blame History

"""Real UI modules against controlled subscription API responses, without runtime startup."""
from __future__ import annotations
import copy
import json
import os
from datetime import datetime, timedelta, timezone
from functools import partial
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from threading import Thread
from urllib.parse import parse_qs, urlparse
import pytest
from tests.test_onboarding_complete_endpoint import (
LIVE_SNAPSHOT, _profile, _profile_account,
onboarding as onboarding, # explicit re-export of the real atomic settings fixture
)
pytestmark = [pytest.mark.ui_browser, pytest.mark.serial]
WEB = Path(__file__).resolve().parents[1] / "web"
@pytest.fixture
def subscription_ui():
if os.environ.get("OUROBOROS_RUN_UI_SMOKE") != "1":
pytest.skip("Set OUROBOROS_RUN_UI_SMOKE=1 to run the browser flow")
playwright = pytest.importorskip("playwright.sync_api")
bootstrap = json.loads((WEB / "tests/fixtures/onboarding_bootstrap.json").read_text())
fixture = json.loads((WEB / "tests/fixtures/subscription_setup.json").read_text())
fixture['status']['quota'] = [{
'subject': {'harness': 'codex', 'subject_id': 'personal'}, 'freshness': 'fresh',
'constraints': [{'id': 'window', 'used_ratio': 0.38,
'resets_at': (datetime.now(timezone.utc) + timedelta(hours=2)).isoformat()}],
}]
posts = []
reads = []
page_errors = []
backend = {}
settings = {
**fixture["preview"]["model_settings"],
"OUROBOROS_SUBAGENTS": fixture["preview"]["available_subagents"],
"OUROBOROS_REVIEWER_SLOTS": json.dumps(fixture["preview"]["reviewer_slots"]),
"OUROBOROS_RUNTIME_MODE": "advanced",
"OUROBOROS_CONTEXT_MODE": "max",
"_meta": {"setup_contract": bootstrap["contract"]},
}
class Handler(SimpleHTTPRequestHandler):
def log_message(self, *_):
pass
def do_GET(self):
if self.path == "/onboarding":
body = (WEB / "onboarding_template.html").read_text().replace(
"__ONBOARDING_BOOTSTRAP__", json.dumps(bootstrap))
self.send_response(200)
self.send_header("Content-Type", "text/html")
self.end_headers()
self.wfile.write(body.encode())
return
self.path = self.path.removeprefix("/static")
super().do_GET()
def respond(route):
request = route.request
path = urlparse(request.url).path
reads.append(request.url)
body = {}
if request.method == "POST":
payload = request.post_data_json or {}
posts.append((path, payload))
if path == "/api/onboarding/subagents/preview":
body = copy.deepcopy(fixture["preview"])
if not payload.get('subscriptionsConnected'):
body['model_settings'] = {key: payload.get(key, '') for key in body['model_settings']}
body['available_subagents'] = {'enabled': True, 'items': []}
body['reviewer_slots'] = {
'triad': [{'slot_id': 'triad_1', 'route': {'kind': 'api_chat', 'target_id': payload.get('OUROBOROS_MODEL')}}],
'scope': [{'slot_id': 'scope_1', 'route': {'kind': 'api_chat', 'target_id': payload.get('OUROBOROS_MODEL')}}],
'advisory': {'enabled': True, 'route': {'kind': 'api_chat', 'target_id': payload.get('OUROBOROS_MODEL')}},
}
body["reviewer_slots"] = json.dumps(body["reviewer_slots"])
elif path == "/api/onboarding/complete":
assert isinstance(payload["OUROBOROS_REVIEWER_SLOTS"], str)
assert isinstance(json.loads(payload["OUROBOROS_REVIEWER_SLOTS"]), dict)
if backend.get("client"):
result = backend["client"].post(path, json=payload)
route.fulfill(status=result.status_code, content_type="application/json", body=result.text)
return
body = {"ok": True, "runtime_mode": "advanced", "restart_required": False}
elif path == "/api/settings":
body = {"ok": True, "saved": True}
elif path == "/api/claudexor/status":
body = fixture["status"]
elif path == "/api/settings":
body = settings
elif path == "/api/reviewer-slots":
body = fixture["preview"]["reviewer_slots"]
elif path == "/api/model-catalog":
body = copy.deepcopy(fixture["catalog"])
profile = parse_qs(urlparse(request.url).query).get("credential_profile_id", [""])[0]
if profile == "work":
body["items"][0].update(credential_profile_id="work", max_context_window=500000)
elif path == "/api/onboarding":
route.fulfill(status=204)
return
elif path == "/api/state":
body = {"supervisor_ready": True, "active_chat_activities": [], "projects": []}
elif path == "/api/projects":
body = {"projects": []}
elif path == "/api/chat/history":
body = {"messages": [], "progress": []}
elif path == "/api/health":
body = {"ok": True, "version": "fixture"}
route.fulfill(content_type="application/json", body=json.dumps(body))
server = ThreadingHTTPServer(("127.0.0.1", 0), partial(Handler, directory=str(WEB)))
thread = Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
with playwright.sync_playwright() as pw:
engine = os.environ.get("OUROBOROS_UI_BROWSER_ENGINE", "chromium")
if engine not in {"chromium", "webkit", "firefox"}:
raise ValueError(f"Unsupported browser engine: {engine}")
browser = getattr(pw, engine).launch(headless=True)
try:
page = browser.new_page(viewport={"width": 1360, "height": 900},
has_touch=os.environ.get("OUROBOROS_UI_HAS_TOUCH") == "1")
page.route_web_socket('**/ws', lambda ws: ws.send(json.dumps({"type": "heartbeat"})))
page.route("**/api/**", respond)
page.on("pageerror", lambda error: page_errors.append(str(error)))
yield {"page": page, "url": f"http://127.0.0.1:{server.server_port}",
"posts": posts, "reads": reads, "fixture": fixture,
"settings": settings, "errors": page_errors, "backend": backend}
assert not page_errors
finally:
browser.close()
finally:
server.shutdown()
server.server_close()
thread.join(timeout=5)
def capture(page, name):
root = os.environ.get("OUROBOROS_UI_EVIDENCE_DIR")
if root:
target = Path(root)
target.mkdir(parents=True, exist_ok=True)
page.screenshot(path=str(target / f"{name}.png"), animations="disabled")
def test_codex_quick_setup_computes_skipped_steps_and_finishes_once(subscription_ui, onboarding):
ui = subscription_ui
ui['backend']['client'] = onboarding.client
onboarding.calls['snapshot_payload'] = {
**LIVE_SNAPSHOT, 'harnesses': [LIVE_SNAPSHOT['harnesses'][1]],
'profiles': {'harnessAccounts': [_profile_account('codex', 'personal')],
'profiles': [_profile('codex', 'personal')]},
'model_catalog': ui['fixture']['catalog']['items'],
}
ui['fixture']['status']['profiles']['profiles'] = ui['fixture']['status']['profiles']['profiles'][:1]
page = ui["page"]
page.goto(ui["url"] + "/onboarding")
page.wait_for_selector('#quick-start-btn:not([hidden])')
assert page.locator('.wizard-step').count() == 5
capture(page, "accounts-connected")
page.click('#quick-start-btn')
page.wait_for_selector('.summary-card')
assert "gpt-test" in page.locator('.summary-card').inner_text()
assert "Deep self-review" in page.locator('.summary-card').inner_text()
capture(page, "summary-quick")
page.click('#next-btn')
page.wait_for_url(ui["url"] + "/")
writes = [body for path, body in ui["posts"] if path == '/api/onboarding/complete']
assert len(writes) == 1
assert writes[0]["OPENROUTER_API_KEY"] == ""
assert writes[0]["OUROBOROS_MODEL"] == 'claudexor::codex=gpt-test'
assert json.loads(writes[0]["OUROBOROS_REVIEWER_SLOTS"])["deep_review"]["subagent_id"] == "codex"
assert not any(path == '/api/settings' for path, _ in ui["posts"])
saved = onboarding.saved()
assert saved['OUROBOROS_MODEL'] == writes[0]['OUROBOROS_MODEL']
assert saved['OUROBOROS_REVIEWER_SLOTS'] == writes[0]['OUROBOROS_REVIEWER_SLOTS']
assert not saved['OPENAI_API_KEY'] and not saved['OPENROUTER_API_KEY']
assert onboarding.calls['supervisor'] == 1
@pytest.mark.parametrize("credential_harness, expected", [("codex", True), ("claude", False)])
def test_quick_start_requires_a_model_source_for_the_connected_harness(subscription_ui, credential_harness, expected):
ui, page = subscription_ui, subscription_ui["page"]
ui["fixture"]["catalog"]["model_sources"] = [
{"id": "opaque-source", "label": "Managed models", "credentialHarness": credential_harness},
]
with page.expect_response("**/api/model-catalog"):
page.goto(ui["url"] + "/onboarding")
page.wait_for_function(
"expected => document.querySelector('#quick-start-btn').hidden !== expected", arg=expected)
assert page.locator("#quick-start-btn").is_visible() is expected
def test_model_roles_pin_context_fallback_and_manual_draft_survive_preview(subscription_ui):
ui = subscription_ui
page = ui["page"]
page.goto(ui["url"] + "/onboarding")
page.wait_for_selector('#quick-start-btn:not([hidden])')
page.click('#next-btn')
page.wait_for_selector('[data-model-role="main"]')
main = page.locator('[data-model-role="main"]')
light = page.locator('[data-model-role="light"]')
main.locator('[data-model-role-account]').select_option('personal')
held_catalog = []
page.route('**/api/model-catalog?*credential_profile_id=work', lambda route: held_catalog.append(route))
light.locator('[data-model-role-account]').select_option('work')
main.locator('summary').click()
light.locator('summary').click()
assert 'not known' in light.locator('[data-model-context-note]').inner_text()
page.wait_for_function("() => document.querySelector('[data-model-role=light] [data-model-role-account]').value === 'work'")
assert held_catalog
work_catalog = copy.deepcopy(ui['fixture']['catalog'])
work_catalog['items'][0].update(credential_profile_id='work', max_context_window=500000)
for route in held_catalog:
route.fulfill(content_type='application/json', body=json.dumps(work_catalog))
page.unroute('**/api/model-catalog?*credential_profile_id=work')
page.wait_for_function("() => document.querySelector('[data-model-role=light] [data-model-context-note]').textContent.includes('500,000')")
assert '872,000' in main.locator('[data-model-context-note]').inner_text()
main.locator('[data-model-role-context]').fill('1000000')
assert 'set by you' in main.locator('[data-model-context-note]').inner_text()
page.locator('[data-model-role="vision"] [data-model-role-account]').select_option('work')
page.click('[data-model-add]')
fallback = page.locator('[data-model-role-group="fallback"]')
fallback.locator('[data-model-role-source]').select_option('subscription:codex')
fallback.locator('[data-model-role-model]').fill('first')
fallback.locator('[data-model-role-account]').select_option('work')
page.click('[data-model-add]')
fallback.locator('[data-model-role-model]').last.fill('openai::second')
fallback.locator('[data-model-up]').last.click()
main.locator('[data-model-role-model]').fill('owner-model')
page.evaluate('window.scrollTo(0, 0)')
capture(page, "models-desktop")
page.click('#next-btn')
page.wait_for_selector('#reviewer-slots-section', state='attached')
page.locator('details').filter(has=page.locator('#reviewer-slots-section')).locator(':scope > summary').click()
assert page.locator('#reviewer-deep-review-row select').count() > 0
capture(page, "review-editable")
deep_review = page.locator('#reviewer-deep-review-row')
deep_review.locator('[data-deep-review-effort]').select_option('high')
deep_review.scroll_into_view_if_needed()
capture(page, "deep-review-editable")
page.click('#next-btn')
page.wait_for_selector('[data-collapse="api-budget"]')
assert not page.locator('[data-collapse="api-budget"]').evaluate('(el) => el.open')
capture(page, "budget-subscription")
page.click('#next-btn')
page.wait_for_selector('.summary-card')
page.wait_for_function("() => !document.querySelector('.wizard-error').textContent")
summary = page.locator('.summary-card')
assert '1,000,000 tokens, set by you' in summary.inner_text()
vision = summary.locator('.summary-kv').filter(has=page.get_by_text('Vision', exact=True))
assert 'Uses Main' in vision.inner_text()
assert 'Account: work' in vision.inner_text()
deep_summary = summary.locator('.summary-kv').filter(has=page.get_by_text('Deep self-review', exact=True))
assert 'Effort: high' in deep_summary.inner_text()
capture(page, "summary-manual")
page.click('#next-btn')
page.wait_for_url(ui["url"] + "/")
body = [body for path, body in ui["posts"] if path == '/api/onboarding/complete'][0]
assert body['OUROBOROS_MODEL'] == 'claudexor::codex=owner-model'
assert body['OUROBOROS_MODEL_LIGHT'] == 'claudexor::codex=gpt-test'
assert body['OUROBOROS_MODEL_ACCOUNTS']['main'] == 'personal'
assert body['OUROBOROS_MODEL_ACCOUNTS']['light'] == 'work'
assert body['OUROBOROS_MODEL_ACCOUNTS']['vision'] == 'work'
assert body['OUROBOROS_MODEL_ACCOUNTS']['fallback'] == ['', 'work']
assert body['OUROBOROS_MODEL_CONTEXT_WINDOWS']['main'] == 1000000
assert body['OUROBOROS_MODEL_FALLBACKS'] == 'openai::second, claudexor::codex=first'
assert json.loads(body['OUROBOROS_REVIEWER_SLOTS'])['deep_review']['effort'] == 'high'
def test_settings_accounts_and_model_roles_use_the_same_compact_components(subscription_ui):
ui = subscription_ui
page = ui['page']
page.goto(ui['url'] + '/#settings')
page.wait_for_selector('#settings-model-roles .model-role-row', state='attached')
assert page.locator('[data-settings-tab="providers"]').inner_text() == 'Accounts'
assert page.locator('[data-settings-panel="providers"] #harness-accounts-section').count() == 1
assert page.locator('[data-settings-panel="agents"] #harness-accounts-section').count() == 0
page.click('[data-settings-tab="models"]')
page.wait_for_function("() => document.querySelector('[data-settings-tab=models]').getAttribute('aria-selected') === 'true'")
main = page.locator('[data-model-role="main"]')
main.locator('[data-model-role-account]').select_option('work')
inputs = main.locator('.model-role-controls > input, .model-role-controls > select')
tops = inputs.evaluate_all('(els) => els.filter(el => !el.hidden).map(el => el.getBoundingClientRect().top)')
assert max(tops) - min(tops) <= 2
assert main.locator('input').first.evaluate('(el) => getComputedStyle(el).fontSize') == '14px'
capture(page, 'settings-models-desktop')
page.set_viewport_size({'width': 390, 'height': 844})
page.wait_for_function("() => document.querySelector('#primary-sidebar').getBoundingClientRect().right <= 1")
assert page.evaluate('document.documentElement.scrollWidth <= window.innerWidth')
capture(page, 'settings-models-narrow')
def test_cursor_only_does_not_claim_model_access_and_api_only_finishes(subscription_ui):
ui = subscription_ui
ui['fixture']['status']['profiles']['profiles'] = [{
'profile': {'profile_id': 'cursor-only', 'harness_id': 'cursor', 'enabled': True},
'status': {'verification': 'passed'},
}]
page = ui['page']
page.goto(ui['url'] + '/onboarding')
page.wait_for_function("() => document.querySelector('[data-agent-family=cursor]').textContent.includes('Connected')")
assert page.locator('#quick-start-btn').is_hidden()
assert page.locator('#next-btn').is_disabled()
assert 'Main still needs Codex' in page.locator('#agents-outcome').inner_text()
# Removing the optional agent gives the API-only path the same five steps.
ui['fixture']['status']['profiles']['profiles'] = []
page.reload()
page.locator('[data-collapse="api-access"] > summary').click()
page.locator('#openai-key').fill('fixture-api-credential')
page.click('#next-btn')
page.wait_for_selector('#main-model')
assert page.locator('#main-model').input_value()
for _ in range(3):
page.click('#next-btn')
page.wait_for_selector('.summary-card')
assert 'no API key' not in page.locator('.summary-card').inner_text()
page.click('#next-btn')
page.wait_for_url(ui['url'] + '/')
body = [body for path, body in ui['posts'] if path == '/api/onboarding/complete'][0]
assert body['OPENAI_API_KEY'] == 'fixture-api-credential'
assert body['OUROBOROS_MODEL'].startswith('openai::')
assert not body['subscriptionsConnected']
Reference in a new issue View git blame Copy permalink