fix: run tool terminal sessions as container user instead of root
- Remove compose-level user: 0:0 override from manifest_compiler.py so the entrypoint can start as root, fix mount ownership, and drop privileges to the container user internally. - Add get_manifest_container_user() helper to resolve the manifest-declared container user (with uid:gid fallback). - Pass container user through TerminalSession, TerminalManager, and the terminal WebSocket handler so docker exec is invoked with --user <user>. - Update and add unit tests for the manifest compiler and terminal session. - Record the additional root-user fix in the fix-pi-container-mount-permissions OpenSpec change/tasks. Quality gates: pytest tests/unit/ (226 passed), pytest tests/services/test_terminal_manager_multi.py (7 passed), ruff check on changed files (clean), mypy on changed files (clean)
This commit is contained in:
Developer
@@ -4,35 +4,36 @@ dir: apps/api/tests/unit
|
||||
index: apps/api/tests/unit/.pi-map.index.md
|
||||
|
||||
## role
|
||||
Contains unit tests for the API application's core services, utilities, and infrastructure components.
|
||||
Contains unit tests for core API services and utilities, covering database migrations, configuration, Docker operations, Git integration, event handling, file management, health monitoring, manifest compilation, notifications, permissions, SSH keys, and terminal sessions.
|
||||
## files
|
||||
- __init__.py | Swaps two tmux panes between windows, preserving active pane state and layout | dep: tmux
|
||||
- test_alembic_migrations.py | Unit tests that verify Alembic database migration files are importable, have correct revision identifiers, and declare expected dependencies. | exp: func:test_home_directory_migration_imports_and_rewrites() → None, call:Path, call:migration_path.exists, call:importlib.util.spec_from_file_location, call:importlib.util.module_from_spec, call:spec.loader.exec_module, call:callable, func:test_merge_migration_resolves_heads() → None, call:Path, call:migration_path.exists, call:importlib.util.spec_from_file_location, call:importlib.util.module_from_spec, call:spec.loader.exec_module, call:callable, func:test_remove_pi_agent_repo_mount_migration_imports() → None, call:Path, call:migration_path.exists, call:importlib.util.spec_from_file_location, call:importlib.util.module_from_spec, call:spec.loader.exec_module, call:callable | dep: importlib.util, pathlib, pytest
|
||||
- test_config.py | Unit tests for application configuration settings including database URLs, authentication defaults, and cookie security policies | exp: func:test_settings_default_database_url_uses_asyncpg(monkeypatch) → None, call:monkeypatch.delenv, 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, src.config.Settings, src.database.build_database_url
|
||||
- test_config_profile_resolver.py | Unit tests for config profile resolution including merge helpers, profile inheritance with cycle detection, and git mount normalization | exp: class:TestMergeFunctions, method:test_merge_env_vars_basic(self) → None, call:_merge_env_vars, method:test_merge_env_vars_tracks_overrides(self) → None, call:_merge_env_vars, method:test_merge_runtime_hints_basic(self) → None, call:_merge_runtime_hints, method:test_merge_files_basic(self) → None, call:_merge_files, method:test_merge_mounts_basic(self) → None, call:_merge_mounts, method:test_merge_mounts_file_override(self) → None, call:_merge_mounts, call:ResolvedMount, method:test_merge_mounts_mode_conflict(self) → None, call:_merge_mounts, call:ResolvedMount, method:test_merge_git_mounts_basic(self) → None, call:_merge_git_mounts, call:len, method:test_merge_git_mounts_concatenate_same_repo_branch(self) → None, call:_merge_git_mounts, call:len, method:test_merge_git_mounts_dedup_same_mapping(self) → None, call:_merge_git_mounts, call:len, method:test_merge_git_mounts_different_repos(self) → None, call:_merge_git_mounts, call:len, method:test_merge_git_mounts_different_branches(self) → None, call:_merge_git_mounts, call:len, call:m.get, class:TestResolveProfile, class:TestApplyResolvedProfile, method:test_mounts_directory_not_individual_files(self, tmp_path) → None, call:ResolvedProfile, call:uuid.uuid4, call:ResolvedMount, call:apply_resolved_profile, call:str, call:len, call:Path(volumes[0]["source"]).is_dir, call:(Path(volumes[0]["source"]) / "config.json").exists, call:(Path(volumes[0]["source"]) / "nested" / "file.txt").exists, method:test_directory_mount_target(self, tmp_path) → None, call:ResolvedProfile, call:uuid.uuid4, call:ResolvedMount, call:apply_resolved_profile, call:str, call:len, call:Path, call:(Path(volumes[0]["source"]) / "z.json").exists, method:test_empty_mount_produces_no_volumes(self, tmp_path) → None, call:ResolvedProfile, call:uuid.uuid4, call:ResolvedMount, call:apply_resolved_profile, call:str, method:test_home_expansion_in_directory_mount_target(self, tmp_path) → None, call:ResolvedProfile, call:uuid.uuid4, call:ResolvedMount, call:apply_resolved_profile, call:str, call:len, call:(Path(volumes[0]["source"]) / "app.toml").exists, call:Path, method:test_readonly_mount_sets_readonly_flag(self, tmp_path) → None, call:ResolvedProfile, call:uuid.uuid4, call:ResolvedMount, call:apply_resolved_profile, call:str, call:len, call:volumes[0].get, method:test_writable_mount_does_not_set_readonly_flag(self, tmp_path) → None, call:ResolvedProfile, call:uuid.uuid4, call:ResolvedMount, call:apply_resolved_profile, call:str, call:len, call:volumes[0].get, class:TestCheckIncludeCycle | dep: uuid, pathlib, pytest, sqlalchemy.ext.asyncio, src.models.config.config_profile, src.services.config.config_profile_resolver
|
||||
- test_docker_build.py | Unit tests for a Docker image build service that verifies command construction, file writing, error handling, and security constraints. | exp: class:TestBuildImage | dep: subprocess, tempfile, pathlib, unittest.mock, pytest, src.services.build.docker_build
|
||||
- __init__.py | Provides a command to swap the position of two tmux panes within a window or between windows | dep: tmux, client, window, layout, cmd-find, cmd-parse, options
|
||||
- test_alembic_migrations.py | Unit tests that verify Alembic database migrations are importable, have correct revision identifiers, and declare expected dependencies without requiring a live database. | exp: func:test_home_directory_migration_imports_and_rewrites() → None, call:Path, call:migration_path.exists, call:importlib.util.spec_from_file_location, call:importlib.util.module_from_spec, call:spec.loader.exec_module, call:callable, func:test_merge_migration_resolves_heads() → None, call:Path, call:migration_path.exists, call:importlib.util.spec_from_file_location, call:importlib.util.module_from_spec, call:spec.loader.exec_module, call:callable, func:test_remove_pi_agent_repo_mount_migration_imports() → None, call:Path, call:migration_path.exists, call:importlib.util.spec_from_file_location, call:importlib.util.module_from_spec, call:spec.loader.exec_module, call:callable | dep: importlib.util, pathlib, pytest, importlib
|
||||
- test_config.py | Unit tests for application configuration settings including database URLs, auth defaults, and cookie policies across environments. | exp: func:test_settings_default_database_url_uses_asyncpg(monkeypatch) → None, call:monkeypatch.delenv, 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
|
||||
- test_config_profile_resolver.py | Unit tests for config profile resolution, merging, and application, including inheritance, cycle detection, and mount normalization. | exp: class:TestMergeFunctions, method:test_merge_env_vars_basic(self) → None, call:_merge_env_vars, method:test_merge_env_vars_tracks_overrides(self) → None, call:_merge_env_vars, method:test_merge_runtime_hints_basic(self) → None, call:_merge_runtime_hints, method:test_merge_files_basic(self) → None, call:_merge_files, method:test_merge_mounts_basic(self) → None, call:_merge_mounts, method:test_merge_mounts_file_override(self) → None, call:_merge_mounts, call:ResolvedMount, method:test_merge_mounts_mode_conflict(self) → None, call:_merge_mounts, call:ResolvedMount, method:test_merge_git_mounts_basic(self) → None, call:_merge_git_mounts, call:len, method:test_merge_git_mounts_concatenate_same_repo_branch(self) → None, call:_merge_git_mounts, call:len, method:test_merge_git_mounts_dedup_same_mapping(self) → None, call:_merge_git_mounts, call:len, method:test_merge_git_mounts_different_repos(self) → None, call:_merge_git_mounts, call:len, method:test_merge_git_mounts_different_branches(self) → None, call:_merge_git_mounts, call:len, call:m.get, class:TestResolveProfile, class:TestApplyResolvedProfile, method:test_mounts_directory_not_individual_files(self, tmp_path) → None, call:ResolvedProfile, call:uuid.uuid4, call:ResolvedMount, call:apply_resolved_profile, call:str, call:len, call:Path(volumes[0]["source"]).is_dir, call:(Path(volumes[0]["source"]) / "config.json").exists, call:(Path(volumes[0]["source"]) / "nested" / "file.txt").exists, method:test_directory_mount_target(self, tmp_path) → None, call:ResolvedProfile, call:uuid.uuid4, call:ResolvedMount, call:apply_resolved_profile, call:str, call:len, call:Path, call:(Path(volumes[0]["source"]) / "z.json").exists, method:test_empty_mount_produces_no_volumes(self, tmp_path) → None, call:ResolvedProfile, call:uuid.uuid4, call:ResolvedMount, call:apply_resolved_profile, call:str, method:test_home_expansion_in_directory_mount_target(self, tmp_path) → None, call:ResolvedProfile, call:uuid.uuid4, call:ResolvedMount, call:apply_resolved_profile, call:str, call:len, call:(Path(volumes[0]["source"]) / "app.toml").exists, call:Path, method:test_readonly_mount_sets_readonly_flag(self, tmp_path) → None, call:ResolvedProfile, call:uuid.uuid4, call:ResolvedMount, call:apply_resolved_profile, call:str, call:len, call:volumes[0].get, method:test_writable_mount_does_not_set_readonly_flag(self, tmp_path) → None, call:ResolvedProfile, call:uuid.uuid4, call:ResolvedMount, call:apply_resolved_profile, call:str, call:len, call:volumes[0].get, class:TestCheckIncludeCycle | dep: uuid, pathlib, pytest, sqlalchemy.ext.asyncio, src.models.config.config_profile, src.services.config.config_profile_resolver
|
||||
- test_docker_build.py | Unit tests for a Docker image build service that verifies subprocess invocation, Dockerfile/context file writing, command structure, error handling, and path traversal prevention. | exp: class:TestBuildImage | dep: subprocess, tempfile, pathlib, unittest.mock, pytest, src.services.build.docker_build
|
||||
- test_docker_service.py | Unit tests for Docker service utilities including container ID/name retrieval and volume sorting by 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.container, src.services.docker.compose, subprocess
|
||||
- test_event_bus.py | Unit tests for InstanceEventBus verifying publish/subscribe, exception isolation, unsubscribe, async callback, and bulk unsubscribe functionality. | 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.instance.event_bus
|
||||
- test_file_service.py | Unit tests for FileService validating directory listing, file read/write operations, binary file rejection, and path traversal security. | 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_event_bus.py | Unit tests for an instance event bus that validates publish/subscribe, error isolation, async callback support, and unsubscribe functionality. | 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.instance.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, src.services.shared.file_service, src.models.Workspace, src.services.shared.file_service.FileService
|
||||
- test_git_operations.py | Unit tests for GitOperations class covering 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_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.git_service
|
||||
- test_git_url_parser.py | Unit tests for git URL parsing utilities that validate extraction of base repository URLs, clone URL validation, and comprehensive URL parsing across GitHub, GitLab, and Bitbucket formats. | exp: class:TestExtractBaseRepoUrl, method:test_github_tree_url(self), call:extract_base_repo_url, method:test_github_blob_url(self), call:extract_base_repo_url, method:test_github_pull_url(self), call:extract_base_repo_url, method:test_github_issues_url(self), call:extract_base_repo_url, method:test_github_valid_url(self), call:extract_base_repo_url, method:test_github_url_with_query_params(self), call:extract_base_repo_url, method:test_gitlab_tree_url(self), call:extract_base_repo_url, method:test_gitlab_blob_url(self), call:extract_base_repo_url, method:test_gitlab_merge_request_url(self), call:extract_base_repo_url, method:test_gitlab_valid_url(self), call:extract_base_repo_url, method:test_bitbucket_src_url(self), call:extract_base_repo_url, method:test_bitbucket_valid_url(self), call:extract_base_repo_url, method:test_ssh_url(self), call:extract_base_repo_url, method:test_ssh_url_without_git_suffix(self), call:extract_base_repo_url, method:test_invalid_url(self), call:extract_base_repo_url, method:test_empty_url(self), call:extract_base_repo_url, class:TestIsValidCloneUrl, method:test_valid_ssh_url(self), call:is_valid_clone_url, method:test_valid_https_url(self), call:is_valid_clone_url, method:test_browser_url(self), call:is_valid_clone_url, method:test_url_without_git_suffix(self), call:is_valid_clone_url, method:test_invalid_url(self), call:is_valid_clone_url, class:TestParseGitUrl, method:test_valid_git_url(self), call:parse_git_url, method:test_browser_url(self), call:parse_git_url, method:test_invalid_url(self), call:parse_git_url, method:test_empty_url(self), call:parse_git_url, method:test_ssh_url(self), call:parse_git_url | dep: src.utils.git_url_parser, pytest
|
||||
- test_health_monitor.py | Unit tests for HealthMonitor state-transition logic covering container crash detection, tunnel failure detection, recovery detection, write optimization, and exception resilience. | exp: func:event_bus() → InstanceEventBus, call:InstanceEventBus, call:bus._reset_for_testing, func:health_monitor(event_bus: InstanceEventBus) → HealthMonitor, call:HealthMonitor, func:_create_running_instance(db_session) → ToolInstance, call:User, call:uuid.uuid4, call:db_session.add, call:db_session.commit, call:ToolInstance, func:test_detects_container_crash(db_session, event_bus: InstanceEventBus, health_monitor: HealthMonitor) → None, call:_create_running_instance, call:events_captured.append, call:event_bus.subscribe, call:patch, call:health_monitor._check_instance, call:db_session.refresh, call:len, call:db_session.execute, call:select(HealthCheck).where, call:result.scalar_one, func:capture_event(payload: InstanceEventPayload) → None, call:events_captured.append, func:test_detects_tunnel_failure(db_session, event_bus: InstanceEventBus, health_monitor: HealthMonitor) → None, call:_create_running_instance, call:events_captured.append, call:event_bus.subscribe, call:patch, call:health_monitor._check_instance, call:db_session.refresh, call:len, call:db_session.execute, call:select(HealthCheck).where, call:result.scalar_one, func:capture_event(payload: InstanceEventPayload) → None, call:events_captured.append, func:test_detects_recovery(db_session, event_bus: InstanceEventBus, health_monitor: HealthMonitor) → None, call:_create_running_instance, call:db_session.commit, call:HealthSnapshot, call:events_captured.append, call:event_bus.subscribe, call:patch, call:health_monitor._check_instance, call:db_session.refresh, call:len, call:db_session.execute, call:select(HealthCheck).where, call:result.scalar_one, func:capture_event(payload: InstanceEventPayload) → None, call:events_captured.append, func:test_skips_writes_when_no_state_change(db_session, event_bus: InstanceEventBus, health_monitor: HealthMonitor) → None, call:_create_running_instance, call:HealthSnapshot, call:patch, call:health_monitor._check_instance, call:db_session.execute, call:select(HealthCheck).where, call:len, call:result.scalars().all, func:test_docker_exception_resilience(db_session, event_bus: InstanceEventBus, health_monitor: HealthMonitor) → None, call:_create_running_instance, call:events_captured.append, call:event_bus.subscribe, call:patch, call:RuntimeError, call:health_monitor._check_instance, call:db_session.execute, call:select(HealthCheck).where, call:result.scalar_one_or_none, func:capture_event(payload: InstanceEventPayload) → None, call:events_captured.append, func:test_monitor_start_stop(health_monitor: HealthMonitor) → None, call:health_monitor.start, call:task.done, call:health_monitor.stop, call:suppress, call:task.cancelled | dep: asyncio, uuid, contextlib, unittest.mock, pytest, sqlalchemy, src.models.system.health_check, src.models.tool.tool_instance, src.models.user.user, src.services.instance.event_bus, src.services.instance.health_monitor
|
||||
- test_home_path_expansion.py | Unit tests for tilde and $HOME expansion in container paths, plus manifest-based home directory resolution | exp: class:TestExpandContainerPath, method:test_tilde_slash_expands(self) → None, call:expand_container_path, method:test_tilde_alone_expands(self) → None, call:expand_container_path, method:test_dollar_home_slash_expands(self) → None, call:expand_container_path, method:test_dollar_home_alone_expands(self) → None, call:expand_container_path, method:test_absolute_path_unchanged(self) → None, call:expand_container_path, method:test_relative_path_unchanged(self) → None, call:expand_container_path, method:test_tilde_in_middle_unchanged(self) → None, call:expand_container_path, method:test_dollar_home_in_middle_unchanged(self) → None, call:expand_container_path, method:test_root_home(self) → None, call:expand_container_path, class:TestGetManifestHomeDir, method:test_with_user_block(self) → None, call:get_manifest_home_dir, method:test_without_user_block(self) → None, call:get_manifest_home_dir, method:test_with_empty_user_name(self) → None, call:get_manifest_home_dir, method:test_with_none_user_name(self) → None, call:get_manifest_home_dir | dep: pytest, src.services.config.config_profile_resolver, src.services.build.manifest_compiler
|
||||
- test_instance_service.py | Unit tests for tool instance service functions including compose file modification, repository mount name resolution, profile/git mount stacking, and manifest instance preparation. | exp: class:TestModifyComposeFile, method:test_extra_volumes_expand_home_dir(self, tmp_path), call:compose_path.write_text, call:modify_compose_file, call:str, call:compose_path.read_text, method:test_working_directory_expands_home_dir(self, tmp_path), call:compose_path.write_text, call:modify_compose_file, call:str, call:compose_path.read_text, class:TestGetRepositoryMountName, method:test_prefers_remote_url_name_over_user_provided_name(self), call:MagicMock, call:_get_repository_mount_name, method:test_parses_browser_url_to_repo_name(self), call:MagicMock, call:_get_repository_mount_name, method:test_uses_workspace_path_basename_when_workspace_provided(self), call:MagicMock, call:_get_repository_mount_name, method:test_falls_back_to_repo_name_when_remote_url_missing(self), call:MagicMock, call:_get_repository_mount_name, method:test_falls_back_to_repo_name_for_unparseable_url(self), call:MagicMock, call:_get_repository_mount_name, class:TestStackProfileMountsWithGitMounts, method:test_exact_overlap_merges_profile_files_into_git_source(self, tmp_path) → None, call:git_source.mkdir, call:(git_source / "existing.txt").write_text, call:profile_source.mkdir, call:(profile_source / "settings.json").write_text, call:str, call:_stack_profile_mounts_with_git_mounts, call:(git_source / "existing.txt").read_text, call:(git_source / "settings.json").read_text, method:test_descendant_overlap_copies_into_subdirectory(self, tmp_path) → None, call:git_source.mkdir, call:(git_source / "README").write_text, call:profile_source.mkdir, call:(profile_source / "settings.json").write_text, call:str, call:_stack_profile_mounts_with_git_mounts, call:(git_source / "agent" / "settings.json").read_text, call:(git_source / "README").read_text, method:test_non_overlapping_mounts_left_untouched(self, tmp_path) → None, call:git_source.mkdir, call:profile_source.mkdir, call:(profile_source / "config").write_text, call:str, call:_stack_profile_mounts_with_git_mounts, method:test_git_source_file_does_not_consume_profile_mount(self, tmp_path) → None, call:git_source.write_text, call:profile_source.mkdir, call:(profile_source / "settings.json").write_text, call:str, call:_stack_profile_mounts_with_git_mounts, method:test_profile_source_file_copied_into_git_source(self, tmp_path) → None, call:git_source.mkdir, call:profile_source.write_text, call:str, call:_stack_profile_mounts_with_git_mounts, call:(git_source / "settings.json").read_text, class:Result, func:test_prepare_manifest_instance_uses_workspace_path_basename(), call:MagicMock, call:AsyncMock, call:Result, call:prepare_manifest_instance, func:session_get(model, obj_id), func:fake_run(cmd), call:Result | dep: unittest.mock, pytest, src.services.tool.instance_service, subprocess, src.services.tool, pathlib
|
||||
- test_git_service.py | Unit tests for GitService covering clone, fetch, pull, and branch existence checks via 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.git_service
|
||||
- test_git_url_parser.py | Unit tests for git URL parsing utilities that validate extraction, validation, and parsing of various GitHub, GitLab, and Bitbucket URL formats. | exp: class:TestExtractBaseRepoUrl, method:test_github_tree_url(self), call:extract_base_repo_url, method:test_github_blob_url(self), call:extract_base_repo_url, method:test_github_pull_url(self), call:extract_base_repo_url, method:test_github_issues_url(self), call:extract_base_repo_url, method:test_github_valid_url(self), call:extract_base_repo_url, method:test_github_url_with_query_params(self), call:extract_base_repo_url, method:test_gitlab_tree_url(self), call:extract_base_repo_url, method:test_gitlab_blob_url(self), call:extract_base_repo_url, method:test_gitlab_merge_request_url(self), call:extract_base_repo_url, method:test_gitlab_valid_url(self), call:extract_base_repo_url, method:test_bitbucket_src_url(self), call:extract_base_repo_url, method:test_bitbucket_valid_url(self), call:extract_base_repo_url, method:test_ssh_url(self), call:extract_base_repo_url, method:test_ssh_url_without_git_suffix(self), call:extract_base_repo_url, method:test_invalid_url(self), call:extract_base_repo_url, method:test_empty_url(self), call:extract_base_repo_url, class:TestIsValidCloneUrl, method:test_valid_ssh_url(self), call:is_valid_clone_url, method:test_valid_https_url(self), call:is_valid_clone_url, method:test_browser_url(self), call:is_valid_clone_url, method:test_url_without_git_suffix(self), call:is_valid_clone_url, method:test_invalid_url(self), call:is_valid_clone_url, class:TestParseGitUrl, method:test_valid_git_url(self), call:parse_git_url, method:test_browser_url(self), call:parse_git_url, method:test_invalid_url(self), call:parse_git_url, method:test_empty_url(self), call:parse_git_url, method:test_ssh_url(self), call:parse_git_url | dep: src.utils.git_url_parser, pytest
|
||||
- test_health_monitor.py | Unit tests for HealthMonitor state-transition logic covering container crash detection, tunnel failure detection, recovery detection, write deduplication, and exception resilience. | exp: func:event_bus() → InstanceEventBus, call:InstanceEventBus, call:bus._reset_for_testing, func:health_monitor(event_bus: InstanceEventBus) → HealthMonitor, call:HealthMonitor, func:_create_running_instance(db_session) → ToolInstance, call:User, call:uuid.uuid4, call:db_session.add, call:db_session.commit, call:ToolInstance, func:test_detects_container_crash(db_session, event_bus: InstanceEventBus, health_monitor: HealthMonitor) → None, call:_create_running_instance, call:events_captured.append, call:event_bus.subscribe, call:patch, call:health_monitor._check_instance, call:db_session.refresh, call:len, call:db_session.execute, call:select(HealthCheck).where, call:result.scalar_one, func:capture_event(payload: InstanceEventPayload) → None, call:events_captured.append, func:test_detects_tunnel_failure(db_session, event_bus: InstanceEventBus, health_monitor: HealthMonitor) → None, call:_create_running_instance, call:events_captured.append, call:event_bus.subscribe, call:patch, call:health_monitor._check_instance, call:db_session.refresh, call:len, call:db_session.execute, call:select(HealthCheck).where, call:result.scalar_one, func:capture_event(payload: InstanceEventPayload) → None, call:events_captured.append, func:test_detects_recovery(db_session, event_bus: InstanceEventBus, health_monitor: HealthMonitor) → None, call:_create_running_instance, call:db_session.commit, call:HealthSnapshot, call:events_captured.append, call:event_bus.subscribe, call:patch, call:health_monitor._check_instance, call:db_session.refresh, call:len, call:db_session.execute, call:select(HealthCheck).where, call:result.scalar_one, func:capture_event(payload: InstanceEventPayload) → None, call:events_captured.append, func:test_skips_writes_when_no_state_change(db_session, event_bus: InstanceEventBus, health_monitor: HealthMonitor) → None, call:_create_running_instance, call:HealthSnapshot, call:patch, call:health_monitor._check_instance, call:db_session.execute, call:select(HealthCheck).where, call:len, call:result.scalars().all, func:test_docker_exception_resilience(db_session, event_bus: InstanceEventBus, health_monitor: HealthMonitor) → None, call:_create_running_instance, call:events_captured.append, call:event_bus.subscribe, call:patch, call:RuntimeError, call:health_monitor._check_instance, call:db_session.execute, call:select(HealthCheck).where, call:result.scalar_one_or_none, func:capture_event(payload: InstanceEventPayload) → None, call:events_captured.append, func:test_monitor_start_stop(health_monitor: HealthMonitor) → None, call:health_monitor.start, call:task.done, call:health_monitor.stop, call:suppress, call:task.cancelled | dep: asyncio, uuid, contextlib, unittest.mock, pytest, sqlalchemy, src.models.system.health_check, src.models.tool.tool_instance, src.models.user.user, src.services.instance.event_bus, src.services.instance.health_monitor
|
||||
- test_home_path_expansion.py | Unit tests for tilde and $HOME expansion in container paths, and manifest-based home directory resolution. | exp: class:TestExpandContainerPath, method:test_tilde_slash_expands(self) → None, call:expand_container_path, method:test_tilde_alone_expands(self) → None, call:expand_container_path, method:test_dollar_home_slash_expands(self) → None, call:expand_container_path, method:test_dollar_home_alone_expands(self) → None, call:expand_container_path, method:test_absolute_path_unchanged(self) → None, call:expand_container_path, method:test_relative_path_unchanged(self) → None, call:expand_container_path, method:test_tilde_in_middle_unchanged(self) → None, call:expand_container_path, method:test_dollar_home_in_middle_unchanged(self) → None, call:expand_container_path, method:test_root_home(self) → None, call:expand_container_path, class:TestGetManifestHomeDir, method:test_with_user_block(self) → None, call:get_manifest_home_dir, method:test_without_user_block(self) → None, call:get_manifest_home_dir, method:test_with_empty_user_name(self) → None, call:get_manifest_home_dir, method:test_with_none_user_name(self) → None, call:get_manifest_home_dir | dep: pytest, src.services.config.config_profile_resolver, src.services.build.manifest_compiler
|
||||
- test_instance_service.py | Unit tests for the tool instance service covering compose file modification, repository mount naming, profile/git mount stacking, and manifest instance preparation. | exp: class:TestModifyComposeFile, method:test_extra_volumes_expand_home_dir(self, tmp_path), call:compose_path.write_text, call:modify_compose_file, call:str, call:compose_path.read_text, method:test_working_directory_expands_home_dir(self, tmp_path), call:compose_path.write_text, call:modify_compose_file, call:str, call:compose_path.read_text, class:TestGetRepositoryMountName, method:test_uses_project_name(self), call:MagicMock, call:_get_repository_mount_name, method:test_slugifies_project_name(self), call:MagicMock, call:_get_repository_mount_name, class:TestStackProfileMountsWithGitMounts, method:test_exact_overlap_merges_profile_files_into_git_source(self, tmp_path) → None, call:git_source.mkdir, call:(git_source / "existing.txt").write_text, call:profile_source.mkdir, call:(profile_source / "settings.json").write_text, call:str, call:_stack_profile_mounts_with_git_mounts, call:(git_source / "existing.txt").read_text, call:(git_source / "settings.json").read_text, method:test_descendant_overlap_copies_into_subdirectory(self, tmp_path) → None, call:git_source.mkdir, call:(git_source / "README").write_text, call:profile_source.mkdir, call:(profile_source / "settings.json").write_text, call:str, call:_stack_profile_mounts_with_git_mounts, call:(git_source / "agent" / "settings.json").read_text, call:(git_source / "README").read_text, method:test_non_overlapping_mounts_left_untouched(self, tmp_path) → None, call:git_source.mkdir, call:profile_source.mkdir, call:(profile_source / "config").write_text, call:str, call:_stack_profile_mounts_with_git_mounts, method:test_git_source_file_does_not_consume_profile_mount(self, tmp_path) → None, call:git_source.write_text, call:profile_source.mkdir, call:(profile_source / "settings.json").write_text, call:str, call:_stack_profile_mounts_with_git_mounts, method:test_profile_source_file_copied_into_git_source(self, tmp_path) → None, call:git_source.mkdir, call:profile_source.write_text, call:str, call:_stack_profile_mounts_with_git_mounts, call:(git_source / "settings.json").read_text, class:Result, func:test_prepare_manifest_instance_uses_workspace_path_basename(), call:MagicMock, call:AsyncMock, call:Result, call:prepare_manifest_instance, func:session_get(model, obj_id), func:fake_run(cmd), call:Result | dep: unittest.mock, pytest, src.services.tool.instance_service, subprocess, src.services.tool
|
||||
- test_lifecycle_hooks.py | Unit tests for lifecycle hook helper functions that derive notification titles and determine whether events should trigger notifications. | exp: class:TestDeriveTitle, method:test_known_event_types(self) → None, call:_derive_title, method:test_unknown_event_type(self) → None, call:_derive_title, class:TestShouldNotify, method:test_error_events_are_notified(self) → None, call:_should_notify, method:test_health_changed_running_is_notified(self) → None, call:_should_notify, method:test_created_started_stopped_restarted_deleted_filtered(self) → None, call:_should_notify, method:test_health_changed_non_running_filtered(self) → None, call:_should_notify | dep: pytest, src.services.instance.lifecycle_hooks
|
||||
- test_manifest_compiler.py | Unit tests for a manifest compiler that generates Dockerfiles, docker-compose files, and entrypoint scripts from configuration manifests. | exp: class:TestGetManifestHomeDir, method:test_home_directory_in_manifest_wins(self) → None, call:get_manifest_home_dir, method:test_user_name_derives_home(self) → None, call:get_manifest_home_dir, method:test_root_fallback(self) → None, call:get_manifest_home_dir, method:test_empty_home_directory_falls_back(self) → None, call:get_manifest_home_dir, class:TestCompileDockerfileHomeDirectory, method:test_env_home_and_workdir_use_home_directory(self) → None, call:compile_dockerfile, method:test_workspace_symlink_created(self) → None, call:compile_dockerfile, method:test_runtime_workspace_not_baked_into_image(self) → None, call:compile_dockerfile, method:test_runtime_working_dir_overrides_home_workdir(self) → None, call:compile_dockerfile, method:test_working_dir_expands_tilde(self) → None, call:compile_dockerfile, class:TestCompileComposeHomeDirectory, method:test_default_repo_mount_synthesized(self) → None, call:compile_compose, method:test_explicit_repo_mount_preserved(self) → None, call:compile_compose, method:test_workspace_name_substituted_in_mount_target(self) → None, call:compile_compose, method:test_working_dir_expands_home(self) → None, call:compile_compose, class:TestCompileEntrypoint, method:test_entrypoint_creates_home_and_workspace(self) → None, call:compile_entrypoint, method:test_entrypoint_removes_stale_placeholder_directory(self) → None, call:compile_entrypoint, method:test_entrypoint_uses_root_then_sudo_for_workspace_symlink(self) → None, call:compile_entrypoint, call:entrypoint.find, method:test_entrypoint_fixes_mount_owners(self) → None, call:compile_entrypoint, func:test_compile_dockerfile_creates_config_dirs_for_user() → None, call:compile_dockerfile, func:test_compile_dockerfile_no_user_does_not_create_home() → None, call:compile_dockerfile, func:test_compile_dockerfile_uses_user_npm_prefix() → None, call:compile_dockerfile, func:test_compile_dockerfile_starts_as_root_and_drops_privileges() → None, call:compile_dockerfile, call:compile_entrypoint, func:test_compile_compose_runs_as_root() → None, call:compile_compose | dep: pytest, src.services.build.manifest_compiler
|
||||
- test_migration_metadata.py | Validates Alembic database migration files by dynamically loading and verifying expected table names and revision chain metadata | exp: func:test_initial_migration_defines_all_core_tables() → None, call:Path(__file__).resolve, call:spec_from_file_location, call:module_from_spec, call:spec.loader.exec_module, func:test_refresh_tokens_migration_has_expected_revision_chain() → None, call:Path(__file__).resolve, call:spec_from_file_location, call:module_from_spec, call:spec.loader.exec_module | dep: pytest, importlib.util, pathlib
|
||||
- test_monitoring_models.py | Unit tests for monitoring models (InstanceEvent and HealthCheck) verifying creation, persistence, and querying capabilities with database migration compatibility. | exp: func:test_instance_event_creation(db_session) → None, call:User, call:uuid.uuid4, call:db_session.add, call:db_session.commit, call:ToolInstance, call:InstanceEvent, call:db_session.refresh, call:isinstance, func:test_health_check_creation(db_session) → None, call:User, call:uuid.uuid4, call:db_session.add, call:db_session.commit, call:ToolInstance, call:HealthCheck, call:db_session.refresh, call:isinstance, func:test_instance_event_query_by_instance(db_session) → None, call:User, call:uuid.uuid4, call:db_session.add, call:db_session.commit, call:ToolInstance, call:InstanceEvent, call:db_session.execute, call:select(InstanceEvent).where, call:result.scalar_one | dep: uuid, datetime, pytest, sqlalchemy, src.models.system.health_check, src.models.system.instance_event, src.models.tool.tool_instance, src.models.user.user
|
||||
- test_notification_service.py | Unit tests for NotificationService covering CRUD operations, filtering, pagination, and ownership validation for user notifications. | exp: func:notification_service() → NotificationService, call:NotificationService, func:user_a(db_session: AsyncSession) → User, call:User, call:uuid.uuid4, call:db_session.add, call:db_session.commit, func:user_b(db_session: AsyncSession) → User, call:User, call:uuid.uuid4, call:db_session.add, call:db_session.commit, func:test_create_notification(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:uuid.uuid4, func:test_list_notifications_orders_by_created_at_desc(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:datetime.now, call:timedelta, call:db_session.commit, call:db_session.refresh, call:notification_service.list_notifications, func:test_list_notifications_excludes_dismissed(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:notification_service.dismiss, call:notification_service.list_notifications, func:test_list_notifications_unread_only(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:notification_service.mark_read, call:notification_service.list_notifications, func:test_get_unread_count(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:range, call:notification_service.create_notification, call:notification_service.mark_read, call:notification_service.get_unread_count, func:test_mark_read_sets_read_at(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:notification_service.mark_read, func:test_mark_all_read_affects_all_unread(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:range, call:notification_service.create_notification, call:notification_service.mark_all_read, call:notification_service.get_unread_count, func:test_dismiss_sets_dismissed_at(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:notification_service.dismiss, call:db_session.execute, call:select(Notification).where, call:result.scalar_one, func:test_mark_read_wrong_owner_raises(db_session: AsyncSession, notification_service: NotificationService, user_a: User, user_b: User) → None, call:notification_service.create_notification, call:pytest.raises, call:notification_service.mark_read, func:test_dismiss_wrong_owner_raises(db_session: AsyncSession, notification_service: NotificationService, user_a: User, user_b: User) → None, call:notification_service.create_notification, call:pytest.raises, call:notification_service.dismiss, func:test_list_notifications_mute_categories(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:notification_service.list_notifications, func:test_get_unread_count_excludes_dismissed(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:notification_service.dismiss, call:notification_service.get_unread_count, func:test_dismiss_all_affects_all_non_dismissed(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:range, call:notification_service.create_notification, call:notification_service.dismiss_all, call:notification_service.list_notifications, func:test_dismiss_all_affects_only_caller(db_session: AsyncSession, notification_service: NotificationService, user_a: User, user_b: User) → None, call:range, call:notification_service.create_notification, call:notification_service.dismiss_all, call:notification_service.list_notifications, func:test_mark_all_read_affects_only_caller(db_session: AsyncSession, notification_service: NotificationService, user_a: User, user_b: User) → None, call:range, call:notification_service.create_notification, call:notification_service.mark_all_read, call:notification_service.get_unread_count | dep: uuid, datetime, pytest, sqlalchemy, sqlalchemy.ext.asyncio, src.models.system.notification, src.models.user.user, src.services.shared.notification_service
|
||||
- test_notifications_api_routes.py | Tests that FastAPI notification routes are ordered correctly 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.system.notifications
|
||||
- test_permission_fixer.py | Unit tests for a Docker container permission fixer service that adjusts mount and SSH permissions via container exec commands. | exp: class:TestApplyMountPermissions, class:TestRunInContainer, class:TestApplySshPermissions, class:TestCheckRootUserAvailable | dep: unittest.mock, pytest, src.services.shared.permission_fixer, subprocess
|
||||
- test_manifest_compiler.py | Unit tests for a manifest compiler that generates Dockerfiles, docker-compose files, and entrypoint scripts from manifest configurations. | exp: class:TestGetManifestHomeDir, method:test_home_directory_in_manifest_wins(self) → None, call:get_manifest_home_dir, method:test_user_name_derives_home(self) → None, call:get_manifest_home_dir, method:test_root_fallback(self) → None, call:get_manifest_home_dir, method:test_empty_home_directory_falls_back(self) → None, call:get_manifest_home_dir, class:TestCompileDockerfileHomeDirectory, method:test_env_home_and_workdir_use_project_directory(self) → None, call:compile_dockerfile, method:test_project_directory_created(self) → None, call:compile_dockerfile, method:test_runtime_workspace_not_baked_into_image(self) → None, call:compile_dockerfile, method:test_runtime_working_dir_overrides_home_workdir(self) → None, call:compile_dockerfile, method:test_working_dir_expands_tilde(self) → None, call:compile_dockerfile, class:TestCompileComposeHomeDirectory, method:test_default_repo_mount_synthesized(self) → None, call:compile_compose, method:test_explicit_repo_mount_preserved(self) → None, call:compile_compose, method:test_workspace_name_substituted_in_mount_target(self) → None, call:compile_compose, method:test_working_dir_expands_home(self) → None, call:compile_compose, class:TestCompileEntrypoint, method:test_entrypoint_creates_home_and_project_directory(self) → None, call:compile_entrypoint, method:test_entrypoint_removes_stale_placeholder_directory(self) → None, call:compile_entrypoint, method:test_entrypoint_does_not_create_workspace_symlink(self) → None, call:compile_entrypoint, method:test_entrypoint_fixes_mount_owners(self) → None, call:compile_entrypoint, func:test_compile_dockerfile_creates_config_dirs_for_user() → None, call:compile_dockerfile, func:test_compile_dockerfile_no_user_does_not_create_home() → None, call:compile_dockerfile, func:test_compile_dockerfile_uses_user_npm_prefix() → None, call:compile_dockerfile, func:test_compile_dockerfile_starts_as_root_and_drops_privileges() → None, call:compile_dockerfile, call:compile_entrypoint, func:test_compile_compose_does_not_pin_root_user() → None, call:compile_compose, func:test_get_manifest_container_user_returns_name() → None, call:get_manifest_container_user, func:test_get_manifest_container_user_falls_back_to_uid_gid() → None, call:get_manifest_container_user, func:test_get_manifest_container_user_returns_none_without_user() → None, call:get_manifest_container_user | dep: pytest, src.services.build.manifest_compiler
|
||||
- test_migration_metadata.py | Validates Alembic migration files by dynamically importing them and verifying table definitions and revision chain metadata. | exp: func:test_initial_migration_defines_all_core_tables() → None, call:Path(__file__).resolve, call:spec_from_file_location, call:module_from_spec, call:spec.loader.exec_module, func:test_refresh_tokens_migration_has_expected_revision_chain() → None, call:Path(__file__).resolve, call:spec_from_file_location, call:module_from_spec, call:spec.loader.exec_module | dep: pytest, importlib.util, pathlib, pathlib.Path
|
||||
- test_monitoring_models.py | Unit tests verifying creation, persistence, and querying of monitoring models (InstanceEvent and HealthCheck) with database migration compatibility. | exp: func:test_instance_event_creation(db_session) → None, call:User, call:uuid.uuid4, call:db_session.add, call:db_session.commit, call:ToolInstance, call:InstanceEvent, call:db_session.refresh, call:isinstance, func:test_health_check_creation(db_session) → None, call:User, call:uuid.uuid4, call:db_session.add, call:db_session.commit, call:ToolInstance, call:HealthCheck, call:db_session.refresh, call:isinstance, func:test_instance_event_query_by_instance(db_session) → None, call:User, call:uuid.uuid4, call:db_session.add, call:db_session.commit, call:ToolInstance, call:InstanceEvent, call:db_session.execute, call:select(InstanceEvent).where, call:result.scalar_one | dep: uuid, datetime, pytest, sqlalchemy, src.models.system.health_check, src.models.system.instance_event, src.models.tool.tool_instance, src.models.user.user
|
||||
- test_notification_service.py | Unit tests for NotificationService covering CRUD operations, filtering, ownership validation, and bulk actions on user notifications. | exp: func:notification_service() → NotificationService, call:NotificationService, func:user_a(db_session: AsyncSession) → User, call:User, call:uuid.uuid4, call:db_session.add, call:db_session.commit, func:user_b(db_session: AsyncSession) → User, call:User, call:uuid.uuid4, call:db_session.add, call:db_session.commit, func:test_create_notification(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:uuid.uuid4, func:test_list_notifications_orders_by_created_at_desc(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:datetime.now, call:timedelta, call:db_session.commit, call:db_session.refresh, call:notification_service.list_notifications, func:test_list_notifications_excludes_dismissed(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:notification_service.dismiss, call:notification_service.list_notifications, func:test_list_notifications_unread_only(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:notification_service.mark_read, call:notification_service.list_notifications, func:test_get_unread_count(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:range, call:notification_service.create_notification, call:notification_service.mark_read, call:notification_service.get_unread_count, func:test_mark_read_sets_read_at(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:notification_service.mark_read, func:test_mark_all_read_affects_all_unread(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:range, call:notification_service.create_notification, call:notification_service.mark_all_read, call:notification_service.get_unread_count, func:test_dismiss_sets_dismissed_at(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:notification_service.dismiss, call:db_session.execute, call:select(Notification).where, call:result.scalar_one, func:test_mark_read_wrong_owner_raises(db_session: AsyncSession, notification_service: NotificationService, user_a: User, user_b: User) → None, call:notification_service.create_notification, call:pytest.raises, call:notification_service.mark_read, func:test_dismiss_wrong_owner_raises(db_session: AsyncSession, notification_service: NotificationService, user_a: User, user_b: User) → None, call:notification_service.create_notification, call:pytest.raises, call:notification_service.dismiss, func:test_list_notifications_mute_categories(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:notification_service.list_notifications, func:test_get_unread_count_excludes_dismissed(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:notification_service.create_notification, call:notification_service.dismiss, call:notification_service.get_unread_count, func:test_dismiss_all_affects_all_non_dismissed(db_session: AsyncSession, notification_service: NotificationService, user_a: User) → None, call:range, call:notification_service.create_notification, call:notification_service.dismiss_all, call:notification_service.list_notifications, func:test_dismiss_all_affects_only_caller(db_session: AsyncSession, notification_service: NotificationService, user_a: User, user_b: User) → None, call:range, call:notification_service.create_notification, call:notification_service.dismiss_all, call:notification_service.list_notifications, func:test_mark_all_read_affects_only_caller(db_session: AsyncSession, notification_service: NotificationService, user_a: User, user_b: User) → None, call:range, call:notification_service.create_notification, call:notification_service.mark_all_read, call:notification_service.get_unread_count | dep: uuid, datetime, pytest, sqlalchemy, sqlalchemy.ext.asyncio, src.models.system.notification, src.models.user.user, src.services.shared.notification_service, Notification, User, NotificationService
|
||||
- test_notifications_api_routes.py | Tests FastAPI route ordering to ensure DELETE /notifications matches the bulk clear endpoint before the parameterized DELETE /notifications/{id} endpoint. | 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.system.notifications
|
||||
- test_permission_fixer.py | Unit tests for a Docker container permission fixer 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.shared.permission_fixer, subprocess
|
||||
- test_readiness_probe.py | Unit tests for a Docker container readiness probe service that executes commands with retry logic and timeout handling | exp: class:TestExecuteProbe, class:TestIntegrationScenarios | dep: unittest.mock, src.services.shared.readiness_probe, subprocess
|
||||
- test_ssh_keys.py | Unit tests for SSH key preparation functionality including file creation, permissions, ownership, and error handling. | exp: class:TestPrepareSshKeyFiles | dep: os, pathlib, unittest.mock, pytest, src.services.shared.ssh_keys
|
||||
- test_ssh_keys.py | Unit tests for SSH key file preparation functionality including file creation, permissions, ownership, and error handling. | exp: class:TestPrepareSshKeyFiles | dep: os, pathlib, unittest.mock, pytest, src.services.shared.ssh_keys
|
||||
- test_terminal_session.py | Unit tests for TerminalSession verifying docker exec command construction with/without container user flag | exp: func:test_start_passes_container_user_to_docker_exec() → None, call:TerminalSession, call:str, call:uuid.uuid4, call:patch, call:AsyncMock, call:session.start, call:args.index, func:test_start_omits_user_when_not_configured() → None, call:TerminalSession, call:str, call:uuid.uuid4, call:patch, call:AsyncMock, call:session.start | dep: uuid, unittest.mock, pytest, src.services.terminal.terminal_session, pty, asyncio, os
|
||||
## arch
|
||||
Standard Python unittest/pytest pattern with isolated test modules per service, heavy use of mocking for external dependencies (Docker, Git, subprocess), and tests covering security constraints, error handling, and state transitions across asynchronous and synchronous operations.
|
||||
Follows pytest-based unit testing architecture with heavy use of mocking (subprocess, filesystem, database) to test services in isolation; tests are organized by service/module with descriptive naming; employs parametrization and fixture-based setup for testing edge cases, error conditions, and path traversal prevention across infrastructure and security-critical components.
|
||||
## tags
|
||||
test, url, call:, git, call:notification, home, mounts, mount
|
||||
test, url, call:, git, call:notification, mounts, home, merge
|
||||
## symbols
|
||||
- TestMergeFunctions
|
||||
- TestResolveProfile
|
||||
|
||||
Reference in New Issue
Block a user