codexpi/tests/test_application_experience.py

from __future__ import annotations

import asyncio
import subprocess
from datetime import UTC, datetime, timedelta
from pathlib import Path

import pytest
from starlette.websockets import WebSocketDisconnect

from codexpi.git_service import branches, file_contents
from codexpi.system_service import codex_account_quota, codex_quota_snapshot, consume_codex_reset_credit
from codexpi.terminal import terminal_argv


def git(root: Path, *args: str) -> None:
    subprocess.run(["git", *args], cwd=root, check=True, capture_output=True)


def make_repo(root: Path) -> None:
    git(root, "init")
    git(root, "config", "user.email", "test@example.invalid")
    git(root, "config", "user.name", "CodexPi Test")
    (root / "app.py").write_text("print('before')\n", encoding="utf-8")
    git(root, "add", "app.py")
    git(root, "commit", "-m", "initial")


def register(logged_in, name: str = "workspace"):
    client, _, projects, _, headers = logged_in
    repo = projects / name
    repo.mkdir()
    make_repo(repo)
    response = client.post("/api/projects", headers=headers, json={"name": name, "path": str(repo)})
    assert response.status_code == 201
    return client, response.json(), repo, headers


def test_dashboard_and_enriched_projects(logged_in):
    client, project, _, _ = register(logged_in)
    dashboard = client.get("/api/dashboard")
    assert dashboard.status_code == 200
    data = dashboard.json()
    assert data["connected"] is True
    assert {"cpu_percent", "ram_percent", "disk_percent", "uptime_seconds", "docker", "codex", "git", "codex_quota"} <= data.keys()
    assert data["codex_quota"]["available"] is False
    listed = client.get("/api/projects").json()[0]
    assert listed["id"] == project["id"]
    assert listed["git_available"] is True
    assert listed["modified_files"] == 0
    assert "last_used" in listed


def test_plain_project_folder_disables_git_without_an_error(logged_in):
    client, _, projects, _, headers = logged_in
    folder = projects / "plain-folder"
    folder.mkdir()
    created = client.post(
        "/api/projects",
        headers=headers,
        json={"name": "Plain folder", "path": str(folder)},
    )
    assert created.status_code == 201
    project_id = created.json()["id"]
    state = client.get(f"/api/projects/{project_id}/git")
    assert state.status_code == 200
    assert state.json() == {"is_repository": False, "branch": None, "files": [], "history": []}
    listed = client.get("/api/projects").json()[0]
    assert listed["git_available"] is False


def test_project_file_download_is_private_and_scoped(logged_in):
    client, project, repo, _ = register(logged_in, "download-project")
    report = repo / "report final.xlsx"
    report.write_bytes(b"spreadsheet-content")
    response = client.get(
        f"/api/projects/{project['id']}/files/download",
        params={"path": "report final.xlsx"},
    )
    assert response.status_code == 200
    assert response.content == b"spreadsheet-content"
    assert response.headers["content-disposition"].startswith("attachment;")
    assert "no-store" in response.headers["cache-control"]
    image = repo / "preview.png"
    image.write_bytes(b"\x89PNG\r\n\x1a\npreview")
    inline = client.get(
        f"/api/projects/{project['id']}/files/download",
        params={"path": "preview.png", "inline": "true"},
    )
    assert inline.status_code == 200
    assert inline.headers["content-disposition"].startswith("inline;")
    assert client.get(
        f"/api/projects/{project['id']}/files/download",
        params={"path": "report final.xlsx", "inline": "true"},
    ).status_code == 415
    assert client.get(f"/api/projects/{project['id']}/files/download", params={"path": "../secret"}).status_code == 404
    client.cookies.clear()
    assert client.get(f"/api/projects/{project['id']}/files/download", params={"path": "report final.xlsx"}).status_code == 401


def test_legacy_remote_workspace_link_downloads_existing_file(logged_in):
    client, app, _, _, _ = logged_in
    project_id = "11111111-2222-4333-8444-555555555555"
    support = app.state.db.path.parent / "remote-workspaces" / project_id
    support.mkdir(parents=True)
    (support / "result.xlsm").write_bytes(b"macro-workbook")
    app.state.db.execute(
        "INSERT INTO projects(id,name,path,full_access,kind,created_at) VALUES(?,?,?,?,?,?)",
        (project_id, "Remote", str(support), 0, "remote", "2026-09-15T00:00:00+00:00"),
    )
    response = client.get(f"/app/data/remote-workspaces/{project_id}/result.xlsm")
    assert response.status_code == 200
    assert response.content == b"macro-workbook"
    assert response.headers["content-disposition"].startswith("attachment;")


