openswarm/backend/tests/test_onboarding.py

"""Onboarding v3 endpoints: identity decode, local scan, prep fail-open."""

import base64
import json
from pathlib import Path

import pytest

from backend.apps.onboarding.identity import build_identity, decode_jwt_payload
from backend.apps.onboarding.local_scan import detect_signal_apps, run_local_scan
from backend.apps.onboarding.models import PrepRequest, ScanResult
from backend.apps.onboarding.prep.build_prep import build_prep
from backend.apps.onboarding.prep.parse_prep import parse_prep
from backend.apps.onboarding.prep.scan_fallback import FALLBACK_STARTERS
from backend.apps.settings.models import AppSettings


def make_jwt(payload: dict) -> str:
    body = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip("=")
    return f"header.{body}.sig"


def test_decode_jwt_payload_roundtrip():
    claims = {"email": "eric@example.com", "https://api.openai.com/auth": {"chatgpt_plan_type": "pro"}}
    assert decode_jwt_payload(make_jwt(claims)) == claims


def test_decode_jwt_payload_garbage_is_empty():
    assert decode_jwt_payload("") == {}
    assert decode_jwt_payload("not-a-jwt") == {}
    assert decode_jwt_payload("a.b.c") == {}


def test_build_identity_codex_plan_and_email():
    token = make_jwt({"email": "eric@example.com", "https://api.openai.com/auth": {"chatgpt_plan_type": "plus"}})
    rows = [
        {"provider": "codex", "isActive": True, "idToken": token},
        {"provider": "gemini-cli", "isActive": True, "email": "eric@gmail.com"},
        {"provider": "claude", "isActive": True, "name": "Account 1"},
        {"provider": "codex", "isActive": False, "idToken": token},
        {"provider": "openrouter", "isActive": True},
    ]
    result = build_identity(rows)
    by_provider = {p.provider: p for p in result.providers}
    assert set(by_provider) == {"codex", "gemini-cli", "claude"}
    assert by_provider["codex"].email == "eric@example.com"
    assert by_provider["codex"].plan == "plus"
    assert by_provider["gemini-cli"].email == "eric@gmail.com"
    assert by_provider["claude"].email is None


def test_build_identity_bad_rows_never_raise():
    rows = [{"provider": "codex", "isActive": True, "idToken": 12345}, {"isActive": True}, {}]
    result = build_identity(rows)
    assert [p.provider for p in result.providers] == ["codex"]
    assert result.providers[0].email is None


def test_run_local_scan_counts_names_only(tmp_path: Path):
    downloads = tmp_path / "Downloads"
    downloads.mkdir()
    (downloads / "Screenshot 2026-07-01.png").write_text("x")
    (downloads / "Screen Shot old.png").write_text("x")
    (downloads / "paper.pdf").write_text("x")
    (downloads / ".hidden").write_text("x")
    (tmp_path / ".gitconfig").write_text("[user]")
    repos = tmp_path / "dev"
    (repos / "proj1" / ".git").mkdir(parents=True)
    (repos / "not-a-repo").mkdir()
    result = run_local_scan(tmp_path)
    downloads_summary = next(f for f in result.folders if f.name == "Downloads")
    assert downloads_summary.entry_count == 3
    assert downloads_summary.screenshot_count == 2
    assert "png" in downloads_summary.top_extensions and "pdf" in downloads_summary.top_extensions
    desktop_summary = next(f for f in result.folders if f.name == "Desktop")
    assert desktop_summary.entry_count == 0
    assert result.git_repo_count == 1
    assert result.has_gitconfig is True
    serialized = json.dumps(result.model_dump())
    assert "[user]" not in serialized


def test_signal_apps_picks_tools_not_noise_and_avoids_substring_traps():
    apps = ["Calculator", "Xcode", "Visual Studio Code", "Figma", "Search", "System Settings", "Adobe Photoshop 2024", "Arc"]
    signal = detect_signal_apps(apps)
    assert "Xcode" in signal and "Visual Studio Code" in signal and "Figma" in signal
    assert "Adobe Photoshop 2024" in signal  # prefix match on a versioned name
    assert "Arc" in signal
    # "Search" contains "arc" as a substring but must NOT be treated as the Arc browser.
    assert "Search" not in signal
    assert "Calculator" not in signal and "System Settings" not in signal


