Developer
c26e9eacfa
- Delete dead repo-workspace code: RepoWorkspacePage, useRepoWorkspace, WorkspaceLayout, FileBrowser, old git components (git-toolbar, file-editor, commit-panel), and repo-workspace.css. - Fix stale backend test imports for moved models/services. - Add GitOperations unit tests. - Add integration tests for workspace files, git, and instances endpoints. - Add frontend tests for WorkspaceDetailPage and ProjectCard. - Update OpenSpec workspace-first-ui tasks and mark change completed. - Regenerate project maps. Quality gates: npm run typecheck, npm run lint, npm test -- --run (87 passed), python3 -m py_compile on changed backend files, pytest backend workspace tests.
143 lines
4.5 KiB
Python
143 lines
4.5 KiB
Python
"""Integration tests for workspace file endpoints."""
|
|
|
|
import asyncio
|
|
import os
|
|
import tempfile
|
|
import uuid
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from src.models import GitRepository, Project, Workspace
|
|
from src.services.shared.workspace_manager import WorkspaceManager
|
|
|
|
|
|
def _get_user_id(client: TestClient) -> uuid.UUID:
|
|
from src.auth.session import decode_session_cookie
|
|
from src.config import Settings
|
|
|
|
settings = Settings()
|
|
cookie = client.cookies.get("session")
|
|
data = decode_session_cookie(settings=settings, cookie_value=cookie)
|
|
if not data:
|
|
raise RuntimeError("No session")
|
|
return uuid.UUID(data["user_id"])
|
|
|
|
|
|
@pytest.fixture
|
|
def test_repo_with_workspace(
|
|
db_session: AsyncSession, authenticated_client: TestClient
|
|
):
|
|
"""Create a project, repo, and workspace backed by a temp directory."""
|
|
user_id = _get_user_id(authenticated_client)
|
|
|
|
async def _create():
|
|
project = Project(name="File Test Project", owner_id=user_id)
|
|
db_session.add(project)
|
|
await db_session.flush()
|
|
|
|
repo = GitRepository(
|
|
name="file-test-repo",
|
|
path="/tmp/file-test-repo",
|
|
remote_url="https://example.com/repo.git",
|
|
project_id=project.id,
|
|
owner_id=user_id,
|
|
)
|
|
db_session.add(repo)
|
|
await db_session.flush()
|
|
|
|
tmpdir = tempfile.mkdtemp()
|
|
ws = Workspace(
|
|
name="dev",
|
|
repo_id=repo.id,
|
|
user_id=user_id,
|
|
branch="main",
|
|
path=tmpdir,
|
|
)
|
|
db_session.add(ws)
|
|
await db_session.commit()
|
|
await db_session.refresh(ws)
|
|
return ws
|
|
|
|
return asyncio.run(_create())
|
|
|
|
|
|
class TestListWorkspaceFiles:
|
|
"""Tests for GET /workspaces/{id}/files."""
|
|
|
|
def test_list_files(
|
|
self, authenticated_client: TestClient, test_repo_with_workspace: Workspace
|
|
):
|
|
ws = test_repo_with_workspace
|
|
os.makedirs(os.path.join(ws.path, "src"))
|
|
with open(os.path.join(ws.path, "README.md"), "w") as f:
|
|
f.write("# Test")
|
|
|
|
response = authenticated_client.get(f"/workspaces/{ws.id}/files")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
names = {e["name"] for e in data["entries"]}
|
|
assert "src" in names
|
|
assert "README.md" in names
|
|
|
|
def test_list_files_path_escapes_workspace(
|
|
self, authenticated_client: TestClient, test_repo_with_workspace: Workspace
|
|
):
|
|
response = authenticated_client.get(
|
|
f"/workspaces/{test_repo_with_workspace.id}/files", params={"path": "../outside"}
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
class TestGetFileContent:
|
|
"""Tests for GET /workspaces/{id}/files/content."""
|
|
|
|
def test_get_content(
|
|
self, authenticated_client: TestClient, test_repo_with_workspace: Workspace
|
|
):
|
|
with open(os.path.join(test_repo_with_workspace.path, "hello.py"), "w") as f:
|
|
f.write("print('hello')")
|
|
|
|
response = authenticated_client.get(
|
|
f"/workspaces/{test_repo_with_workspace.id}/files/content",
|
|
params={"path": "hello.py"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["content"] == "print('hello')"
|
|
|
|
def test_get_missing_file(
|
|
self, authenticated_client: TestClient, test_repo_with_workspace: Workspace
|
|
):
|
|
response = authenticated_client.get(
|
|
f"/workspaces/{test_repo_with_workspace.id}/files/content",
|
|
params={"path": "missing.txt"},
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
class TestWriteFileContent:
|
|
"""Tests for POST /workspaces/{id}/files/content."""
|
|
|
|
def test_write_file(
|
|
self, authenticated_client: TestClient, test_repo_with_workspace: Workspace
|
|
):
|
|
response = authenticated_client.post(
|
|
f"/workspaces/{test_repo_with_workspace.id}/files/content",
|
|
json={"path": "nested/file.txt", "content": "content"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert os.path.exists(
|
|
os.path.join(test_repo_with_workspace.path, "nested", "file.txt")
|
|
)
|
|
|
|
def test_write_file_missing_path(
|
|
self, authenticated_client: TestClient, test_repo_with_workspace: Workspace
|
|
):
|
|
response = authenticated_client.post(
|
|
f"/workspaces/{test_repo_with_workspace.id}/files/content",
|
|
json={"content": "content"},
|
|
)
|
|
assert response.status_code == 400
|