def test_codex_quota_reads_latest_valid_rate_limit(tmp_path: Path):
    sessions = tmp_path / "sessions"
    sessions.mkdir()
    events = sessions / "rollout.jsonl"
    events.write_text(
        '{"type":"event_msg","timestamp":"2026-09-13T20:00:00Z","payload":{"type":"token_count","rate_limits":{"plan_type":"pro","primary":{"used_percent":37.4,"window_minutes":300,"resets_at":1789333200},"secondary":null}}}\n'
        '{"type":"response_item","payload":{"rate_limits":{"primary":{"used_percent":99,"window_minutes":1,"resets_at":1}}}}\n',
        encoding="utf-8",
    )
    quota = codex_quota_snapshot(sessions)
    assert quota["available"] is True
    assert quota["primary"] == {"used_percent": 37.4, "remaining_percent": 62.6, "window_minutes": 300, "resets_at": 1789333200}
    assert quota["secondary"] is None
    assert quota["plan_type"] == "pro"
    assert quota["observed_at"] == "2026-09-13T20:00:00Z"
    assert quota["limits"][0]["id"] == "codex"
    assert quota["credits"]["balance"] == "0"


def test_codex_quota_aggregates_model_specific_limits(tmp_path: Path):
    sessions = tmp_path / "sessions"
    sessions.mkdir()
    (sessions / "general.jsonl").write_text(
        '{"type":"event_msg","timestamp":"2026-09-13T20:00:00Z","payload":{"type":"token_count","rate_limits":{"limit_id":"codex","primary":{"used_percent":17,"window_minutes":10080,"resets_at":1789906246}}}}\n',
        encoding="utf-8",
    )
    (sessions / "spark.jsonl").write_text(
        '{"type":"event_msg","timestamp":"2026-09-13T20:01:00Z","payload":{"type":"token_count","rate_limits":{"limit_id":"codex-spark","limit_name":"GPT-5.3-Codex-Spark","primary":{"used_percent":0,"window_minutes":300,"resets_at":1789333200},"secondary":{"used_percent":0,"window_minutes":10080,"resets_at":1789906246}}}}\n',
        encoding="utf-8",
    )
    quota = codex_quota_snapshot(sessions)
    assert {item["id"] for item in quota["limits"]} == {"codex", "codex-spark"}
    assert next(item for item in quota["limits"] if item["id"] == "codex-spark")["name"] == "GPT-5.3-Codex-Spark"


def test_codex_account_quota_reads_real_reset_credit_count(monkeypatch):
    monkeypatch.setattr("codexpi.system_service.codex_app_server_request", lambda *_: {
        "rateLimits": {"limitId": "codex", "primary": {"usedPercent": 17, "windowDurationMins": 10080, "resetsAt": 1789906246}},
        "rateLimitsByLimitId": {"codex": {"limitId": "codex", "primary": {"usedPercent": 17, "windowDurationMins": 10080, "resetsAt": 1789906246}}},
        "rateLimitResetCredits": {"availableCount": 3, "credits": [
            {"id": "one", "title": "Full reset", "description": "Granted", "expiresAt": 1790000000, "grantedAt": 1789000000, "status": "available", "resetType": "codexRateLimits"},
            {"id": "two", "title": "Full reset", "description": "Granted", "expiresAt": 1791000000, "grantedAt": 1789000000, "status": "available", "resetType": "codexRateLimits"},
            {"id": "three", "title": "Full reset", "description": "Granted", "expiresAt": 1792000000, "grantedAt": 1789000000, "status": "available", "resetType": "codexRateLimits"},
        ]},
    })
    quota = codex_account_quota("codex")
    assert quota["reset_credits"]["available_count"] == 3
    assert len(quota["reset_credits"]["credits"]) == 3
    assert quota["primary"]["remaining_percent"] == 83


def test_consume_codex_reset_credit_uses_selected_credit(monkeypatch):
    captured = {}
    def request(_binary, method, params):
        captured.update(method=method, params=params)
        return {"outcome": "reset"}
    monkeypatch.setattr("codexpi.system_service.codex_app_server_request", request)
    assert consume_codex_reset_credit("codex", "credit-one", "attempt-one") == "reset"
    assert captured == {"method": "account/rateLimitResetCredit/consume", "params": {"creditId": "credit-one", "idempotencyKey": "attempt-one"}}


