374 lines
18 KiB
Python
374 lines
18 KiB
Python
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"
|