def test_parse_prep_strict_and_lenient():
    good = '{"greeting": "Hey!", "starters": [{"title": "A", "prompt": "do a"}, {"title": "B", "prompt": "do b"}]}'
    parsed = parse_prep(f"Sure! Here you go: {good}")
    assert parsed is not None
    assert parsed.greeting == "Hey!"
    assert [s.title for s in parsed.starters] == ["A", "B"]
    # Reason is optional; missing reason defaults to empty, never drops the starter.
    assert parsed.starters[0].reason == ""
    assert parse_prep("no json here") is None
    assert parse_prep('{"greeting": "hi", "starters": []}') is None


def test_parse_prep_carries_reasons():
    rich = (
        '{"greeting": "Hey!", "app_title": "Lift Log", "app_prompt": "Build me a lifting tracker", '
        '"app_reason": "you plan lifts with ChatGPT daily", '
        '"starters": [{"title": "Audit Downloads", "prompt": "audit it", "reason": "1,305 files piling up there"}]}'
    )
    parsed = parse_prep(rich)
    assert parsed is not None
    assert parsed.starters[0].reason == "1,305 files piling up there"
    assert parsed.app_reason == "you plan lifts with ChatGPT daily"


@pytest.mark.asyncio
async def test_build_prep_fails_open_without_provider(monkeypatch):
    async def boom(*args, **kwargs):
        raise ValueError("no provider connected")

    monkeypatch.setattr("backend.apps.agents.providers.registry.resolve_aux_model", boom)
    result = await build_prep(AppSettings(), PrepRequest(scan=ScanResult(), picked_apps=["notion"]))
    assert result.greeting == ""
    assert result.starters == FALLBACK_STARTERS


# --------------------------------------------------------------------------- hero menu ---------------------------------------------------------------------------

def test_parse_menu_strict_and_partial():
    from backend.apps.onboarding.prep.menu import parse_menu

    full = json.dumps({
        "computer": [{"title": "Frame shots", "prompt": "do it"}],
        "research": [{"title": "Vector DBs", "prompt": "compare"}],
        "web": [],
        "build": [{"title": "Jump log", "prompt": "Build me a jump log"}],
    })
    menu = parse_menu(full)
    assert menu is not None
    assert menu.computer[0].title == "Frame shots"
    assert menu.web == []
    assert parse_menu("not json at all") is None
    assert parse_menu(json.dumps({"computer": [], "research": [], "web": [], "build": []})) is None


def test_fill_menu_always_serves_four_per_category():
    from backend.apps.onboarding.prep.menu import MENU_CATEGORIES, fill_menu, parse_menu

    partial = parse_menu(json.dumps({
        "computer": [{"title": "Frame shots", "prompt": "do it"}],
        "research": [], "web": [], "build": [],
    }))
    scan = ScanResult(signal_apps=["Figma"], folders=[], git_repo_count=2)
    menu = fill_menu(partial, scan)
    for category in MENU_CATEGORIES:
        rows = getattr(menu, category)
        assert len(rows) == 4, category
        assert len({r.title for r in rows}) == 4, category
    # LLM rows lead, scan-grounded rows fill before statics.
    assert menu.computer[0].title == "Frame shots"
    assert any(r.title == "Recap my projects" for r in menu.computer)
    assert menu.research[0].title == "Figma tips"
    # No scan at all still fills from statics.
    empty = fill_menu(None, None)
    for category in MENU_CATEGORIES:
        assert len(getattr(empty, category)) == 4


@pytest.mark.asyncio
async def test_build_prep_attaches_menu_even_on_fallback(monkeypatch):

    async def boom(*a, **k):
        raise RuntimeError("no aux")

    monkeypatch.setattr("backend.apps.agents.providers.registry.resolve_aux_model", boom)
    out = await build_prep(AppSettings(), PrepRequest(scan=ScanResult()))
    assert out.menu is not None
    from backend.apps.onboarding.prep.menu import MENU_CATEGORIES
    for category in MENU_CATEGORIES:
        assert len(getattr(out.menu, category)) == 4