def test_quota_reset_endpoint_requires_csrf_and_records_history(logged_in, monkeypatch):
    client, _, _, _, headers = logged_in
    monkeypatch.setattr("codexpi.main.consume_codex_reset_credit", lambda *_: "reset")
    assert client.post("/api/codex/quota/reset", json={"credit_id": "credit-one"}).status_code == 403
    response = client.post("/api/codex/quota/reset", headers=headers, json={"credit_id": "credit-one"})
    assert response.status_code == 200
    assert response.json()["outcome"] == "reset"
    assert len(client.get("/api/dashboard").json()["codex_quota"]["reset_history"]) == 1


def test_codex_quota_ignores_malformed_or_missing_data(tmp_path: Path):
    (tmp_path / "broken.jsonl").write_text('{broken}\n', encoding="utf-8")
    assert codex_quota_snapshot(tmp_path)["available"] is False


def test_file_diff_and_branch_data(logged_in):
    client, project, repo, _ = register(logged_in, "diff-repo")
    (repo / "app.py").write_text("print('after')\nprint('new')\n", encoding="utf-8")
    response = client.get(f"/api/projects/{project['id']}/git/file", params={"path": "app.py"})
    assert response.status_code == 200
    value = response.json()
    assert value["old_text"] == "print('before')\n"
    assert value["new_text"].splitlines()[-1] == "print('new')"
    assert value["added"] == 2 and value["deleted"] == 1
    values = asyncio.run(branches(repo))
    assert len(values) == 1 and values[0]["current"] is True
    direct = asyncio.run(file_contents(repo, "app.py"))
    assert direct["path"] == "app.py"


def test_websocket_reconnect_restores_same_session(logged_in):
    client, project, _, headers = register(logged_in, "reconnect-repo")
    conversation = client.post(
        f"/api/projects/{project['id']}/conversations", headers=headers, json={"title": "Persistent task"}
    ).json()
    for _ in range(2):
        with client.websocket_connect(
            f"/ws/conversations/{conversation['id']}", headers={"origin": "http://testserver"}
        ) as socket:
            state = socket.receive_json()
            assert state["type"] == "state"
            assert state["conversation"]["id"] == conversation["id"]


def test_websocket_rejects_missing_origin(logged_in):
    client, project, _, headers = register(logged_in, "origin-repo")
    conversation = client.post(
        f"/api/projects/{project['id']}/conversations", headers=headers, json={"title": "Origin check"}
    ).json()
    with pytest.raises(WebSocketDisconnect):
        with client.websocket_connect(f"/ws/conversations/{conversation['id']}"):
            pass


def test_expired_browser_session_is_rejected(logged_in):
    client, app, _, _, _ = logged_in
    past = (datetime.now(UTC) - timedelta(minutes=1)).isoformat()
    app.state.db.execute("UPDATE sessions SET expires_at=?", (past,))
    assert client.get("/api/auth/me").status_code == 401


def test_terminal_uses_persistent_scoped_tmux_session():
    value = terminal_argv("1234567890abcdef", "/usr/bin/tmux", "/bin/bash")
    assert value == ["/usr/bin/tmux", "new-session", "-A", "-s", "codexpi-1234567890ab"]
    assert "sudo" not in value
    assert terminal_argv("project", None, "/bin/bash") == ["/bin/bash"]


def test_pwa_assets_are_generated_and_declared():
    root = Path(__file__).resolve().parents[1] / "frontend"
    manifest = (root / "public" / "manifest.webmanifest").read_text(encoding="utf-8")
    service_worker = (root / "public" / "sw.js").read_text(encoding="utf-8")
    index = (root / "index.html").read_text(encoding="utf-8")
    assert '"display": "standalone"' in manifest
    assert "manifest.webmanifest" in index and "apple-touch-icon" in index
    assert "addEventListener('push'" in service_worker
    assert "addEventListener('notificationclick'" in service_worker
    for name in ("icon-192.png", "icon-512.png", "icon-maskable-512.png", "apple-touch-icon.png"):
        assert (root / "public" / "icons" / name).is_file()


def test_session_model_and_effort_options_are_validated(logged_in):
    client, project, _, headers = register(logged_in, "model-options")
    options = client.get("/api/codex/options")
    assert options.status_code == 200
    assert "gpt-5.6-sol" in [item["id"] for item in options.json()["models"]]
    created = client.post(
        f"/api/projects/{project['id']}/conversations",
        headers=headers,
        json={"title": "Configured", "model": "gpt-5.6-sol", "reasoning_effort": "high"},
    )
    assert created.status_code == 201
    assert created.json()["model"] == "gpt-5.6-sol"
    assert created.json()["reasoning_effort"] == "high"
    rejected = client.post(
        f"/api/projects/{project['id']}/conversations",
        headers=headers,
        json={"title": "Bad", "model": "shell; command", "reasoning_effort": "high"},
    )
    assert rejected.status_code == 400


