feat: complete workspace-first-ui cleanup and tests
- 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.
This commit is contained in:
Developer
@@ -2,7 +2,7 @@
|
||||
dir: apps/api/tests
|
||||
|
||||
## role
|
||||
Shared testing infrastructure and fixtures for the FastAPI API application
|
||||
Provides shared test infrastructure and fixtures for the FastAPI API application.
|
||||
## parent
|
||||
index: apps/api/.pi-map.index.md
|
||||
map: apps/api/.pi-map.md
|
||||
|
||||
@@ -4,11 +4,11 @@ dir: apps/api/tests
|
||||
index: apps/api/tests/.pi-map.index.md
|
||||
|
||||
## role
|
||||
Shared testing infrastructure and fixtures for the FastAPI API application
|
||||
Provides shared test infrastructure and fixtures for the FastAPI API application.
|
||||
## files
|
||||
- conftest.py | Provides shared pytest fixtures for FastAPI testing with async SQLite database, authenticated clients, and test data setup. | exp: func:test_client() → Generator[TestClient, None, None], call:create_async_engine, call:engine.begin, call:conn.run_sync, call:asyncio.run, call:init_db, call:async_sessionmaker, call:patch, call:TestClient, call:app.dependency_overrides.pop, call:engine.dispose, func:init_db(), call:engine.begin, call:conn.run_sync, func:override_get_db_session() → AsyncGenerator[AsyncSession, None], call:async_sessionmaker, func:db_session(test_client) → AsyncGenerator[AsyncSession, None], call:app.dependency_overrides.get, call:override_fn, call:gen.asend, call:gen.aclose, call:create_async_engine, call:engine.begin, call:conn.run_sync, call:async_sessionmaker, call:engine.dispose, func:authenticated_client(test_client) → Generator[TestClient, None, None], call:str, call:uuid.uuid4, call:Settings, call:app.dependency_overrides.get, call:override_fn, call:gen.asend, call:User, call:uuid.UUID, call:session.add, call:session.commit, call:gen.aclose, call:asyncio.run, call:create_test_user, call:create_session_cookie, call:test_client.cookies.set, func:create_test_user(), call:app.dependency_overrides.get, call:override_fn, call:gen.asend, call:User, call:uuid.UUID, call:session.add, call:session.commit, call:gen.aclose, func:test_project_and_repo(authenticated_client) → tuple[str, str], call:uuid.uuid4, call:Settings, call:authenticated_client.cookies.get, call:decode_session_cookie, call:uuid.UUID, call:asyncio.run, call:get_user_id, call:app.dependency_overrides.get, call:override_fn, call:gen.asend, call:Project, call:session.add, call:GitRepository, call:session.commit, call:gen.aclose, call:create_project_and_repo, call:str, raise:RuntimeError, func:get_user_id(), call:Settings, call:authenticated_client.cookies.get, call:decode_session_cookie, call:uuid.UUID, func:create_project_and_repo(), call:app.dependency_overrides.get, call:override_fn, call:gen.asend, call:Project, call:session.add, call:GitRepository, call:session.commit, call:gen.aclose, func:admin_client(test_client) → Generator[TestClient, None, None], call:str, call:uuid.uuid4, call:Settings, call:app.dependency_overrides.get, call:override_fn, call:gen.asend, call:User, call:uuid.UUID, call:session.add, call:session.commit, call:gen.aclose, call:asyncio.run, call:create_admin_user, call:create_session_cookie, call:test_client.cookies.set, func:create_admin_user(), call:app.dependency_overrides.get, call:override_fn, call:gen.asend, call:User, call:uuid.UUID, call:session.add, call:session.commit, call:gen.aclose | dep: asyncio, os, typing, unittest.mock, pytest, pytest_asyncio, fastapi.testclient, sqlalchemy.ext.asyncio, src.config, src.models.base, src.main, src.auth.dependencies, uuid, src.auth.session, src.models.user, src.models.project, src.models.git_repository, fastapi, sqlalchemy, src.models, src.auth
|
||||
## arch
|
||||
Pytest fixture-based dependency injection pattern with async SQLite test database, authenticated HTTP client factories, and modular test data setup utilities
|
||||
Pytest fixture-based testing architecture with async SQLite database setup, authenticated HTTP client factories, and test data seeding patterns.
|
||||
## tags
|
||||
call:app.dependency, call:create, overrides.get, call:override, fn, call:gen.asend, call:gen.aclose, session
|
||||
## symbols
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
dir: apps/api/tests/integration
|
||||
|
||||
## role
|
||||
Integration and unit test suite for the API backend, covering authentication, REST endpoints, database models, and core services with real database dependencies.
|
||||
Integration and unit test suite for the FastAPI backend API, covering authentication, CRUD APIs, workspace/git operations, and event/notification systems using real databases and external dependencies.
|
||||
## parent
|
||||
index: apps/api/tests/.pi-map.index.md
|
||||
map: apps/api/tests/.pi-map.md
|
||||
@@ -24,6 +24,9 @@ map: apps/api/tests/.pi-map.md
|
||||
- test_tool_types_api.py
|
||||
- test_tool_types_api_extended.py
|
||||
- test_users_api.py
|
||||
- test_workspace_files.py
|
||||
- test_workspace_git.py
|
||||
- test_workspace_instances.py
|
||||
- test_workspaces_api.py
|
||||
## links
|
||||
index: apps/api/tests/integration/.pi-map.index.md
|
||||
|
||||
@@ -4,7 +4,7 @@ dir: apps/api/tests/integration
|
||||
index: apps/api/tests/integration/.pi-map.index.md
|
||||
|
||||
## role
|
||||
Integration and unit test suite for the API backend, covering authentication, REST endpoints, database models, and core services with real database dependencies.
|
||||
Integration and unit test suite for the FastAPI backend API, covering authentication, CRUD APIs, workspace/git operations, and event/notification systems using real databases and external dependencies.
|
||||
## files
|
||||
- __init__.py | Empty file with no functionality
|
||||
- test_auth_api.py | Integration tests for FastAPI authentication endpoints (login, me, logout) using a real database and session cookies. | exp: func:_prepare_auth_test_db() → None, call:create_async_engine, call:build_database_url, call:engine.begin, call:connection.run_sync, call:connection.execute, call:text, call:engine.dispose, call:asyncio.run, call:_run, func:_run() → None, call:create_async_engine, call:build_database_url, call:engine.begin, call:connection.run_sync, call:connection.execute, call:text, call:engine.dispose, func:_load_app(), call:importlib.reload, func:_insert_test_user(user_id: str) → None, call:create_async_engine, call:build_database_url, call:engine.begin, call:connection.run_sync, call:async_sessionmaker, call:session_factory, call:User, call:uuid.UUID, call:session.merge, call:session.commit, call:engine.dispose, call:asyncio.run, call:_run, func:_run() → None, call:create_async_engine, call:build_database_url, call:engine.begin, call:connection.run_sync, call:async_sessionmaker, call:session_factory, call:User, call:uuid.UUID, call:session.merge, call:session.commit, call:engine.dispose, func:test_login_redirects_to_authentik_authorize_endpoint() → None, call:_prepare_auth_test_db, call:_load_app, call:TestClient, call:client.get, func:test_me_returns_401_without_session_cookie() → None, call:_prepare_auth_test_db, call:_load_app, call:TestClient, call:client.get, func:test_me_returns_user_with_valid_session() → None, call:_prepare_auth_test_db, call:_insert_test_user, call:_load_app, call:Settings, call:create_session_cookie, call:TestClient, call:client.get, call:response.json, func:test_logout_clears_session_cookie() → None, call:_prepare_auth_test_db, call:_load_app, call:TestClient, call:client.post, call:response.headers.get | dep: uuid, asyncio, importlib, fastapi.testclient, pytest, sqlalchemy, sqlalchemy.ext.asyncio, src.auth.session, src.config, src.models, src.models.user, fastapi, src.database, src.api.auth, src.main
|
||||
@@ -21,20 +21,23 @@ Integration and unit test suite for the API backend, covering authentication, RE
|
||||
- test_tool_types_api.py | Integration tests for a FastAPI tool types REST API endpoint covering CRUD operations, authentication, and validation rules | exp: func:_prepare_test_db() → None, call:create_async_engine, call:build_database_url, call:engine.begin, call:connection.run_sync, call:connection.execute, call:text, call:engine.dispose, call:asyncio.run, call:_run, func:_run() → None, call:create_async_engine, call:build_database_url, call:engine.begin, call:connection.run_sync, call:connection.execute, call:text, call:engine.dispose, func:_load_app(), call:hasattr, call:asyncio.run, call:database_module.engine.dispose, call:importlib.reload, func:_mint_token(user_id: str) → str, call:Settings, call:create_session_cookie, call:datetime.now, call:timedelta, func:_insert_user(user_id: str, email) → None, call:create_async_engine, call:build_database_url, call:engine.begin, call:connection.run_sync, call:async_sessionmaker, call:session_factory, call:User, call:uuid.UUID, call:session.merge, call:session.commit, call:engine.dispose, call:asyncio.run, call:_run, func:_run() → None, call:create_async_engine, call:build_database_url, call:engine.begin, call:connection.run_sync, call:async_sessionmaker, call:session_factory, call:User, call:uuid.UUID, call:session.merge, call:session.commit, call:engine.dispose, func:_insert_tool_type(tool_type_id: str, name: str, display_name: str, compose_template: str, created_by_id) → None, call:create_async_engine, call:build_database_url, call:async_sessionmaker, call:session_factory, call:ToolType, call:uuid.UUID, call:session.merge, call:session.commit, call:engine.dispose, call:asyncio.run, call:_run, func:_run() → None, call:create_async_engine, call:build_database_url, call:async_sessionmaker, call:session_factory, call:ToolType, call:uuid.UUID, call:session.merge, call:session.commit, call:engine.dispose, func:test_list_tool_types_requires_authentication() → None, call:_prepare_test_db, call:_load_app, call:TestClient, call:client.get, func:test_list_tool_types_returns_all_types() → None, call:_prepare_test_db, call:_insert_user, call:_insert_tool_type, call:_load_app, call:TestClient, call:client.cookies.set, call:_mint_token, call:client.get, call:response.json, call:len, call:next, func:test_get_tool_type_by_id() → None, call:_prepare_test_db, call:_insert_user, call:_insert_tool_type, call:_load_app, call:TestClient, call:client.cookies.set, call:_mint_token, call:client.get, call:response.json, func:test_get_tool_type_not_found() → None, call:_prepare_test_db, call:_insert_user, call:_load_app, call:TestClient, call:client.cookies.set, call:_mint_token, call:client.get, func:test_create_tool_type_successfully() → None, call:_prepare_test_db, call:_insert_user, call:_load_app, call:TestClient, call:client.cookies.set, call:_mint_token, call:client.post, call:response.json, func:test_create_tool_type_duplicate_name() → None, call:_prepare_test_db, call:_insert_user, call:_insert_tool_type, call:_load_app, call:TestClient, call:client.cookies.set, call:_mint_token, call:client.post, func:test_create_tool_type_invalid_yaml() → None, call:_prepare_test_db, call:_insert_user, call:_load_app, call:TestClient, call:client.cookies.set, call:_mint_token, call:client.post, func:test_create_tool_type_missing_services() → None, call:_prepare_test_db, call:_insert_user, call:_load_app, call:TestClient, call:client.cookies.set, call:_mint_token, call:client.post, func:test_create_tool_type_missing_required_variable() → None, call:_prepare_test_db, call:_insert_user, call:_load_app, call:TestClient, call:client.cookies.set, call:_mint_token, call:client.post, func:test_update_tool_type_successfully() → None, call:_prepare_test_db, call:_insert_user, call:_insert_tool_type, call:_load_app, call:TestClient, call:client.cookies.set, call:_mint_token, call:client.put, call:response.json, func:test_update_tool_type_not_found() → None, call:_prepare_test_db, call:_insert_user, call:_load_app, call:TestClient, call:client.cookies.set, call:_mint_token, call:client.put, func:test_delete_tool_type_successfully() → None, call:_prepare_test_db, call:_insert_user, call:_insert_tool_type, call:_load_app, call:TestClient, call:client.cookies.set, call:_mint_token, call:client.delete, call:client.get, func:test_delete_tool_type_not_found() → None, call:_prepare_test_db, call:_insert_user, call:_load_app, call:TestClient, call:client.cookies.set, call:_mint_token, call:client.delete | dep: uuid, datetime, asyncio, pytest, fastapi.testclient, sqlalchemy, sqlalchemy.ext.asyncio, src.auth.session, src.config, src.models, src.models.tool_type, src.models.user, importlib, fastapi, src.database, src.api.auth, src.api.tool_types, src.main
|
||||
- test_tool_types_api_extended.py | Integration tests for a FastAPI tool types API endpoint covering CRUD operations with extended fields including dockerfile templates, compose templates, readiness probes, startup commands, and validation logic. | exp: class:TestToolTypesAPIExtended, method:test_create_tool_type_with_dockerfile(self, authenticated_client: TestClient) → None, call:authenticated_client.post, call:response.json, method:test_create_tool_type_with_readiness_probe(self, authenticated_client: TestClient) → None, call:authenticated_client.post, call:response.json, method:test_create_tool_type_invalid_definition_type(self, authenticated_client: TestClient) → None, call:authenticated_client.post, method:test_create_tool_type_dockerfile_without_template(self, authenticated_client: TestClient) → None, call:authenticated_client.post, method:test_update_tool_type_with_new_fields(self, authenticated_client: TestClient) → None, call:authenticated_client.post, call:create_response.json, call:authenticated_client.put, call:response.json, method:test_validate_tool_type_compose(self, authenticated_client: TestClient) → None, call:authenticated_client.post, call:response.json, method:test_validate_tool_type_invalid_compose(self, authenticated_client: TestClient) → None, call:authenticated_client.post, call:response.json, method:test_validate_tool_type_dockerfile(self, authenticated_client: TestClient) → None, call:authenticated_client.post, call:response.json, method:test_get_tool_type_returns_new_fields(self, authenticated_client: TestClient) → None, call:authenticated_client.post, call:create_response.json, call:authenticated_client.get, call:response.json, method:test_create_tool_type_without_port_fails(self, authenticated_client: TestClient) → None, call:authenticated_client.post, call:response.json, call:str, method:test_create_tool_type_with_port_mismatch_fails(self, authenticated_client: TestClient) → None, call:authenticated_client.post, call:response.json, method:test_create_tool_type_with_startup_command(self, authenticated_client: TestClient) → None, call:authenticated_client.post, call:response.json, method:test_update_tool_type_startup_command(self, authenticated_client: TestClient) → None, call:authenticated_client.post, call:create_response.json, call:authenticated_client.put, call:response.json, method:test_get_tool_type_returns_startup_command(self, authenticated_client: TestClient) → None, call:authenticated_client.post, call:create_response.json, call:authenticated_client.get, call:response.json, call:str | dep: pytest, fastapi.testclient, fastapi.testclient.TestClient
|
||||
- test_users_api.py | Integration tests for user profile API endpoints including get, update, and avatar upload with authentication and validation | exp: func:_prepare_users_test_db() → None, call:create_async_engine, call:build_database_url, call:engine.begin, call:connection.run_sync, call:connection.execute, call:text, call:engine.dispose, call:asyncio.run, call:_run, func:_run() → None, call:create_async_engine, call:build_database_url, call:engine.begin, call:connection.run_sync, call:connection.execute, call:text, call:engine.dispose, func:_load_app(), call:importlib.reload, func:_insert_test_user(user_id: str) → None, call:create_async_engine, call:build_database_url, call:engine.begin, call:connection.run_sync, call:async_sessionmaker, call:session_factory, call:User, call:uuid.UUID, call:session.merge, call:session.commit, call:engine.dispose, call:asyncio.run, call:_run, func:_run() → None, call:create_async_engine, call:build_database_url, call:engine.begin, call:connection.run_sync, call:async_sessionmaker, call:session_factory, call:User, call:uuid.UUID, call:session.merge, call:session.commit, call:engine.dispose, func:_create_auth_cookie(user_id: str) → str, call:Settings, call:create_session_cookie, call:datetime.now, call:timedelta, func:test_get_profile_returns_401_without_cookie() → None, call:_prepare_users_test_db, call:_load_app, call:TestClient, call:client.get, func:test_get_profile_returns_user_data() → None, call:_prepare_users_test_db, call:_insert_test_user, call:_load_app, call:TestClient, call:client.cookies.set, call:_create_auth_cookie, call:client.get, call:response.json, func:test_update_profile_changes_name_and_email() → None, call:_prepare_users_test_db, call:_insert_test_user, call:_load_app, call:TestClient, call:client.cookies.set, call:_create_auth_cookie, call:client.put, call:response.json, func:test_update_profile_rejects_empty_name() → None, call:_prepare_users_test_db, call:_insert_test_user, call:_load_app, call:TestClient, call:client.cookies.set, call:_create_auth_cookie, call:client.put, func:test_update_profile_rejects_invalid_email() → None, call:_prepare_users_test_db, call:_insert_test_user, call:_load_app, call:TestClient, call:client.cookies.set, call:_create_auth_cookie, call:client.put, func:test_upload_avatar_updates_avatar_url() → None, call:_prepare_users_test_db, call:_insert_test_user, call:_load_app, call:TestClient, call:client.cookies.set, call:_create_auth_cookie, call:client.post, call:io.BytesIO, call:response.json, call:data["avatar_url"].startswith, func:test_upload_avatar_rejects_invalid_file_type() → None, call:_prepare_users_test_db, call:_insert_test_user, call:_load_app, call:TestClient, call:client.cookies.set, call:_create_auth_cookie, call:client.post, call:io.BytesIO, func:test_upload_avatar_rejects_oversized_file() → None, call:_prepare_users_test_db, call:_insert_test_user, call:_load_app, call:TestClient, call:client.cookies.set, call:_create_auth_cookie, call:client.post, call:io.BytesIO | dep: uuid, datetime, asyncio, io, fastapi.testclient, pytest, sqlalchemy, sqlalchemy.ext.asyncio, src.auth.session, src.config, src.models, src.models.user, importlib, fastapi, src.database, src.api.users, src.main
|
||||
- test_workspaces_api.py | Integration tests for FastAPI workspace API endpoints covering CRUD, sync, and deletion operations with mocked services. | exp: class:TestListWorkspaces, method:test_list_empty(self, authenticated_client: TestClient, test_repo: GitRepository), call:authenticated_client.get, call:response.json, method:test_list_with_workspaces(self, authenticated_client: TestClient, db_session: AsyncSession, test_repo: GitRepository), call:Workspace, call:db_session.add, call:db_session.commit, call:asyncio.run, call:_commit, call:authenticated_client.get, call:response.json, call:len, class:TestCreateWorkspace, method:test_create_success(self, authenticated_client: TestClient, test_repo: GitRepository), call:Workspace, call:uuid.uuid4, call:patch.object, call:authenticated_client.post, call:response.json, call:mock_create.assert_called_once, method:test_create_missing_name(self, authenticated_client: TestClient, test_repo: GitRepository), call:authenticated_client.post, call:response.json, method:test_create_duplicate_name(self, authenticated_client: TestClient, db_session: AsyncSession, test_repo: GitRepository), call:Workspace, call:db_session.add, call:db_session.commit, call:asyncio.run, call:_commit, call:patch.object, call:Exception, call:authenticated_client.post, class:TestDeleteWorkspace, method:test_delete_without_instances(self, authenticated_client: TestClient, db_session: AsyncSession, test_repo: GitRepository), call:Workspace, call:db_session.add, call:db_session.commit, call:db_session.refresh, call:asyncio.run, call:_commit_refresh, call:patch.object, call:authenticated_client.delete, call:response.json, method:test_delete_with_instances_force(self, authenticated_client: TestClient, db_session: AsyncSession, test_repo: GitRepository), call:Workspace, call:db_session.add, call:db_session.commit, call:db_session.refresh, call:asyncio.run, call:_commit_refresh, call:patch.object, call:authenticated_client.delete, class:TestSyncWorkspace, method:test_sync_success(self, authenticated_client: TestClient, db_session: AsyncSession, test_repo: GitRepository), call:Workspace, call:db_session.add, call:db_session.commit, call:db_session.refresh, call:asyncio.run, call:_commit_refresh, call:patch.object, call:MagicMock, call:authenticated_client.post, call:response.json, method:test_sync_branch_deleted(self, authenticated_client: TestClient, db_session: AsyncSession, test_repo: GitRepository), call:Workspace, call:db_session.add, call:db_session.commit, call:db_session.refresh, call:asyncio.run, call:_commit_refresh, call:patch.object, call:MagicMock, call:authenticated_client.post, call:response.json, func:_get_user_id_from_client(client: TestClient) → uuid.UUID, call:Settings, call:client.cookies.get, call:decode_session_cookie, call:uuid.UUID, raise:RuntimeError, func:test_repo(db_session: AsyncSession, authenticated_client: TestClient), call:_get_user_id_from_client, call:Project, call:db_session.add, call:db_session.flush, call:GitRepository, call:db_session.commit, call:db_session.refresh, call:asyncio.run, call:_create, func:_create(), call:Project, call:db_session.add, call:db_session.flush, call:GitRepository, call:db_session.commit, call:db_session.refresh | dep: asyncio, uuid, unittest.mock, pytest, fastapi.testclient, sqlalchemy.ext.asyncio, src.models.git_repository, src.models.project, src.models.tool_instance, src.models.tool_type, src.models.workspace, src.services.workspace_manager, src.auth.session, src.config, fastapi, sqlalchemy, src.models
|
||||
- test_workspace_files.py | Integration tests for workspace file endpoints (list, read, write) with path traversal security checks | exp: class:TestListWorkspaceFiles, method:test_list_files(self, authenticated_client: TestClient, test_repo_with_workspace: Workspace), call:os.makedirs, call:os.path.join, call:open, call:f.write, call:authenticated_client.get, call:response.json, method:test_list_files_path_escapes_workspace(self, authenticated_client: TestClient, test_repo_with_workspace: Workspace), call:authenticated_client.get, class:TestGetFileContent, method:test_get_content(self, authenticated_client: TestClient, test_repo_with_workspace: Workspace), call:open, call:os.path.join, call:f.write, call:authenticated_client.get, call:response.json, method:test_get_missing_file(self, authenticated_client: TestClient, test_repo_with_workspace: Workspace), call:authenticated_client.get, class:TestWriteFileContent, method:test_write_file(self, authenticated_client: TestClient, test_repo_with_workspace: Workspace), call:authenticated_client.post, call:os.path.exists, call:os.path.join, method:test_write_file_missing_path(self, authenticated_client: TestClient, test_repo_with_workspace: Workspace), call:authenticated_client.post, func:_get_user_id(client: TestClient) → uuid.UUID, call:Settings, call:client.cookies.get, call:decode_session_cookie, call:uuid.UUID, raise:RuntimeError, func:test_repo_with_workspace(db_session: AsyncSession, authenticated_client: TestClient), call:_get_user_id, call:Project, call:db_session.add, call:db_session.flush, call:GitRepository, call:tempfile.mkdtemp, call:Workspace, call:db_session.commit, call:db_session.refresh, call:asyncio.run, call:_create, func:_create(), call:Project, call:db_session.add, call:db_session.flush, call:GitRepository, call:tempfile.mkdtemp, call:Workspace, call:db_session.commit, call:db_session.refresh | dep: asyncio, os, tempfile, uuid, unittest.mock, pytest, fastapi.testclient, sqlalchemy.ext.asyncio, src.models, src.services.shared.workspace_manager, src.auth.session, src.config, fastapi.testclient.TestClient, sqlalchemy.ext.asyncio.AsyncSession
|
||||
- test_workspace_git.py | Integration tests for workspace Git endpoints (status, commit, history, checkout) using real git repositories and FastAPI test client. | exp: class:TestGitStatus, method:test_status_clean(self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace), call:authenticated_client.get, call:response.json, method:test_status_modified(self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace), call:open, call:os.path.join, call:f.write, call:authenticated_client.get, call:response.json, class:TestGitCommit, method:test_commit(self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace), call:open, call:os.path.join, call:f.write, call:authenticated_client.post, call:response.json, method:test_commit_missing_message(self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace), call:authenticated_client.post, class:TestGitHistory, method:test_history(self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace), call:authenticated_client.get, call:response.json, call:len, class:TestGitCheckout, method:test_checkout(self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace), call:_run_git, call:authenticated_client.post, call:response.json, func:_run_git(*args: str, cwd: str) → None, call:subprocess.run, func:_get_user_id(client: TestClient) → uuid.UUID, call:Settings, call:client.cookies.get, call:decode_session_cookie, call:uuid.UUID, raise:RuntimeError, func:test_repo_with_git_workspace(db_session: AsyncSession, authenticated_client: TestClient), call:_get_user_id, call:Project, call:db_session.add, call:db_session.flush, call:GitRepository, call:tempfile.mkdtemp, call:_run_git, call:open, call:os.path.join, call:f.write, call:Workspace, call:db_session.commit, call:db_session.refresh, call:asyncio.run, call:_create, func:_create(), call:Project, call:db_session.add, call:db_session.flush, call:GitRepository, call:tempfile.mkdtemp, call:_run_git, call:open, call:os.path.join, call:f.write, call:Workspace, call:db_session.commit, call:db_session.refresh | dep: asyncio, os, subprocess, tempfile, uuid, pytest, fastapi.testclient, sqlalchemy.ext.asyncio, src.models, src.auth.session, src.config, fastapi.testclient.TestClient, sqlalchemy.ext.asyncio.AsyncSession
|
||||
- test_workspace_instances.py | Integration tests for workspace instance REST API endpoints (listing and creating tool instances within workspaces) | exp: class:TestListWorkspaceInstances, method:test_list_empty(self, authenticated_client: TestClient, test_workspace_with_tool_type), call:authenticated_client.get, call:response.json, method:test_list_instances(self, authenticated_client: TestClient, db_session: AsyncSession, test_workspace_with_tool_type), call:ToolInstance, call:db_session.add, call:db_session.commit, call:asyncio.run, call:_create_instance, call:authenticated_client.get, call:response.json, call:len, class:TestCreateWorkspaceInstance, method:test_create_instance(self, authenticated_client: TestClient, test_workspace_with_tool_type), call:ToolInstance, call:uuid.uuid4, call:datetime.now, call:patch, call:authenticated_client.post, call:str, call:response.json, func:_get_user_id(client: TestClient) → uuid.UUID, call:Settings, call:client.cookies.get, call:decode_session_cookie, call:uuid.UUID, raise:RuntimeError, func:test_workspace_with_tool_type(db_session: AsyncSession, authenticated_client: TestClient), call:_get_user_id, call:Project, call:db_session.add, call:db_session.flush, call:GitRepository, call:tempfile.mkdtemp, call:Workspace, call:ToolType, call:db_session.commit, call:db_session.refresh, call:asyncio.run, call:_create, func:_create(), call:Project, call:db_session.add, call:db_session.flush, call:GitRepository, call:tempfile.mkdtemp, call:Workspace, call:ToolType, call:db_session.commit, call:db_session.refresh | dep: asyncio, tempfile, uuid, datetime, unittest.mock, pytest, fastapi.testclient, sqlalchemy.ext.asyncio, src.models, src.auth.session, src.config, src.api.workspace.workspace_instances
|
||||
- test_workspaces_api.py | Integration tests for workspace API endpoints including list, create, delete, and sync operations with repository/project scoping | exp: class:TestListWorkspaces, method:test_list_empty(self, authenticated_client: TestClient, test_repo: GitRepository), call:authenticated_client.get, call:response.json, method:test_list_with_workspaces(self, authenticated_client: TestClient, db_session: AsyncSession, test_repo: GitRepository), call:Workspace, call:db_session.add, call:db_session.commit, call:asyncio.run, call:_commit, call:authenticated_client.get, call:response.json, call:len, class:TestCreateWorkspace, method:test_create_success(self, authenticated_client: TestClient, test_repo: GitRepository), call:Workspace, call:uuid.uuid4, call:patch.object, call:authenticated_client.post, call:response.json, call:mock_create.assert_called_once, method:test_create_missing_name(self, authenticated_client: TestClient, test_repo: GitRepository), call:authenticated_client.post, call:response.json, method:test_create_duplicate_name(self, authenticated_client: TestClient, db_session: AsyncSession, test_repo: GitRepository), call:Workspace, call:db_session.add, call:db_session.commit, call:asyncio.run, call:_commit, call:patch.object, call:Exception, call:authenticated_client.post, class:TestDeleteWorkspace, method:test_delete_without_instances(self, authenticated_client: TestClient, db_session: AsyncSession, test_repo: GitRepository), call:Workspace, call:db_session.add, call:db_session.commit, call:db_session.refresh, call:asyncio.run, call:_commit_refresh, call:patch.object, call:authenticated_client.delete, call:response.json, method:test_delete_with_instances_force(self, authenticated_client: TestClient, db_session: AsyncSession, test_repo: GitRepository), call:Workspace, call:db_session.add, call:db_session.commit, call:db_session.refresh, call:asyncio.run, call:_commit_refresh, call:patch.object, call:authenticated_client.delete, class:TestSyncWorkspace, method:test_sync_success(self, authenticated_client: TestClient, db_session: AsyncSession, test_repo: GitRepository), call:Workspace, call:db_session.add, call:db_session.commit, call:db_session.refresh, call:asyncio.run, call:_commit_refresh, call:patch.object, call:MagicMock, call:authenticated_client.post, call:response.json, method:test_sync_branch_deleted(self, authenticated_client: TestClient, db_session: AsyncSession, test_repo: GitRepository), call:Workspace, call:db_session.add, call:db_session.commit, call:db_session.refresh, call:asyncio.run, call:_commit_refresh, call:patch.object, call:MagicMock, call:authenticated_client.post, call:response.json, func:_get_user_id_from_client(client: TestClient) → uuid.UUID, call:Settings, call:client.cookies.get, call:decode_session_cookie, call:uuid.UUID, raise:RuntimeError, func:test_repo(db_session: AsyncSession, authenticated_client: TestClient), call:_get_user_id_from_client, call:Project, call:db_session.add, call:db_session.flush, call:GitRepository, call:db_session.commit, call:db_session.refresh, call:asyncio.run, call:_create, func:_create(), call:Project, call:db_session.add, call:db_session.flush, call:GitRepository, call:db_session.commit, call:db_session.refresh | dep: asyncio, uuid, unittest.mock, pytest, fastapi.testclient, sqlalchemy.ext.asyncio, src.models, src.services.shared.workspace_manager, src.auth.session, src.config, fastapi, sqlalchemy
|
||||
## arch
|
||||
Test pyramid with integration tests dominating; uses FastAPI TestClient, real SQLAlchemy database sessions, session cookie authentication, pub/sub event bus mocking, and service-layer mocking for external dependencies (git, SSH, workspaces).
|
||||
Test-driven pyramid with integration tests as primary layer using FastAPI TestClient, real SQLAlchemy databases, actual git repositories, and pub/sub event bus; unit tests limited to isolated git utilities; shared fixtures likely in conftest.py for dependency injection, database setup/teardown, and authenticated client sessions.
|
||||
## tags
|
||||
test, call:, call:authenticated, user, call:db, call:create, create, call:response.json
|
||||
test, call:, call:db, call:authenticated, user, call:response.json, create, call:create
|
||||
## symbols
|
||||
- TestConfigProfilesAPI
|
||||
- TestGitStatus
|
||||
- TestBranchOperations
|
||||
- TestCommit
|
||||
- TestToolTypesAPIExtended
|
||||
- TestListWorkspaces
|
||||
- TestCreateWorkspace
|
||||
- TestDeleteWorkspace
|
||||
- TestListWorkspaceFiles
|
||||
- TestGetFileContent
|
||||
- TestWriteFileContent
|
||||
## workflows
|
||||
- change integration behavior
|
||||
read: __init__.py, test_auth_api.py, test_auth_services.py
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
"""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
|
||||
@@ -0,0 +1,160 @@
|
||||
"""Integration tests for workspace git endpoints."""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.models import GitRepository, Project, Workspace
|
||||
|
||||
|
||||
def _run_git(*args: str, cwd: str) -> None:
|
||||
subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True)
|
||||
|
||||
|
||||
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_git_workspace(
|
||||
db_session: AsyncSession, authenticated_client: TestClient
|
||||
):
|
||||
"""Create a workspace backed by a real git repo with commits."""
|
||||
user_id = _get_user_id(authenticated_client)
|
||||
|
||||
async def _create():
|
||||
project = Project(name="Git Test Project", owner_id=user_id)
|
||||
db_session.add(project)
|
||||
await db_session.flush()
|
||||
|
||||
repo = GitRepository(
|
||||
name="git-test-repo",
|
||||
path="/tmp/git-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()
|
||||
_run_git("init", cwd=tmpdir)
|
||||
_run_git("config", "user.email", "test@test.com", cwd=tmpdir)
|
||||
_run_git("config", "user.name", "Test User", cwd=tmpdir)
|
||||
with open(os.path.join(tmpdir, "README.md"), "w") as f:
|
||||
f.write("# Test\n")
|
||||
_run_git("add", "README.md", cwd=tmpdir)
|
||||
_run_git("commit", "-m", "Initial commit", cwd=tmpdir)
|
||||
|
||||
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 TestGitStatus:
|
||||
"""Tests for GET /workspaces/{id}/git/status."""
|
||||
|
||||
def test_status_clean(
|
||||
self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace
|
||||
):
|
||||
response = authenticated_client.get(
|
||||
f"/workspaces/{test_repo_with_git_workspace.id}/git/status"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["branch"] == "main"
|
||||
assert data["modified"] == []
|
||||
|
||||
def test_status_modified(
|
||||
self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace
|
||||
):
|
||||
with open(os.path.join(test_repo_with_git_workspace.path, "README.md"), "a") as f:
|
||||
f.write("modified\n")
|
||||
|
||||
response = authenticated_client.get(
|
||||
f"/workspaces/{test_repo_with_git_workspace.id}/git/status"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert "README.md" in response.json()["modified"]
|
||||
|
||||
|
||||
class TestGitCommit:
|
||||
"""Tests for POST /workspaces/{id}/git/commit."""
|
||||
|
||||
def test_commit(
|
||||
self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace
|
||||
):
|
||||
with open(os.path.join(test_repo_with_git_workspace.path, "new.py"), "w") as f:
|
||||
f.write("print('hello')\n")
|
||||
|
||||
response = authenticated_client.post(
|
||||
f"/workspaces/{test_repo_with_git_workspace.id}/git/commit",
|
||||
json={"message": "Add new file"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["status"] == "committed"
|
||||
|
||||
def test_commit_missing_message(
|
||||
self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace
|
||||
):
|
||||
response = authenticated_client.post(
|
||||
f"/workspaces/{test_repo_with_git_workspace.id}/git/commit",
|
||||
json={},
|
||||
)
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
class TestGitHistory:
|
||||
"""Tests for GET /workspaces/{id}/git/history."""
|
||||
|
||||
def test_history(
|
||||
self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace
|
||||
):
|
||||
response = authenticated_client.get(
|
||||
f"/workspaces/{test_repo_with_git_workspace.id}/git/history"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert len(data["commits"]) >= 1
|
||||
assert data["commits"][0]["message"] == "Initial commit"
|
||||
|
||||
|
||||
class TestGitCheckout:
|
||||
"""Tests for POST /workspaces/{id}/git/checkout."""
|
||||
|
||||
def test_checkout(
|
||||
self, authenticated_client: TestClient, test_repo_with_git_workspace: Workspace
|
||||
):
|
||||
_run_git("checkout", "-b", "feature", cwd=test_repo_with_git_workspace.path)
|
||||
_run_git("checkout", "main", cwd=test_repo_with_git_workspace.path)
|
||||
|
||||
response = authenticated_client.post(
|
||||
f"/workspaces/{test_repo_with_git_workspace.id}/git/checkout",
|
||||
json={"branch": "feature"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["branch"] == "feature"
|
||||
@@ -0,0 +1,151 @@
|
||||
"""Integration tests for workspace instance endpoints."""
|
||||
|
||||
import asyncio
|
||||
import tempfile
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.models import GitRepository, Project, ToolInstance, ToolType, Workspace
|
||||
|
||||
|
||||
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_workspace_with_tool_type(
|
||||
db_session: AsyncSession, authenticated_client: TestClient
|
||||
):
|
||||
"""Create a project, repo, workspace, and tool type."""
|
||||
user_id = _get_user_id(authenticated_client)
|
||||
|
||||
async def _create():
|
||||
project = Project(name="Instance Test Project", owner_id=user_id)
|
||||
db_session.add(project)
|
||||
await db_session.flush()
|
||||
|
||||
repo = GitRepository(
|
||||
name="instance-test-repo",
|
||||
path="/tmp/instance-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.flush()
|
||||
|
||||
tool_type = ToolType(
|
||||
name="test-tool",
|
||||
display_name="Test Tool",
|
||||
default_port=8080,
|
||||
category="dev",
|
||||
)
|
||||
db_session.add(tool_type)
|
||||
await db_session.commit()
|
||||
await db_session.refresh(ws)
|
||||
await db_session.refresh(tool_type)
|
||||
return ws, tool_type, project
|
||||
|
||||
return asyncio.run(_create())
|
||||
|
||||
|
||||
class TestListWorkspaceInstances:
|
||||
"""Tests for GET /workspaces/{id}/instances."""
|
||||
|
||||
def test_list_empty(
|
||||
self, authenticated_client: TestClient, test_workspace_with_tool_type
|
||||
):
|
||||
ws, _, _ = test_workspace_with_tool_type
|
||||
response = authenticated_client.get(f"/workspaces/{ws.id}/instances")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == []
|
||||
|
||||
def test_list_instances(
|
||||
self,
|
||||
authenticated_client: TestClient,
|
||||
db_session: AsyncSession,
|
||||
test_workspace_with_tool_type,
|
||||
):
|
||||
ws, tool_type, project = test_workspace_with_tool_type
|
||||
|
||||
async def _create_instance():
|
||||
instance = ToolInstance(
|
||||
name="test-instance",
|
||||
display_name="Test Instance",
|
||||
tool_type_id=tool_type.id,
|
||||
repository_id=ws.repo_id,
|
||||
project_id=project.id,
|
||||
owner_id=ws.user_id,
|
||||
workspace_id=ws.id,
|
||||
status="running",
|
||||
)
|
||||
db_session.add(instance)
|
||||
await db_session.commit()
|
||||
|
||||
asyncio.run(_create_instance())
|
||||
|
||||
response = authenticated_client.get(f"/workspaces/{ws.id}/instances")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert len(data) == 1
|
||||
assert data[0]["display_name"] == "Test Instance"
|
||||
|
||||
|
||||
class TestCreateWorkspaceInstance:
|
||||
"""Tests for POST /workspaces/{id}/instances."""
|
||||
|
||||
def test_create_instance(
|
||||
self,
|
||||
authenticated_client: TestClient,
|
||||
test_workspace_with_tool_type,
|
||||
):
|
||||
ws, tool_type, project = test_workspace_with_tool_type
|
||||
mock_instance = ToolInstance(
|
||||
id=uuid.uuid4(),
|
||||
name="mock-instance",
|
||||
display_name="Mock Instance",
|
||||
tool_type_id=tool_type.id,
|
||||
repository_id=ws.repo_id,
|
||||
project_id=project.id,
|
||||
owner_id=ws.user_id,
|
||||
workspace_id=ws.id,
|
||||
status="pending",
|
||||
created_at=datetime.now(),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"src.api.workspace.workspace_instances.create_tool_instance",
|
||||
return_value=mock_instance,
|
||||
):
|
||||
response = authenticated_client.post(
|
||||
f"/workspaces/{ws.id}/instances",
|
||||
json={"tool_type_id": str(tool_type.id), "display_name": "New Instance"},
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["display_name"] == "Mock Instance"
|
||||
assert data["status"] == "pending"
|
||||
@@ -8,12 +8,14 @@ import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.models.git_repository import GitRepository
|
||||
from src.models.project import Project
|
||||
from src.models.tool_instance import ToolInstance
|
||||
from src.models.tool_type import ToolType
|
||||
from src.models.workspace import Workspace
|
||||
from src.services.workspace_manager import WorkspaceManager
|
||||
from src.models import (
|
||||
GitRepository,
|
||||
Project,
|
||||
ToolInstance,
|
||||
ToolType,
|
||||
Workspace,
|
||||
)
|
||||
from src.services.shared.workspace_manager import WorkspaceManager
|
||||
|
||||
|
||||
def _get_user_id_from_client(client: TestClient) -> uuid.UUID:
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
dir: apps/api/tests/unit
|
||||
|
||||
## role
|
||||
Unit test suite for the API backend, covering configuration, Docker services, git operations, file handling, health monitoring, notifications, permissions, and lifecycle management.
|
||||
Unit test suite for the API backend, covering configuration, Docker operations, Git services, file handling, health monitoring, notifications, and SSH key management.
|
||||
## parent
|
||||
index: apps/api/tests/.pi-map.index.md
|
||||
map: apps/api/tests/.pi-map.md
|
||||
@@ -18,6 +18,7 @@ map: apps/api/tests/.pi-map.md
|
||||
- test_file_service.py
|
||||
- test_git_mount_resolution.py
|
||||
- test_git_mounts.py
|
||||
- test_git_operations.py
|
||||
- test_git_repository_clone_preflight.py
|
||||
- test_git_repository_working_clones.py
|
||||
- test_git_service.py
|
||||
@@ -32,9 +33,7 @@ map: apps/api/tests/.pi-map.md
|
||||
- test_notifications_api_routes.py
|
||||
- test_permission_fixer.py
|
||||
- test_readiness_probe.py
|
||||
- test_session_branch_selection.py
|
||||
- test_ssh_keys.py
|
||||
- test_tool_instances_legacy.py
|
||||
## links
|
||||
index: apps/api/tests/unit/.pi-map.index.md
|
||||
map: apps/api/tests/unit/.pi-map.md
|
||||
|
||||
@@ -4,7 +4,7 @@ dir: apps/api/tests/unit
|
||||
index: apps/api/tests/unit/.pi-map.index.md
|
||||
|
||||
## role
|
||||
Unit test suite for the API backend, covering configuration, Docker services, git operations, file handling, health monitoring, notifications, permissions, and lifecycle management.
|
||||
Unit test suite for the API backend, covering configuration, Docker operations, Git services, file handling, health monitoring, notifications, and SSH key management.
|
||||
## files
|
||||
- __init__.py | Empty file with no functionality
|
||||
- test_config.py | Unit tests for application configuration settings, database URL construction, and environment-specific behavior | exp: func:test_settings_default_database_url_uses_asyncpg() → None, call:Settings, func:test_build_database_url_uses_explicit_values() → None, call:build_database_url, func:test_settings_prefers_explicit_database_url_env(monkeypatch) → None, call:monkeypatch.setenv, call:Settings, func:test_auth_settings_have_secure_defaults() → None, call:Settings, call:settings.resolved_authentik_authorize_url.endswith, call:settings.resolved_authentik_token_url.endswith, call:settings.resolved_authentik_jwks_url.endswith, func:test_cookie_policy_is_strict_in_production(monkeypatch) → None, call:monkeypatch.setenv, call:Settings, func:test_cookie_policy_is_relaxed_for_local_dev(monkeypatch) → None, call:monkeypatch.setenv, call:Settings | dep: pytest, src.config, src.database
|
||||
@@ -12,9 +12,10 @@ Unit test suite for the API backend, covering configuration, Docker services, gi
|
||||
- test_docker_build.py | Unit tests for a Docker image build service that validates successful builds, failures, command structure, file writing, context handling, path traversal prevention, timeouts, and exceptions. | exp: class:TestBuildImage | dep: subprocess, tempfile, pathlib, unittest.mock, pytest, src.services.docker_build
|
||||
- test_docker_service.py | Unit tests for Docker service utility functions including container ID/name retrieval and volume sorting by mount specificity. | exp: class:TestGetContainerId, class:TestGetContainerName, class:TestSortVolumesBySpecificity, method:test_parent_before_child(self) → None, call:sort_volumes_by_specificity, method:test_stable_sort_for_equal_depth(self) → None, call:sort_volumes_by_specificity, method:test_with_type_suffix(self) → None, call:sort_volumes_by_specificity, method:test_empty_list(self) → None, call:sort_volumes_by_specificity, method:test_single_volume(self) → None, call:sort_volumes_by_specificity, method:test_duplicate_target_warning(self, caplog) → None, call:caplog.at_level, call:sort_volumes_by_specificity | dep: unittest.mock, logging, src.services.docker, subprocess
|
||||
- test_event_bus.py | Unit tests for InstanceEventBus verifying publish/subscribe behavior, exception isolation, unsubscribe functionality, and async callback support | exp: func:event_bus() → InstanceEventBus, call:InstanceEventBus, call:bus._reset_for_testing, func:sample_payload() → InstanceEventPayload, call:str, call:uuid.uuid4, func:test_publish_delivers_to_all_subscribers(event_bus: InstanceEventBus, sample_payload: InstanceEventPayload) → None, call:received.append, call:event_bus.subscribe, call:event_bus.publish, call:len, func:callback_1(payload: InstanceEventPayload) → None, call:received.append, func:callback_2(payload: InstanceEventPayload) → None, call:received.append, func:callback_3(payload: InstanceEventPayload) → None, call:received.append, func:test_subscriber_exception_isolation(event_bus: InstanceEventBus, sample_payload: InstanceEventPayload) → None, call:received.append, call:event_bus.subscribe, call:event_bus.publish, raise:RuntimeError, func:bad_callback(_payload: InstanceEventPayload) → None, raise:RuntimeError, func:good_callback(_payload: InstanceEventPayload) → None, call:received.append, func:test_unsubscribe_removes_callback(event_bus: InstanceEventBus, sample_payload: InstanceEventPayload) → None, call:received.append, call:event_bus.subscribe, call:unsubscribe, call:event_bus.publish, func:callback(_payload: InstanceEventPayload) → None, call:received.append, func:test_publish_to_empty_subscriber_list(event_bus: InstanceEventBus, sample_payload: InstanceEventPayload) → None, call:event_bus.publish, func:test_async_subscriber_supported(event_bus: InstanceEventBus, sample_payload: InstanceEventPayload) → None, call:asyncio.sleep, call:received.append, call:event_bus.subscribe, call:event_bus.publish, func:async_callback(_payload: InstanceEventPayload) → None, call:asyncio.sleep, call:received.append, func:test_unsubscribe_all_clears_subscribers(event_bus: InstanceEventBus, sample_payload: InstanceEventPayload) → None, call:received.append, call:event_bus.subscribe, call:event_bus.unsubscribe_all, call:event_bus.publish, func:callback(_payload: InstanceEventPayload) → None, call:received.append | dep: asyncio, uuid, typing, pytest, src.services.event_bus
|
||||
- test_file_service.py | Unit tests for FileService validating directory listing, file read/write, binary rejection, and path traversal prevention. | exp: class:TestFileService, method:test_list_directory_empty(self, temp_workspace: Workspace), call:FileService, call:service.list_directory, method:test_list_directory_with_files(self, temp_workspace: Workspace), call:os.makedirs, call:os.path.join, call:open, call:f.write, call:FileService, call:service.list_directory, call:len, method:test_read_file(self, temp_workspace: Workspace), call:open, call:os.path.join, call:f.write, call:FileService, call:service.read_file, method:test_read_binary_file_rejected(self, temp_workspace: Workspace), call:open, call:os.path.join, call:f.write, call:FileService, call:pytest.raises, call:service.read_file, method:test_write_file(self, temp_workspace: Workspace), call:FileService, call:service.write_file, call:os.path.exists, call:os.path.join, call:open, call:f.read, method:test_path_escapes_workspace(self, temp_workspace: Workspace), call:FileService, call:pytest.raises, call:service.list_directory, func:temp_workspace(), call:tempfile.TemporaryDirectory, call:Workspace | dep: os, tempfile, pytest, src.models.workspace, src.services.file_service
|
||||
- test_file_service.py | Unit tests for FileService validating directory listing, file read/write, binary rejection, and path traversal prevention. | exp: class:TestFileService, method:test_list_directory_empty(self, temp_workspace: Workspace), call:FileService, call:service.list_directory, method:test_list_directory_with_files(self, temp_workspace: Workspace), call:os.makedirs, call:os.path.join, call:open, call:f.write, call:FileService, call:service.list_directory, call:len, method:test_read_file(self, temp_workspace: Workspace), call:open, call:os.path.join, call:f.write, call:FileService, call:service.read_file, method:test_read_binary_file_rejected(self, temp_workspace: Workspace), call:open, call:os.path.join, call:f.write, call:FileService, call:pytest.raises, call:service.read_file, method:test_write_file(self, temp_workspace: Workspace), call:FileService, call:service.write_file, call:os.path.exists, call:os.path.join, call:open, call:f.read, method:test_path_escapes_workspace(self, temp_workspace: Workspace), call:FileService, call:pytest.raises, call:service.list_directory, func:temp_workspace(), call:tempfile.TemporaryDirectory, call:Workspace | dep: os, tempfile, pytest, src.models, src.services.shared.file_service, src.models.Workspace, src.services.shared.file_service.FileService
|
||||
- test_git_mount_resolution.py | Unit tests for git mount resolution functions including glob expansion, branch checkout, and mount validation in tool instances. | exp: class:TestExpandGlobSource, method:test_no_glob_single_file(self, tmp_path: Path) → None, call:test_file.write_text, call:_expand_glob_source, call:str, call:len, method:test_no_glob_missing_file(self, tmp_path: Path) → None, call:_expand_glob_source, call:str, call:len, method:test_glob_pattern(self, tmp_path: Path) → None, call:(tmp_path / "file1.txt").write_text, call:(tmp_path / "file2.txt").write_text, call:(tmp_path / "other.py").write_text, call:_expand_glob_source, call:str, call:len, call:all, call:f.endswith, method:test_glob_recursive(self, tmp_path: Path) → None, call:subdir.mkdir, call:(subdir / "nested.txt").write_text, call:_expand_glob_source, call:str, call:len, method:test_glob_limit_enforced(self, tmp_path: Path) → None, call:range, call:(tmp_path / f"file{i}.txt").write_text, call:_expand_glob_source, call:str, call:len, method:test_glob_escapes_repo(self, tmp_path: Path) → None, call:other_dir.mkdir, call:(other_dir / "outside.txt").write_text, call:_expand_glob_source, call:str, call:all, call:r.startswith, class:TestCheckoutBranch, method:test_checkout_existing_branch(self, tmp_path: Path) → None, call:os.system, call:(tmp_path / "file.txt").write_text, call:_checkout_branch, call:str, call:os.popen(f"cd {tmp_path} && git branch --show-current").read().strip, method:test_checkout_nonexistent_branch(self, tmp_path: Path) → None, call:os.system, call:(tmp_path / "file.txt").write_text, call:_checkout_branch, call:str, class:TestResolveSingleGitMount | dep: os, pathlib, pytest, src.api.tool_instances, pathlib.Path
|
||||
- test_git_mounts.py | Unit tests for git mount resolution functions that handle multi-mapping support, normalization, glob expansion, and Docker volume bind mounting. | exp: class:TestNormalizeGitMount, method:test_legacy_to_mappings(self) → None, call:_normalize_git_mount, method:test_already_mappings(self) → None, call:_normalize_git_mount, method:test_missing_target_path_no_mappings(self) → None, call:_normalize_git_mount, class:TestResolveGitMountMappings, method:test_single_mapping(self) → None, call:tempfile.TemporaryDirectory, call:os.makedirs, call:os.path.join, call:_resolve_git_mount_mappings, call:len, method:test_multiple_mappings(self) → None, call:tempfile.TemporaryDirectory, call:os.makedirs, call:os.path.join, call:_resolve_git_mount_mappings, call:len, method:test_relative_target_path(self) → None, call:tempfile.TemporaryDirectory, call:os.makedirs, call:os.path.join, call:_resolve_git_mount_mappings, call:len, method:test_glob_expansion(self) → None, call:tempfile.TemporaryDirectory, call:os.makedirs, call:os.path.join, call:_resolve_git_mount_mappings, call:len, method:test_missing_target_path_skipped(self) → None, call:tempfile.TemporaryDirectory, call:_resolve_git_mount_mappings, call:len, method:test_no_working_directory_for_relative_target(self) → None, call:tempfile.TemporaryDirectory, call:os.makedirs, call:os.path.join, call:_resolve_git_mount_mappings, call:len, class:TestResolveSingleGitMount, class:TestExpandGlobSource, method:test_no_glob(self) → None, call:tempfile.TemporaryDirectory, call:os.path.join, call:open(path, "w").close, call:_expand_glob_source, method:test_no_glob_missing(self) → None, call:tempfile.TemporaryDirectory, call:os.path.join, call:_expand_glob_source, method:test_glob_pattern(self) → None, call:tempfile.TemporaryDirectory, call:open(os.path.join(tmp, "a.txt"), "w").close, call:os.path.join, call:open(os.path.join(tmp, "b.txt"), "w").close, call:_expand_glob_source, call:len | dep: os, tempfile, unittest.mock, pytest, src.api.tool_instances
|
||||
- test_git_operations.py | Unit tests for GitOperations class covering git status, commit, history, and branch operations | exp: class:TestGitOperationsStatus, method:test_status_clean(self, temp_workspace: Workspace), call:GitOperations, call:asyncio.run, call:git.status, method:test_status_modified(self, temp_workspace: Workspace), call:open, call:os.path.join, call:f.write, call:GitOperations, call:asyncio.run, call:git.status, method:test_status_untracked(self, temp_workspace: Workspace), call:open, call:os.path.join, call:f.write, call:GitOperations, call:asyncio.run, call:git.status, class:TestGitOperationsCommit, method:test_commit_stages_and_commits(self, temp_workspace: Workspace), call:open, call:os.path.join, call:f.write, call:GitOperations, call:asyncio.run, call:git.commit, call:git.status, call:git.history, method:test_commit_fails_without_changes(self, temp_workspace: Workspace), call:GitOperations, call:pytest.raises, call:asyncio.run, call:git.commit, class:TestGitOperationsHistory, method:test_history_returns_commits(self, temp_workspace: Workspace), call:GitOperations, call:asyncio.run, call:git.history, call:len, method:test_history_filters_by_path(self, temp_workspace: Workspace), call:open, call:os.path.join, call:f.write, call:GitOperations, call:asyncio.run, call:git.commit, call:git.history, call:len, class:TestGitOperationsBranches, method:test_branches_lists_main(self, temp_workspace: Workspace), call:GitOperations, call:asyncio.run, call:git.branches, method:test_checkout_switches_branch(self, temp_workspace: Workspace), call:_run_git, call:GitOperations, call:asyncio.run, call:git.checkout, call:git.status, func:_run_git(*args: str, cwd: str) → None, call:subprocess.run, func:temp_workspace(), call:tempfile.TemporaryDirectory, call:_run_git, call:os.path.join, call:open, call:f.write, call:Workspace | dep: asyncio, os, subprocess, tempfile, pytest, src.models, src.services.git.git_operations, src.models.Workspace, src.services.git.git_operations.GitOperations
|
||||
- test_git_repository_clone_preflight.py | Unit tests for git repository clone URL building and preflight accessibility checks | exp: func:test_build_provider_clone_url_uses_fixed_host() → None, call:_build_provider_clone_url, func:test_preflight_remote_repository_allows_accessible_repo() → None, call:Mock, call:patch, call:_preflight_remote_repository, call:run_mock.assert_called_once, func:test_preflight_remote_repository_rejects_missing_repo() → None, call:Mock, call:patch, call:pytest.raises, call:_preflight_remote_repository | dep: unittest.mock, pytest, fastapi, src.api.git_repositories, subprocess
|
||||
- test_git_repository_working_clones.py | Tests git repository cloning and initialization utilities with fallback strategies for different git versions | exp: func:test_clone_working_repository_uses_normal_clone() → None, call:Mock, call:patch, call:_clone_working_repository, call:run_mock.assert_called_once, func:test_clone_working_repository_raises_on_failure() → None, call:Mock, call:patch, call:pytest.raises, call:_clone_working_repository, func:test_init_working_repository_prefers_init_b() → None, call:Mock, call:patch, call:_init_working_repository, func:test_init_working_repository_falls_back_to_symbolic_ref() → None, call:Mock, call:patch, call:_init_working_repository, func:test_create_branch_uses_orphan_checkout_when_head_is_unborn() → None, call:patch, call:create_branch, raise:RuntimeError, func:mock_run(repo_path: str, *args: str) → str, raise:RuntimeError | dep: unittest.mock, pytest, fastapi, src.api.git_repositories, src.utils.git_control, subprocess
|
||||
- test_git_service.py | Unit tests for GitService class covering clone, fetch, pull, and branch_exists_remotely operations with mocked subprocess calls. | exp: class:TestGitServiceClone, class:TestGitServiceFetch, class:TestGitServicePull, class:TestGitServiceBranchExistsRemotely, method:test_branch_exists(self), call:MagicMock, call:patch, call:GitService.branch_exists_remotely, call:mock_run.assert_called_once_with, method:test_branch_not_exists(self), call:MagicMock, call:patch, call:GitService.branch_exists_remotely, method:test_ls_remote_fails(self), call:MagicMock, call:patch, call:GitService.branch_exists_remotely | dep: asyncio, unittest.mock, pytest, src.services.git_service
|
||||
@@ -29,13 +30,11 @@ Unit test suite for the API backend, covering configuration, Docker services, gi
|
||||
- test_notifications_api_routes.py | Tests that FastAPI notification routes are declared in correct order so bulk DELETE /notifications matches before parameterized DELETE /notifications/{id} | exp: func:test_delete_notifications_route_order() → None, call:FastAPI, call:app.include_router, call:TestClient, call:client.delete | dep: fastapi, fastapi.testclient, src.api.notifications
|
||||
- test_permission_fixer.py | Unit tests for a Docker container permission fixer service that applies chown/chmod/file_mode to mounts and SSH directories. | exp: class:TestApplyMountPermissions, class:TestRunInContainer, class:TestApplySshPermissions, class:TestCheckRootUserAvailable | dep: unittest.mock, pytest, src.services.permission_fixer, subprocess
|
||||
- test_readiness_probe.py | Unit tests for a Docker container readiness probe service that executes commands via docker exec with retry logic and timeout handling. | exp: class:TestExecuteProbe, class:TestIntegrationScenarios | dep: unittest.mock, src.services.readiness_probe, subprocess
|
||||
- test_session_branch_selection.py | Tests session creation with branch selection and new branch creation via git operations and API request validation. | exp: class:TestCreateInstanceRequest, method:test_default_values(self), call:CreateInstanceRequest, method:test_clone_mode_with_branch(self), call:CreateInstanceRequest, method:test_new_branch_field(self), call:CreateInstanceRequest, class:TestBranchCreationInClone, method:test_create_local_branch_success(self), call:tempfile.TemporaryDirectory, call:subprocess.run, call:os.path.join, call:open, call:f.write, call:branches_result.stdout.strip, method:test_create_local_branch_invalid_name(self), call:tempfile.TemporaryDirectory, call:subprocess.run, call:os.path.join, call:open, call:f.write, class:TestCreateInstanceAPI, method:test_create_instance_request_validation(self), call:CreateInstanceRequest, method:test_create_instance_with_new_branch_sets_instance_branch(self) | dep: os, subprocess, tempfile, src.api.tool_instances, src.api.tool_instances.CreateInstanceRequest
|
||||
- test_ssh_keys.py | Unit tests for SSH key preparation functionality that decrypts encrypted private keys and writes SSH key files with proper permissions and optional ownership. | exp: class:TestPrepareSshKeyFiles | dep: os, pathlib, unittest.mock, pytest, src.services.ssh_keys
|
||||
- test_tool_instances_legacy.py | Unit tests verifying legacy tool instance fallback paths (dockerfile, compose, legacy definition types) continue using original startup flow after manifest-based flow introduction | exp: class:TestCreateInstanceDockerfileLegacy, class:TestCreateInstanceComposeLegacy, class:TestCreateInstanceManifestNotCalledForLegacy, class:TestStartInstanceLegacyFallback, class:TestStartInstanceSshPermissions, class:TestStartInstanceManifestBranch, func:fake_user_id() → uuid.UUID, call:uuid.uuid4, func:fake_project_id() → uuid.UUID, call:uuid.uuid4, func:fake_repo_id() → uuid.UUID, call:uuid.uuid4, func:fake_tool_type_id() → uuid.UUID, call:uuid.uuid4, func:fake_instance_id() → uuid.UUID, call:uuid.uuid4, func:mock_session(fake_user_id, fake_project_id, fake_repo_id, fake_tool_type_id), call:AsyncMock, call:User, call:MagicMock, call:GitRepository, call:getattr, call:datetime.now, func:_get(model, pk), func:_add(instance), call:getattr, call:datetime.now | dep: os, uuid, datetime, unittest.mock, pytest, src.api.tool_instances, src.models.git_repository, src.models.tool_instance, src.models.tool_type, src.models.ssh_key, src.models.user, fastapi
|
||||
## arch
|
||||
Pytest-based unit testing with heavy mocking of external dependencies (Docker, subprocess, git, filesystem), organized by service/component under test, using parameterized tests and fixture-based setup for isolated component verification.
|
||||
pytest-based unit testing with heavy mocking of external dependencies (Docker, Git subprocesses, databases), organized by service/component under test, using parameterized tests and fixture-based setup for isolated component verification.
|
||||
## tags
|
||||
test, url, call:, git, call:notification, instance, mount, call:len
|
||||
test, call:, url, git, call:notification, mount, call:len, call:db
|
||||
## symbols
|
||||
- TestMergeFunctions
|
||||
- TestResolveProfile
|
||||
|
||||
@@ -5,8 +5,8 @@ import tempfile
|
||||
|
||||
import pytest
|
||||
|
||||
from src.models.workspace import Workspace
|
||||
from src.services.file_service import FileService
|
||||
from src.models import Workspace
|
||||
from src.services.shared.file_service import FileService
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
"""Unit tests for GitOperations."""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
|
||||
from src.models import Workspace
|
||||
from src.services.git.git_operations import GitOperations
|
||||
|
||||
|
||||
def _run_git(*args: str, cwd: str) -> None:
|
||||
subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_workspace():
|
||||
"""Create a temporary git workspace."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
_run_git("init", cwd=tmpdir)
|
||||
_run_git("config", "user.email", "test@test.com", cwd=tmpdir)
|
||||
_run_git("config", "user.name", "Test User", cwd=tmpdir)
|
||||
|
||||
readme = os.path.join(tmpdir, "README.md")
|
||||
with open(readme, "w") as f:
|
||||
f.write("# Test\n")
|
||||
_run_git("add", "README.md", cwd=tmpdir)
|
||||
_run_git("commit", "-m", "Initial commit", cwd=tmpdir)
|
||||
|
||||
ws = Workspace(
|
||||
id="00000000-0000-0000-0000-000000000001",
|
||||
name="test-ws",
|
||||
repo_id="00000000-0000-0000-0000-000000000002",
|
||||
user_id="00000000-0000-0000-0000-000000000003",
|
||||
branch="main",
|
||||
path=tmpdir,
|
||||
)
|
||||
yield ws
|
||||
|
||||
|
||||
class TestGitOperationsStatus:
|
||||
"""Tests for GitOperations.status."""
|
||||
|
||||
def test_status_clean(self, temp_workspace: Workspace):
|
||||
git = GitOperations(temp_workspace)
|
||||
status = asyncio.run(git.status())
|
||||
assert status.branch == "main"
|
||||
assert status.modified == []
|
||||
assert status.added == []
|
||||
assert status.deleted == []
|
||||
assert status.untracked == []
|
||||
|
||||
def test_status_modified(self, temp_workspace: Workspace):
|
||||
with open(os.path.join(temp_workspace.path, "README.md"), "a") as f:
|
||||
f.write("modified\n")
|
||||
|
||||
git = GitOperations(temp_workspace)
|
||||
status = asyncio.run(git.status())
|
||||
assert "README.md" in status.modified
|
||||
|
||||
def test_status_untracked(self, temp_workspace: Workspace):
|
||||
with open(os.path.join(temp_workspace.path, "new.py"), "w") as f:
|
||||
f.write("print('hello')\n")
|
||||
|
||||
git = GitOperations(temp_workspace)
|
||||
status = asyncio.run(git.status())
|
||||
assert "new.py" in status.untracked
|
||||
|
||||
|
||||
class TestGitOperationsCommit:
|
||||
"""Tests for GitOperations.commit."""
|
||||
|
||||
def test_commit_stages_and_commits(self, temp_workspace: Workspace):
|
||||
with open(os.path.join(temp_workspace.path, "new.py"), "w") as f:
|
||||
f.write("print('hello')\n")
|
||||
|
||||
git = GitOperations(temp_workspace)
|
||||
asyncio.run(git.commit("Add new file"))
|
||||
|
||||
status = asyncio.run(git.status())
|
||||
assert "new.py" not in status.untracked
|
||||
assert "new.py" not in status.added
|
||||
|
||||
history = asyncio.run(git.history())
|
||||
assert history[0].message == "Add new file"
|
||||
|
||||
def test_commit_fails_without_changes(self, temp_workspace: Workspace):
|
||||
git = GitOperations(temp_workspace)
|
||||
with pytest.raises(RuntimeError):
|
||||
asyncio.run(git.commit("Nothing to commit"))
|
||||
|
||||
|
||||
class TestGitOperationsHistory:
|
||||
"""Tests for GitOperations.history."""
|
||||
|
||||
def test_history_returns_commits(self, temp_workspace: Workspace):
|
||||
git = GitOperations(temp_workspace)
|
||||
history = asyncio.run(git.history())
|
||||
assert len(history) >= 1
|
||||
assert history[0].message == "Initial commit"
|
||||
|
||||
def test_history_filters_by_path(self, temp_workspace: Workspace):
|
||||
with open(os.path.join(temp_workspace.path, "new.py"), "w") as f:
|
||||
f.write("print('hello')\n")
|
||||
|
||||
git = GitOperations(temp_workspace)
|
||||
asyncio.run(git.commit("Add new file"))
|
||||
|
||||
history = asyncio.run(git.history(path="new.py"))
|
||||
assert len(history) == 1
|
||||
assert history[0].message == "Add new file"
|
||||
|
||||
|
||||
class TestGitOperationsBranches:
|
||||
"""Tests for GitOperations.branches."""
|
||||
|
||||
def test_branches_lists_main(self, temp_workspace: Workspace):
|
||||
git = GitOperations(temp_workspace)
|
||||
branches, current = asyncio.run(git.branches())
|
||||
assert "main" in branches
|
||||
assert current == "main"
|
||||
|
||||
def test_checkout_switches_branch(self, temp_workspace: Workspace):
|
||||
_run_git("checkout", "-b", "feature", cwd=temp_workspace.path)
|
||||
_run_git("checkout", "main", cwd=temp_workspace.path)
|
||||
|
||||
git = GitOperations(temp_workspace)
|
||||
asyncio.run(git.checkout("feature"))
|
||||
|
||||
status = asyncio.run(git.status())
|
||||
assert status.branch == "feature"
|
||||
Reference in New Issue
Block a user