fix: pi container repo mount target and npm update permissions

- Add Alembic migration to update built-in pi-agent manifest:
  * repo mount target from /workspace to ~/{{WORKSPACE_NAME}}
  * keep /workspace as compatibility symlink via working_dir
  * update startup chown target to $HOME/$WORKSPACE_NAME
- Pass REPO_NAME and WORKSPACE_NAME to compile_compose from instance_service
- Substitute {{WORKSPACE_NAME}} in manifest mount targets and expose it as
  a container env var so the entrypoint can create the /workspace symlink
- Generate entrypoint workspace symlink from runtime WORKSPACE_NAME env var
- Install npm_global packages into {home_dir}/.npm-global with PATH so the
  non-root container user can update global packages
- Update manifest compiler unit tests for the new behavior

Quality gates:
- pytest tests/unit: 207 passed
- ruff: clean on changed files
- mypy: clean on changed files
- alembic heads: single head
This commit is contained in:
Developer
2026-06-14 18:45:32 +00:00
parent c8db6ce933
commit fe82a248ec
30 changed files with 364 additions and 120 deletions
+1 -1
View File
@@ -2,7 +2,7 @@
dir: apps/api
## role
FastAPI backend API that manages projects, git repositories, and development tools through Docker containers with PostgreSQL persistence.
Self-hosted FastAPI backend API that manages projects, git repositories, and development tools via Docker instances with PostgreSQL database.
## parent
index: apps/.pi-map.index.md
map: apps/.pi-map.md
+2 -2
View File
@@ -4,7 +4,7 @@ dir: apps/api
index: apps/api/.pi-map.index.md
## role
Self-hosted FastAPI backend API that manages projects, git repositories, and development tools via Docker instances.
Self-hosted FastAPI backend API that manages projects, git repositories, and development tools via Docker instances with PostgreSQL database.
## files
- .dockerignore | Specifies files and directories to exclude from Docker build context to reduce image size and avoid copying unnecessary files into containers. | dep: Docker
- Dockerfile | Multi-stage Docker build for a Python application with Docker socket access, Cloudflare tunneling, and database dependency waiting | dep: python:3.11-slim, gcc, libpq-dev, docker-ce-cli, docker-compose-plugin, cloudflared, uvicorn, pyproject.toml dependencies
@@ -14,7 +14,7 @@ Self-hosted FastAPI backend API that manages projects, git repositories, and dev
- uv.lock | Lock file for the uv Python package manager that pins exact dependency versions and their artifact hashes for reproducible installations | dep: uv, Python 3.11+, aiosqlite, alembic, annotated-doc, annotated-types, anyio, ast-serialize, asyncpg, and many other PyPI packages
- wait-for-db.sh | Wait for a PostgreSQL database to become available before executing a command, with configurable retry logic. | dep: nc (netcat), sh (POSIX shell), sleep
## arch
Multi-stage containerized architecture with async PostgreSQL database, Alembic migrations, Cloudflare tunneling, Docker socket access, and uv-based Python dependency management.
Modern Python async architecture using FastAPI with SQLAlchemy/Alembic for database migrations, multi-stage Docker containerization with Docker socket access for container orchestration, Cloudflare tunneling for secure external access, and uv for fast dependency management.
## tags
docker, alembic, python, database, fastapi, postgresql, asyncpg, uvicorn
## symbols
+1 -1
View File
@@ -2,7 +2,7 @@
dir: apps/api/alembic
## role
Database migration infrastructure for the API application, enabling version-controlled schema changes with async SQLAlchemy support.
Database migration infrastructure for the API application using Alembic with async SQLAlchemy support.
## parent
index: apps/api/.pi-map.index.md
map: apps/api/.pi-map.md
+2 -2
View File
@@ -4,12 +4,12 @@ dir: apps/api/alembic
index: apps/api/alembic/.pi-map.index.md
## role
Database migration infrastructure for the API application, enabling version-controlled schema changes with async SQLAlchemy support.
Database migration infrastructure for the API application using Alembic with async SQLAlchemy support.
## files
- env.py | Configures Alembic database migration environment with async SQLAlchemy support for a project. | exp: func:run_migrations_offline() → None, call:context.configure, call:context.begin_transaction, call:context.run_migrations, func:do_run_migrations(connection: Connection) → None, call:context.configure, call:context.begin_transaction, call:context.run_migrations, func:run_async_migrations() → None, call:async_engine_from_config, call:config.get_section, call:connectable.connect, call:connection.run_sync, call:connectable.dispose, func:run_migrations_online() → None, call:asyncio.run, call:run_async_migrations | dep: logging.config, alembic, sqlalchemy, sqlalchemy.engine, sqlalchemy.ext.asyncio, src.config, src.models, asyncio
- script.py.mako | Alembic database migration script template that generates upgrade/downgrade functions for SQLAlchemy schema migrations | dep: alembic, sqlalchemy
## arch
Alembic migration framework with Mako templating for script generation, async SQLAlchemy engine configuration, and environment-based database connection management.
Template-based migration generation with environment configuration for async database operations, following standard Alembic project structure with Mako templating for reproducible schema change scripts.
## tags
migrations, run, sqlalchemy, async, alembic, call:context.configure, call:context.begin, transaction
## symbols
+2 -1
View File
@@ -2,7 +2,7 @@
dir: apps/api/alembic/versions
## role
Database schema evolution and versioning management for the API application using Alembic migration scripts.
Database schema version control and incremental migration management for the API application, tracking the evolution of tables supporting users, SSH keys, projects, git repositories, tool types/instances, config profiles, workspaces, monitoring, notifications, and terminal sessions.
## parent
index: apps/api/alembic/.pi-map.index.md
map: apps/api/alembic/.pi-map.md
@@ -49,6 +49,7 @@ map: apps/api/alembic/.pi-map.md
- 2026_06_01_add_workspaces.py
- 2026_06_13_make_clone_mode_nullable.py
- 2026_06_14_104415_add_tool_type_home_directory.py
- 2026_06_14_182955_fix_pi_agent_home_directory_mount.py
- 398082499c30_add_tool_config_fields.py
- 6fc7bfcf199f_merge_remove_is_builtin_and_add_config_.py
- 86cec91fdb00_merge_profile_resolver_and_workspaces_.py
+4 -3
View File
@@ -4,7 +4,7 @@ dir: apps/api/alembic/versions
index: apps/api/alembic/versions/.pi-map.index.md
## role
Database schema evolution and versioning management for the API application using Alembic migration scripts.
Database schema version control and incremental migration management for the API application, tracking the evolution of tables supporting users, SSH keys, projects, git repositories, tool types/instances, config profiles, workspaces, monitoring, notifications, and terminal sessions.
## files
- 0001_initial_schema.py | Defines the initial database schema migration creating five tables (users, ssh_keys, projects, git_repositories, user_configs) with relationships, indexes, and constraints using Alembic. | exp: func:upgrade() → None, call:op.create_table, call:sa.Column, call:sa.String, call:postgresql.UUID, call:sa.DateTime, call:sa.func.now, call:sa.PrimaryKeyConstraint, call:sa.UniqueConstraint, call:op.create_index, call:op.f, call:sa.Text, call:sa.ForeignKeyConstraint, call:sa.Boolean, call:postgresql.JSONB, func:downgrade() → None, call:op.drop_table, call:op.drop_index, call:op.f | dep: alembic, sqlalchemy.dialects, sqlalchemy, postgresql dialect
- 0002_refresh_tokens.py | Alembic database migration that creates a refresh_tokens table with indexes for user authentication token management | exp: func:upgrade() → None, call:op.get_bind, call:sa.inspect, call:inspector.has_table, call:op.create_table, call:sa.Column, call:postgresql.UUID, call:sa.String, call:sa.DateTime, call:sa.ForeignKeyConstraint, call:sa.PrimaryKeyConstraint, call:sa.UniqueConstraint, call:inspector.get_indexes, call:op.f, call:op.create_index, func:downgrade() → None, call:op.get_bind, call:sa.inspect, call:inspector.has_table, call:inspector.get_indexes, call:op.f, call:op.drop_index, call:op.drop_table | dep: alembic, sqlalchemy.dialects, sqlalchemy, sqlalchemy.dialects.postgresql
@@ -46,6 +46,7 @@ Database schema evolution and versioning management for the API application usin
- 2026_06_01_add_workspaces.py | Alembic database migration that creates a workspaces table with foreign keys to git_repositories and users, adds indexes, and adds a workspace_id column to tool_instances | exp: func:upgrade() → None, call:op.create_table, call:sa.Column, call:sa.Uuid, call:sa.String, call:sa.ForeignKey, call:sa.DateTime, call:sa.text, call:sa.UniqueConstraint, call:op.create_index, call:op.add_column, func:downgrade() → None, call:op.drop_index, call:op.drop_column, call:op.drop_table | dep: collections.abc, alembic, sqlalchemy
- 2026_06_13_make_clone_mode_nullable.py | Alembic database migration that makes the `clone_mode` column in `tool_instances` table nullable to support workspace-first cleanup workflow. | exp: func:upgrade() → None, call:op.alter_column, call:sa.String, func:downgrade() → None, call:op.alter_column, call:sa.String | dep: alembic, sqlalchemy
- 2026_06_14_104415_add_tool_type_home_directory.py | Alembic database migration that adds a `home_directory` column to `tool_types` table and updates template strings to use a configurable workspace path instead of hardcoded `/workspace` | exp: func:upgrade() → None, call:op.add_column, call:sa.Column, call:sa.String, call:op.execute, call:sa.update(tool_types) .where(tool_types.c.compose_template.is_not(None)) .values, call:tool_types.c.compose_template.is_not, call:sa.func.replace, call:sa.update(tool_types) .where(tool_types.c.dockerfile_template.is_not(None)) .values, call:tool_types.c.dockerfile_template.is_not, func:downgrade() → None, call:op.execute, call:sa.update(tool_types) .where(tool_types.c.compose_template.is_not(None)) .values, call:tool_types.c.compose_template.is_not, call:sa.func.replace, call:sa.update(tool_types) .where(tool_types.c.dockerfile_template.is_not(None)) .values, call:tool_types.c.dockerfile_template.is_not, call:op.drop_column | dep: typing, alembic, sqlalchemy.sql, sqlalchemy
- 2026_06_14_182955_fix_pi_agent_home_directory_mount.py | Alembic database migration that fixes the pi-agent tool definition manifest by changing repo mount target from /workspace to ~/{{WORKSPACE_NAME}}, updating working directory, and adjusting startup script ownership | exp: func:_find_pi_agent_manifest(conn: sa.Connection) → tuple[Union[str, None], Union[dict, None]], call:conn.execute( sa.select(tool_definition_manifests.c.id, tool_definition_manifests.c.manifest) .where(tool_definition_manifests.c.name == "pi-agent") ).fetchone, call:sa.select(tool_definition_manifests.c.id, tool_definition_manifests.c.manifest) .where, call:dict, func:_update_manifest(conn: sa.Connection, manifest_id: str, manifest: dict) → None, call:conn.execute, call:sa.update(tool_definition_manifests) .where(tool_definition_manifests.c.id == manifest_id) .values, func:upgrade() → None, call:op.get_bind, call:_find_pi_agent_manifest, call:manifest.get, call:mount.get, call:manifest.setdefault, call:_update_manifest, func:downgrade() → None, call:op.get_bind, call:_find_pi_agent_manifest, call:manifest.get, call:mount.get, call:manifest.setdefault, call:_update_manifest | dep: typing, alembic, sqlalchemy.sql, sqlalchemy
- 398082499c30_add_tool_config_fields.py | Alembic database migration that adds five new columns (port_override, start_command, working_directory, environment_variables, volumes) to the tool_configs table with a port range check constraint. | exp: func:upgrade() → None, call:op.add_column, call:sa.Column, call:sa.Integer, call:sa.Text, call:postgresql.JSONB, call:op.create_check_constraint, call:sa.text, func:downgrade() → None, call:op.drop_constraint, call:op.drop_column | dep: alembic, sqlalchemy.dialects, sqlalchemy, sqlalchemy.dialects.postgresql
- 6fc7bfcf199f_merge_remove_is_builtin_and_add_config_.py | Alembic database migration that merges two parallel revision branches (removing is_builtin and adding config_profiles) into a single history line | exp: func:upgrade() → None, func:downgrade() → None | dep: alembic
- 86cec91fdb00_merge_profile_resolver_and_workspaces_.py | Alembic database migration that merges two divergent migration branches (profile resolver and workspaces) into a single head | exp: func:upgrade() → None, func:downgrade() → None | dep: alembic
@@ -53,9 +54,9 @@ Database schema evolution and versioning management for the API application usin
- 8ed7dd80973d_create_config_folders_table.py | Alembic database migration that creates a config_folders table with user-owned configuration folders supporting JSONB file storage and project overrides | exp: func:upgrade() → None, call:op.create_table, call:sa.Column, call:postgresql.UUID, call:sa.text, call:sa.ForeignKey, call:sa.String, call:sa.Text, call:postgresql.JSONB, call:sa.Boolean, call:sa.DateTime, call:sa.UniqueConstraint, call:op.create_index, func:downgrade() → None, call:op.drop_index, call:op.drop_table | dep: alembic, sqlalchemy.dialects, sqlalchemy, sqlalchemy.dialects.postgresql
- af8512103d67_add_tool_type_fields.py | Alembic database migration that adds new columns (definition_type, dockerfile_template, build_context, readiness_probe) to the tool_types table with a CHECK constraint on definition_type. | exp: func:upgrade() → None, call:op.add_column, call:sa.Column, call:sa.String, call:sa.Text, call:postgresql.JSONB, call:op.create_check_constraint, call:sa.text, func:downgrade() → None, call:op.drop_constraint, call:op.drop_column | dep: alembic, sqlalchemy.dialects, sqlalchemy, sqlalchemy.dialects.postgresql
- f3d2dc90ba3a_merge_single_interface_and_clone_mode.py | Alembic database migration that merges two prior revisions (single_interface and clone_mode) into a single migration path | exp: func:upgrade() → None, func:downgrade() → None | dep: typing, alembic
- fc8f1a20cbf6_merge_home_directory_and_pi_agent_mount_.py | Alembic database migration that merges two revision branches by declaring them as down revisions without performing any schema changes. | exp: func:upgrade() → None, func:downgrade() → None | dep: alembic, sqlalchemy
- fc8f1a20cbf6_merge_home_directory_and_pi_agent_mount_.py | Alembic database migration that merges two divergent migration branches (home directory cleanup and pi agent mount cleanup) into a single revision history | exp: func:upgrade() → None, func:downgrade() → None | dep: alembic
## arch
Sequential and branched migration pattern with merge resolution, using Alembic's revision-based approach with upgrade/downgrade functions, including data migrations, conditional schema changes, and cross-dialect support (PostgreSQL/SQLite).
Linear and branched Alembic migration history with merge migrations reconciling divergent branches; each migration is an autonomous upgrade/downgrade script using SQLAlchemy operations, with some including data migrations, conditional logic for idempotency, dialect-specific handling (PostgreSQL/SQLite), and direct file-system modifications alongside schema changes.
## tags
column, table, call:op.drop, alembic, downgrade, upgrade, key, call:sa.text
## symbols
@@ -0,0 +1,90 @@
"""fix_pi_agent_home_directory_mount
Revision ID: 2026_06_14_182955
Revises: fc8f1a20cbf6
Create Date: 2026-06-14 18:29:55.000000
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
from sqlalchemy.sql import column, table
# revision identifiers, used by Alembic.
revision: str = "2026_06_14_182955"
down_revision: Union[str, Sequence[str], None] = "fc8f1a20cbf6"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
tool_definition_manifests = table(
"tool_definition_manifests",
column("id", sa.UUID),
column("name", sa.String),
column("manifest", sa.JSON),
)
def _find_pi_agent_manifest(conn: sa.Connection) -> tuple[Union[str, None], Union[dict, None]]:
result = conn.execute(
sa.select(tool_definition_manifests.c.id, tool_definition_manifests.c.manifest)
.where(tool_definition_manifests.c.name == "pi-agent")
).fetchone()
if result is None:
return None, None
return result.id, dict(result.manifest)
def _update_manifest(conn: sa.Connection, manifest_id: str, manifest: dict) -> None:
conn.execute(
sa.update(tool_definition_manifests)
.where(tool_definition_manifests.c.id == manifest_id)
.values(manifest=manifest)
)
def upgrade() -> None:
conn = op.get_bind()
manifest_id, manifest = _find_pi_agent_manifest(conn)
if not manifest_id or not manifest:
return
# Mount the repo under the configured home directory, preserving the repo
# directory name via the WORKSPACE_NAME runtime variable.
for mount in manifest.get("mounts", []):
if mount.get("source_type") == "repo":
mount["target"] = "~/{{WORKSPACE_NAME}}"
# Keep /workspace as a compatibility symlink to the real mount path.
runtime = manifest.setdefault("runtime", {})
runtime["working_dir"] = "/workspace"
# Update the startup script to chown the real mount path.
scripts = manifest.setdefault("scripts", {})
scripts["startup"] = [
'if [ -n "$WORKSPACE_NAME" ]; then sudo chown -R user:user "$HOME/$WORKSPACE_NAME" 2>/dev/null || true; fi',
]
_update_manifest(conn, manifest_id, manifest)
def downgrade() -> None:
conn = op.get_bind()
manifest_id, manifest = _find_pi_agent_manifest(conn)
if not manifest_id or not manifest:
return
for mount in manifest.get("mounts", []):
if mount.get("source_type") == "repo":
mount["target"] = "/workspace"
runtime = manifest.setdefault("runtime", {})
runtime["working_dir"] = "/workspace"
scripts = manifest.setdefault("scripts", {})
scripts["startup"] = [
"if [ -d /workspace ]; then sudo chown -R user:user /workspace 2>/dev/null || true; fi",
]
_update_manifest(conn, manifest_id, manifest)
+1 -1
View File
@@ -2,7 +2,7 @@
dir: apps/api/src
## role
Core application bootstrap and infrastructure layer for the Headquarter API, providing configuration, database connectivity, logging, and FastAPI application initialization.
Core API application package for the "Headquarter API" providing configuration, database infrastructure, logging, and FastAPI application initialization.
## parent
index: apps/api/.pi-map.index.md
map: apps/api/.pi-map.md
+2 -2
View File
@@ -4,7 +4,7 @@ dir: apps/api/src
index: apps/api/src/.pi-map.index.md
## role
Core API application package for the Headquarter service, providing configuration, database connectivity, structured logging, and FastAPI application initialization.
Core API application package for the "Headquarter API" providing configuration, database infrastructure, logging, and FastAPI application initialization.
## files
- __init__.py | Marks the directory as a Python package for the Headquarter API.
- config.py | Defines application configuration settings with environment-based overrides using Pydantic, including database URLs, service domains, OAuth/Authentik integration, JWT/session settings, and computed properties for environment-specific behavior. | exp: class:Settings, func:build_database_url(user: str, password: str, host: str, port: int, database: str) → str | dep: pydantic, pydantic_settings
@@ -12,7 +12,7 @@ Core API application package for the Headquarter service, providing configuratio
- logging_config.py | Configures structured JSON logging with correlation ID injection, custom formatters, and HTTP request/exception middleware for a FastAPI application. | exp: class:CorrelationIdFilter, method:filter(self, record: logging.LogRecord) → bool, call:get_correlation_id, class:JSONFormatter, method:format(self, record: logging.LogRecord) → str, call:self.formatTime, call:record.getMessage, call:getattr, call:self.formatException, call:json.dumps, method:formatTime(self, record: logging.LogRecord, datefmt) → str, call:time.strftime, call:time.gmtime, class:RequestLoggingMiddleware, method:dispatch(self, request: Request, call_next: Callable) → Response, call:time.time, call:logger.info, call:call_next, call:int, call:logger.error, call:type, call:traceback.format_exc, class:ExceptionLoggingMiddleware, method:dispatch(self, request: Request, call_next: Callable) → Response, call:call_next, call:logger.critical, call:traceback.format_exc, func:configure_logging(level) → None, call:JSONFormatter, call:logging.StreamHandler, call:console_handler.setFormatter, call:console_handler.addFilter, call:CorrelationIdFilter, call:root_logger.setLevel, call:logging.getLogger("uvicorn").setLevel, call:logging.getLogger("uvicorn.access").setLevel, call:logging.getLogger("sqlalchemy.engine").setLevel, call:logger.info, call:logging.getLevelName | dep: json, logging, sys, time, traceback, collections.abc, fastapi, starlette.middleware.base, src.services.shared.correlation
- main.py | Initializes and configures a FastAPI application for the "Headquarter API" with database setup, middleware, routing, and background services. | exp: func:_sanitize_validation_errors(errors), call:error.get, call:str, call:ctx.items, call:isinstance, call:type, call:sanitized.append, func:validation_exception_handler(request: Request, exc: RequestValidationError), call:exc.errors, call:logger.warning, call:_sanitize_validation_errors, call:JSONResponse, func:on_startup(), call:logger.info, call:init_database, call:logger.error, call:sys.exit, call:_health_monitor.start, call:seed_builtin_tool_types, func:on_shutdown(), call:logger.info, call:_health_monitor.stop | dep: logging, os, fastapi, fastapi.exceptions, fastapi.middleware.cors, fastapi.responses, fastapi.staticfiles, src.api.config, src.api.project, src.api.system, src.api.tool, src.api.user, src.api.workspace, src.config, src.models, src.database, src.logging_config, src.seeds.builtin_tool_types, src.services.instance, src.services.shared, sys, src.api.*
## arch
Layered architecture with environment-based Pydantic configuration, async SQLAlchemy with Alembic migrations, structured JSON logging with correlation IDs, and FastAPI middleware/routing pattern with background services integration.
Layered architecture with Pydantic-based config management, async SQLAlchemy with Alembic migrations, structured JSON logging with correlation IDs, and FastAPI middleware/routing pattern.
## tags
src, database, logging, call:logger.info, api, middleware, fastapi, filter
## symbols
+1 -1
View File
@@ -2,7 +2,7 @@
dir: apps/api/src/services
## role
Marks the services directory as a Python package for organizing business logic and service-layer abstractions.
Provides business logic and service layer abstractions for the API application.
## parent
index: apps/api/src/.pi-map.index.md
map: apps/api/src/.pi-map.md
+2 -2
View File
@@ -4,11 +4,11 @@ dir: apps/api/src/services
index: apps/api/src/services/.pi-map.index.md
## role
This directory serves as a Python package namespace for organizing service-layer modules in the API application.
Provides business logic and service layer abstractions for the API application.
## files
- __init__.py | Empty file with no functionality
## arch
Standard Python package structure using __init__.py to define an importable module directory, following conventional layered architecture patterns.
Minimal or placeholder package structure with no implemented services yet, following standard Python package conventions.
## tags
init, empty, functionality
## symbols
@@ -143,13 +143,6 @@ def compile_dockerfile(manifest: dict) -> str:
lines.append(" rm -rf /var/lib/apt/lists/*")
lines.append("")
# NPM global packages
npm_packages = manifest.get("packages", {}).get("npm_global", [])
if npm_packages:
pkg_list = " ".join(shlex.quote(p) for p in npm_packages)
lines.append(f"RUN npm install -g {pkg_list}")
lines.append("")
# Pip packages
pip_packages = manifest.get("packages", {}).get("pip", [])
if pip_packages:
@@ -161,6 +154,7 @@ def compile_dockerfile(manifest: dict) -> str:
user = manifest.get("user")
home_dir = get_manifest_home_dir(manifest)
workspace_name = manifest.get("workspace_name", "{{WORKSPACE_NAME}}")
npm_prefix = ""
if user:
name = user["name"]
uid = user["uid"]
@@ -195,6 +189,20 @@ def compile_dockerfile(manifest: dict) -> str:
)
lines.append("")
# NPM global packages: install into a user-writable prefix so the
# container user can update global packages without touching
# /usr/lib/node_modules (which is owned by root).
npm_packages = manifest.get("packages", {}).get("npm_global", [])
if npm_packages:
pkg_list = " ".join(shlex.quote(p) for p in npm_packages)
npm_prefix = f"{home_dir}/.npm-global"
lines.append(
f"RUN mkdir -p {npm_prefix} && "
f"npm install -g --prefix {npm_prefix} {pkg_list}"
)
lines.append(f"ENV PATH={npm_prefix}/bin:$PATH")
lines.append("")
# Build scripts
build_scripts = manifest.get("scripts", {}).get("build", [])
for script in build_scripts:
@@ -287,8 +295,6 @@ def compile_entrypoint(manifest: dict) -> str:
user = manifest.get("user")
home_dir = get_manifest_home_dir(manifest)
workspace_name = manifest.get("workspace_name", "{{WORKSPACE_NAME}}")
workspace_target = f"{home_dir}/{workspace_name}"
# Permission fixer preamble: run as root when possible, else fall back to
# passwordless sudo configured in the Dockerfile.
@@ -315,7 +321,8 @@ def compile_entrypoint(manifest: dict) -> str:
lines.append(f"USER_UID='{uid}'")
lines.append(f"USER_GID='{gid}'")
lines.append(f"HOME_DIR='{home_dir}'")
lines.append(f"WORKSPACE_TARGET='{workspace_target}'")
lines.append('WORKSPACE_NAME="${WORKSPACE_NAME:-workspace}"')
lines.append('WORKSPACE_TARGET="${HOME_DIR}/${WORKSPACE_NAME}"')
lines.append("")
lines.append("fix_owner() {")
lines.append(" local path=\"$1\"")
@@ -343,8 +350,13 @@ def compile_entrypoint(manifest: dict) -> str:
target = mount.get("target")
if not target:
continue
# Expand any ~/$HOME placeholders in the mount target.
expanded = target.replace("~", home_dir).replace("$HOME", home_dir)
# Expand any ~/$HOME placeholders and the runtime workspace name
# in the mount target so ownership is fixed at container startup.
expanded = (
target.replace("~", home_dir)
.replace("$HOME", home_dir)
.replace("{{WORKSPACE_NAME}}", "${WORKSPACE_NAME}")
)
if expanded.startswith(home_dir) and not mount.get("readonly", False):
lines.append(f'fix_owner "{expanded}"')
lines.append("")
@@ -409,6 +421,12 @@ def compile_compose(manifest: dict, variables: dict[str, Any]) -> str:
if env:
service["environment"] = dict(env)
# Expose the workspace/repo name so the entrypoint can finalize the
# /workspace compatibility symlink at container startup.
if "environment" not in service:
service["environment"] = {}
service["environment"]["WORKSPACE_NAME"] = workspace_name
# Merge extra env from config
extra_env = variables.get("EXTRA_ENV", {})
if extra_env:
@@ -426,6 +444,7 @@ def compile_compose(manifest: dict, variables: dict[str, Any]) -> str:
if mount.get("source_type") == "repo":
has_explicit_repo_mount = True
target = expand_container_path(mount["target"], home_dir)
target = target.replace("{{WORKSPACE_NAME}}", workspace_name)
readonly = ":ro" if mount.get("readonly", False) else ""
volumes.append(f"{source}:{target}{readonly}")
+6 -2
View File
@@ -2,12 +2,14 @@
dir: apps/api/src/services/tool
## role
Orchestrates end-to-end deployment and runtime management of development tool instances via containerized environments with remote access capabilities.
Provides containerized execution environment for tools by managing Docker instances, git repositories, and compose orchestration.
## parent
index: apps/api/src/services/.pi-map.index.md
map: apps/api/src/services/.pi-map.md
## children
-
- apps/api/src/services/tool/.ruff_cache
index: apps/api/src/services/tool/.ruff_cache/.pi-map.index.md
map: apps/api/src/services/tool/.ruff_cache/.pi-map.md
## files
- instance_service.py
## links
@@ -16,5 +18,7 @@ map: apps/api/src/services/tool/.pi-map.md
## workflows
- change tool behavior
read: instance_service.py
- explore tool subdirectories
index: apps/api/src/services/tool/.ruff_cache/.pi-map.index.md
## dirty
-
File diff suppressed because one or more lines are too long
@@ -880,12 +880,15 @@ async def prepare_manifest_instance(
# The actual resolution happens in resolve_git_mounts; we store placeholder
git_mount_vars[f"GIT_MOUNT_{ref}"] = ""
repo_name = os.path.basename(os.path.normpath(repo_path))
variables = {
"IMAGE_TAG": image_tag,
"INSTANCE_NAME": instance.name.lower(),
"INSTANCE_DIR": instance_dir,
"WORKSPACE_PATH": repo_path,
"REPO_PATH": repo_path,
"REPO_NAME": repo_name,
"WORKSPACE_NAME": repo_name,
"SSH_PATH": ssh_path,
"TOOL_PORT": instance.port or 0,
"EXTRA_ENV": env_vars,
@@ -1070,12 +1073,15 @@ async def create_tool_instance(
# Manifest templates use WORKSPACE_PATH; REPO_PATH is retained as a
# deprecated alias for backward compatibility with older templates.
repo_name = os.path.basename(os.path.normpath(repo_path))
variables = {
"IMAGE_TAG": image_tag,
"INSTANCE_NAME": instance_name.lower(),
"INSTANCE_DIR": instance_dir,
"WORKSPACE_PATH": repo_path,
"REPO_PATH": repo_path,
"REPO_NAME": repo_name,
"WORKSPACE_NAME": repo_name,
"SSH_PATH": "",
"TOOL_PORT": tool_port,
"EXTRA_ENV": {},
+1 -1
View File
@@ -2,7 +2,7 @@
dir: apps/api/tests
## role
Provides shared test infrastructure and fixtures for the API application's test suite.
Provides shared test infrastructure and fixtures for API integration testing with async database and authentication support.
## parent
index: apps/api/.pi-map.index.md
map: apps/api/.pi-map.md
+2 -2
View File
@@ -4,11 +4,11 @@ dir: apps/api/tests
index: apps/api/tests/.pi-map.index.md
## role
Provides shared test infrastructure and fixtures for the API application's test suite.
Provides shared test infrastructure and fixtures for API integration testing with async database and authentication support.
## files
- conftest.py | Provides shared pytest fixtures for testing a FastAPI application 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.user, src.models.project.project, src.models.project.git_repository, fastapi, sqlalchemy, aiosqlite, src.models, src.auth
## arch
Pytest plugin architecture with dependency-injected async fixtures for database, HTTP client, and authentication state management.
Pytest fixture-based architecture using dependency injection for async SQLite database sessions, JWT-authenticated HTTP clients, and modular test data setup across FastAPI test suites.
## tags
call:app.dependency, call:create, overrides.get, call:override, fn, call:gen.asend, call:gen.aclose, user
## symbols
+1 -1
View File
@@ -2,7 +2,7 @@
dir: apps/api/tests/unit
## role
Unit test suite for the API application covering core services, utilities, and infrastructure components.
Contains unit tests for the API application's core services, utilities, and infrastructure components.
## parent
index: apps/api/tests/.pi-map.index.md
map: apps/api/tests/.pi-map.md
+5 -5
View File
@@ -4,10 +4,10 @@ dir: apps/api/tests/unit
index: apps/api/tests/unit/.pi-map.index.md
## role
Unit test suite for the API application covering core services, utilities, and infrastructure components.
Contains unit tests for the API application's core services, utilities, and infrastructure components.
## files
- __init__.py | Empty file with no functionality
- test_alembic_migrations.py | Unit tests that verify Alembic database migrations are importable, have correct 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 | dep: importlib.util, pathlib, pytest, importlib
- 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 | dep: importlib.util, pathlib, pytest, importlib
- test_config.py | Tests configuration settings and database URL building for an application, verifying defaults, environment variable overrides, and environment-specific behavior. | 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 | Tests the config profile resolution system 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 Dockerfile creation, command structure, context file handling, path traversal prevention, and error handling. | exp: class:TestBuildImage | dep: subprocess, tempfile, pathlib, unittest.mock, pytest, src.services.build.docker_build
@@ -19,9 +19,9 @@ Unit test suite for the API application covering core services, utilities, and i
- test_git_url_parser.py | Tests for git URL parsing utilities that extract base repository URLs, validate clone URLs, and parse various git URL formats across GitHub, GitLab, and Bitbucket. | 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, exception resilience, and start/stop lifecycle. | 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 home directory path expansion (~ and $HOME) in container paths and manifest 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 home directory expansion in docker-compose file modifications. | 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 | dep: pytest, src.services.tool.instance_service
- test_instance_service.py | Unit tests for home directory expansion in docker-compose file modification via instance_service | 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 | dep: pytest, src.services.tool.instance_service
- 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 Docker configurations (Dockerfile, compose, entrypoint) based on manifest specifications. | 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_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_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_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 | dep: pytest, src.services.build.manifest_compiler
- 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_home_directory(self) → None, call:compile_dockerfile, method:test_workspace_symlink_created(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_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 | dep: pytest, src.services.build.manifest_compiler
- test_migration_metadata.py | Tests Alembic database migration files for correct 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, sorting, and ownership isolation. | 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, NotificationService, Notification, User, AsyncSession
@@ -30,7 +30,7 @@ Unit test suite for the API application covering core services, utilities, and i
- test_readiness_probe.py | Unit tests for a Docker container readiness probe service that executes commands via docker exec with retry logic. | 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
## arch
pytest-based unit testing with heavy mocking of external dependencies (Docker, Git, subprocess, database) to test components in isolation without live services.
Standard Python unittest/pytest pattern with heavy use of mocking for external dependencies (Docker, Git, subprocess, database) to test business logic in isolation without live services.
## tags
test, url, call:notification, git, home, call:, merge, call:db
## symbols
+44 -1
View File
@@ -92,7 +92,8 @@ class TestCompileDockerfileHomeDirectory:
}
dockerfile = compile_dockerfile(manifest)
assert "ln -sfn /home/custom/{{WORKSPACE_NAME}} /workspace" in dockerfile
assert "mkdir -p /home/custom" in dockerfile
assert "ln -sfn" in dockerfile
def test_runtime_working_dir_overrides_home_workdir(self) -> None:
manifest = {
@@ -168,6 +169,30 @@ class TestCompileComposeHomeDirectory:
assert "/host/repos/my-app:/opt/code:ro" in compose
assert "/home/custom/my-app" not in compose
def test_workspace_name_substituted_in_mount_target(self) -> None:
manifest = {
"base_image": "ubuntu:24.04",
"interface_type": "terminal",
"home_directory": "/home/custom",
"user": {"name": "dev", "uid": 1000, "gid": 1000},
"mounts": [
{"source_type": "repo", "target": "~/{{WORKSPACE_NAME}}"}
],
}
variables = {
"IMAGE_TAG": "test:latest",
"INSTANCE_NAME": "test-instance",
"REPO_PATH": "/host/repos/my-app",
"WORKSPACE_NAME": "my-app",
"TOOL_PORT": 0,
"EXTRA_ENV": {},
"EXTRA_VOLUMES": [],
}
compose = compile_compose(manifest, variables)
assert "/host/repos/my-app:/home/custom/my-app" in compose
assert "WORKSPACE_NAME: my-app" in compose
def test_working_dir_expands_home(self) -> None:
manifest = {
"base_image": "ubuntu:24.04",
@@ -190,6 +215,23 @@ class TestCompileComposeHomeDirectory:
assert "working_dir: /home/custom/code" in compose
@pytest.mark.unit
def test_compile_dockerfile_uses_user_npm_prefix() -> None:
"""npm global packages must be installed into a user-writable prefix."""
manifest = {
"base_image": "ubuntu:24.04",
"interface_type": "terminal",
"home_directory": "/home/custom",
"user": {"name": "dev", "uid": 1000, "gid": 1000},
"packages": {"npm_global": ["@scope/pkg"]},
}
dockerfile = compile_dockerfile(manifest)
assert "npm install -g --prefix /home/custom/.npm-global" in dockerfile
assert "/home/custom/.npm-global/bin:$PATH" in dockerfile
assert "ENV PATH=/home/custom/.npm-global/bin:$PATH" in dockerfile
@pytest.mark.unit
class TestCompileEntrypoint:
"""Tests for the generated permission-fixing entrypoint."""
@@ -206,6 +248,7 @@ class TestCompileEntrypoint:
assert 'mkdir -p "$HOME_DIR"' in entrypoint
assert 'mkdir -p "$WORKSPACE_TARGET"' in entrypoint
assert 'ln -sfn "$WORKSPACE_TARGET" /workspace' in entrypoint
assert 'WORKSPACE_NAME="${WORKSPACE_NAME:-workspace}"' in entrypoint
def test_entrypoint_fixes_mount_owners(self) -> None:
manifest = {