def test_unnamed_sessions_are_numbered_auto_titled_and_archivable(logged_in, monkeypatch):
    client, app, _, headers = register(logged_in, "conversation-lifecycle")
    first = client.post(
        f"/api/projects/{app['id']}/conversations", headers=headers, json={"title": ""}
    )
    second = client.post(
        f"/api/projects/{app['id']}/conversations", headers=headers, json={}
    )
    assert first.status_code == 201 and first.json()["title"] == "Session 1"
    assert second.status_code == 201 and second.json()["title"] == "Session 2"

    codex = client.app.state.codex
    monkeypatch.setattr(codex, "detect", lambda: "/bin/true")
    monkeypatch.setattr(codex, "_spawn", lambda *_: 12345)
    queued = client.post(
        f"/api/conversations/{first.json()['id']}/queue",
        headers=headers,
        json={"content": "Ajouter un gestionnaire de dépôts élégant"},
    )
    assert queued.status_code == 202
    active = client.get(f"/api/projects/{app['id']}/conversations").json()
    assert next(item for item in active if item["id"] == first.json()["id"])["title"] == "Ajouter un gestionnaire de dépôts élégant"

    archived = client.put(
        f"/api/conversations/{first.json()['id']}/archive",
        headers=headers,
        json={"archived": True},
    )
    assert archived.status_code == 200
    assert first.json()["id"] not in {item["id"] for item in client.get(f"/api/projects/{app['id']}/conversations").json()}
    archived_rows = client.get(f"/api/projects/{app['id']}/conversations", params={"archived": True}).json()
    assert first.json()["id"] in {item["id"] for item in archived_rows}


def test_repository_can_be_initialized_committed_and_switched(logged_in):
    client, _, projects, _, headers = logged_in
    folder = projects / "new-repository"
    folder.mkdir()
    project = client.post(
        "/api/projects", headers=headers, json={"name": "New repository", "path": str(folder)}
    ).json()
    created = client.post(
        f"/api/projects/{project['id']}/git/initialize", headers=headers, json={}
    )
    assert created.status_code == 201 and created.json()["branch"] == "main"
    git(folder, "config", "user.email", "test@example.invalid")
    git(folder, "config", "user.name", "CodexPi Test")
    (folder / "app.py").write_text("print('hello')\n", encoding="utf-8")
    saved = client.post(
        f"/api/projects/{project['id']}/git/push",
        headers=headers,
        json={"message": "Initial project"},
    )
    assert saved.status_code == 200 and saved.json()["pushed"] is False
    assert client.get(f"/api/projects/{project['id']}/git").json()["history"][0]["subject"] == "Initial project"
    assert client.post(
        f"/api/projects/{project['id']}/git/branch", headers=headers, json={"name": "feature/demo"}
    ).status_code == 200
    assert client.post(
        f"/api/projects/{project['id']}/git/switch", headers=headers, json={"name": "main"}
    ).status_code == 200
    ignored = client.put(
        f"/api/projects/{project['id']}/git/gitignore",
        headers=headers,
        json={"content": ".env\n*.log\n"},
    )
    assert ignored.status_code == 200
    assert client.get(f"/api/projects/{project['id']}/git/gitignore").json()["content"] == ".env\n*.log\n"


def test_new_project_can_initialize_git_and_first_session_gets_a_worktree(logged_in):
    client, _, projects, _, headers = logged_in
    folder = projects / "from-zero"
    folder.mkdir()
    project_response = client.post(
        "/api/projects", headers=headers,
        json={"name": "From zero", "path": str(folder), "create_repository": True},
    )
    assert project_response.status_code == 201
    project = project_response.json()
    assert (folder / ".git").exists()
    conversation = client.post(
        f"/api/projects/{project['id']}/conversations", headers=headers,
        json={
            "title": "Feature", "create_branch": True, "branch_name": "feature/first-chat",
            "commit_description": "Première fonctionnalité isolée",
        },
    )
    assert conversation.status_code == 201, conversation.text
    value = conversation.json()
    assert value["branch_name"] == "feature/first-chat"
    assert value["base_branch"] == "main"
    assert Path(value["worktree_path"]).is_dir()
    assert subprocess.run(["git", "branch", "--show-current"], cwd=value["worktree_path"], capture_output=True, text=True).stdout.strip() == "feature/first-chat"