codexpi/tests/test_docker_django.py

from __future__ import annotations

import os
import subprocess
import sys
from pathlib import Path

import yaml


ROOT = Path(__file__).resolve().parents[1]


def test_pi_repair_launcher_recovers_when_invoked_with_sh():
    completed = subprocess.run(
        ["sh", str(ROOT / "repair-rpi-with-codex.sh"), "--help"],
        cwd=ROOT,
        capture_output=True,
        text=True,
        timeout=10,
        check=False,
    )

    assert completed.returncode == 0, completed.stdout + completed.stderr
    assert "Usage: ./repair-rpi-with-codex.sh" in completed.stdout
    assert completed.stderr == ""


def test_pi_repair_launcher_enables_managed_worktrees_for_current_codex_cli():
    launcher = (ROOT / "repair-rpi-with-codex.sh").read_text(encoding="utf-8")

    assert "--enable worktrees" in launcher
    assert launcher.index("--enable worktrees") < launcher.index("--worktree")


def test_compose_only_publishes_nginx_on_configured_lan_port():
    compose = yaml.safe_load((ROOT / "docker-compose.yml").read_text(encoding="utf-8"))
    services = compose["services"]
    assert "ports" not in services["api"]
    assert "ports" not in services["django-web"]
    assert services["nginx"]["ports"] == ["${CODEXPI_BIND_ADDRESS:-0.0.0.0}:${CODEXPI_PROXY_PORT:-8765}:8080"]
    assert "uvicorn django_ui.asgi:application" in services["django-web"]["command"]
    assert "uvicorn codexpi.main:app" in services["api"]["command"]


def test_nginx_waits_for_application_dns_and_health_after_an_abrupt_reboot():
    compose = yaml.safe_load((ROOT / "docker-compose.yml").read_text(encoding="utf-8"))
    nginx = compose["services"]["nginx"]
    command = nginx["command"][-1]

    assert "http://$${django_upstream}:8001/api/bootstrap" in command
    assert "http://$${api_upstream}:8000/api/health" in command
    assert "until" in command
    assert nginx["healthcheck"]["test"][-1] == "http://127.0.0.1:8080/proxy-health"


def test_api_allows_nested_codex_sandbox_without_dropping_other_guards():
    compose = yaml.safe_load((ROOT / "docker-compose.yml").read_text(encoding="utf-8"))
    api = compose["services"]["api"]

    # Bubblewrap needs unprivileged namespace syscalls inside Docker. The API
    # remains non-root, read-only, capability-free, and cannot gain privileges.
    assert "seccomp:unconfined" in api["security_opt"]
    assert "no-new-privileges:true" in api["security_opt"]
    assert api["read_only"] is True
    assert api["cap_drop"] == ["ALL"]


def test_api_receives_the_current_docker_socket_group_without_extra_capabilities():
    compose = yaml.safe_load((ROOT / "docker-compose.yml").read_text(encoding="utf-8"))
    api = compose["services"]["api"]

    assert api["group_add"] == ["${DOCKER_GID:-998}"]
    assert "/var/run/docker.sock:/var/run/docker.sock" in api["volumes"]
    assert api["user"] == "${PUID:-1000}:${PGID:-1000}"
    assert api["cap_drop"] == ["ALL"]


def test_api_receives_host_submounts_without_propagating_container_mounts_back():
    compose = yaml.safe_load((ROOT / "docker-compose.yml").read_text(encoding="utf-8"))
    api = compose["services"]["api"]

    projects_mount = next(
        mount
        for mount in api["volumes"]
        if isinstance(mount, dict) and mount.get("target") == "/home/config"
    )
    assert projects_mount == {
        "type": "bind",
        "source": "/home/config",
        "target": "/home/config",
        "bind": {"propagation": "rslave"},
    }


def test_boot_service_refreshes_docker_access_as_the_non_root_host_user():
    unit = (ROOT / "deploy" / "systemd" / "codexpi.service").read_text(encoding="utf-8")
    repair_unit = (ROOT / "deploy" / "systemd" / "codexpi-docker-access.service").read_text(encoding="utf-8")
    repair_timer = (ROOT / "deploy" / "systemd" / "codexpi-docker-access.timer").read_text(encoding="utf-8")
    compose_wrapper = (ROOT / "scripts" / "compose.sh").read_text(encoding="utf-8")
    repair_script = (ROOT / "scripts" / "ensure-docker-access.sh").read_text(encoding="utf-8")
    installer = (ROOT / "scripts" / "install-boot-service.sh").read_text(encoding="utf-8")
    install_script = (ROOT / "scripts" / "install.sh").read_text(encoding="utf-8")

    assert "BindsTo=docker.service" in unit
    assert "PartOf=docker.service" in unit
    assert "User=__CODEXPI_UID__" in unit
    assert "Group=__CODEXPI_GID__" in unit
    assert "SupplementaryGroups=__DOCKER_GROUP__" in unit
    assert "ExecStart=/home/config/CodexPi/scripts/compose.sh up -d" in unit
    assert 'stat -c \'%g\' "$docker_socket"' in compose_wrapper
    assert 'export DOCKER_GID=' in compose_wrapper
    assert "User=__CODEXPI_UID__" in repair_unit
    assert "SupplementaryGroups=__DOCKER_GROUP__" in repair_unit
    assert "NoNewPrivileges=true" in repair_unit
    assert "ConditionPathIsSocket=/var/run/docker.sock" in repair_unit
    assert "OnUnitActiveSec=60s" in repair_timer
    assert "docker inspect --format '{{json .HostConfig.GroupAdd}}'" in repair_script
    assert "codexpi\\.session_worker" in repair_script
    assert 'up -d --no-deps --force-recreate api' in repair_script
    assert "stat -c '%u' \"$BASE\"" in installer
    assert "s/__CODEXPI_UID__/$codexpi_uid/g" in installer
    assert "s/__DOCKER_GROUP__/$docker_group/g" in installer
    assert "enable --now codexpi-docker-access.timer" in installer
    assert '"$BASE/scripts/install-boot-service.sh"' in install_script


def test_image_has_no_virtualenv_or_root_runtime():
    dockerfile = (ROOT / "Dockerfile").read_text(encoding="utf-8")
    assert "python3 -m venv" not in dockerfile
    assert "USER node" in dockerfile
    assert "@openai/codex" in dockerfile


def test_non_root_nginx_uses_writable_tmpfs_paths():
    nginx = (ROOT / "deploy" / "nginx" / "nginx.conf").read_text(encoding="utf-8")
    for directive in ("client_body", "proxy", "fastcgi", "uwsgi", "scgi"):
        assert f"{directive}_temp_path /tmp/{directive};" in nginx
    assert "/var/cache/nginx/" not in nginx
    assert "location ^~ /api/conversations/" in nginx
    assert "client_max_body_size 101m;" in nginx
    assert "proxy_request_buffering off;" in nginx


def test_large_conversation_responses_are_not_buffered_in_small_nginx_tmpfs():
    nginx = (ROOT / "deploy" / "nginx" / "nginx.conf").read_text(encoding="utf-8")
    conversations_location = nginx.split("location ^~ /api/conversations/ {", 1)[1].split("}", 1)[0]

    assert "proxy_buffering off;" in conversations_location


def test_loutre_finance_has_a_dedicated_private_application_route():
    nginx = (ROOT / "deploy" / "nginx" / "nginx.conf").read_text(encoding="utf-8")

    assert "location = /loutre-finance { return 308 /loutre-finance/; }" in nginx
    finance = nginx.split("location ^~ /loutre-finance/ {", 1)[1].split("}", 1)[0]
    assert "proxy_pass http://loutre-finance:8000;" in finance
    assert "proxy_set_header Cookie $http_cookie;" in finance
    assert "proxy_set_header X-Forwarded-Proto $forwarded_proto;" in finance


def test_smart_inventory_has_a_dedicated_private_application_route():
    nginx = (ROOT / "deploy" / "nginx" / "nginx.conf").read_text(encoding="utf-8")

    assert "location = /smart-inventory { return 308 /smart-inventory/; }" in nginx
    inventory = nginx.split("location ^~ /smart-inventory/ {", 1)[1].split("}", 1)[0]
    assert "proxy_pass http://smart-inventory:8000/;" in inventory
    assert "proxy_set_header Cookie $http_cookie;" in inventory
    assert "proxy_set_header X-Forwarded-Proto $forwarded_proto;" in inventory


def test_auto_care_has_a_dedicated_private_application_route():
    nginx = (ROOT / "deploy" / "nginx" / "nginx.conf").read_text(encoding="utf-8")

    assert "location = /auto-care { return 308 /auto-care/; }" in nginx
    auto_care = nginx.split("location ^~ /auto-care/ {", 1)[1].split("}", 1)[0]
    assert "proxy_pass http://auto-care:8000;" in auto_care
    assert "proxy_set_header Cookie $http_cookie;" in auto_care
    assert "proxy_set_header X-Forwarded-Proto $forwarded_proto;" in auto_care


def test_print_pilot_has_a_dedicated_private_application_route():
    nginx = (ROOT / "deploy" / "nginx" / "nginx.conf").read_text(encoding="utf-8")

    assert "location = /print-pilot { return 308 /print-pilot/; }" in nginx
    print_pilot = nginx.split("location ^~ /print-pilot/ {", 1)[1].split("}", 1)[0]
    assert "proxy_pass http://print-pilot:8000;" in print_pilot
    assert "proxy_set_header Cookie $http_cookie;" in print_pilot
    assert 'proxy_set_header X-User "";' in print_pilot
    assert 'proxy_set_header X-Role "";' in print_pilot


def test_loutre_home_has_a_dedicated_private_application_route():
    nginx = (ROOT / "deploy" / "nginx" / "nginx.conf").read_text(encoding="utf-8")

    assert "location = /loutre-home { return 308 /loutre-home/; }" in nginx
    loutre_home = nginx.split("location ^~ /loutre-home/ {", 1)[1].split("}", 1)[0]
    assert "proxy_pass http://loutre-home:8017;" in loutre_home
    assert "proxy_set_header Cookie $http_cookie;" in loutre_home
    assert 'proxy_set_header X-User "";' in loutre_home
    assert 'proxy_set_header X-Role "";' in loutre_home


def test_network_guardian_has_a_dedicated_private_application_route():
    nginx = (ROOT / "deploy" / "nginx" / "nginx.conf").read_text(encoding="utf-8")

    assert "location = /network-guardian { return 308 /network-guardian/; }" in nginx
    guardian = nginx.split("location ^~ /network-guardian/ {", 1)[1].split("}", 1)[0]
    assert "proxy_pass http://network-guardian:8897/;" in guardian
    assert "proxy_set_header Cookie $http_cookie;" in guardian
    assert 'proxy_set_header X-User "";' in guardian
    assert 'proxy_set_header X-Role "";' in guardian


def test_reverse_proxy_manager_has_a_dedicated_private_application_route():
    nginx = (ROOT / "deploy" / "nginx" / "nginx.conf").read_text(encoding="utf-8")

    assert "location = /reverse-proxy-manager { return 308 /reverse-proxy-manager/; }" in nginx
    manager = nginx.split("location ^~ /reverse-proxy-manager/ {", 1)[1].split("}", 1)[0]
    assert "proxy_pass http://reverse-proxy-manager:8000;" in manager
    assert "proxy_set_header Cookie $http_cookie;" in manager
    assert 'proxy_set_header X-User "";' in manager
    assert 'proxy_set_header X-Role "";' in manager
    assert "proxy_hide_header Set-Cookie;" in manager


def test_application_routes_use_independent_app_store_icon_namespace():
    nginx = (ROOT / "deploy" / "nginx" / "nginx.conf").read_text(encoding="utf-8")

    assert "location ^~ /loutre-app-store/app-icons/ {" in nginx
    for slug in (
        "loutre-os", "loutre-finance", "smart-inventory", "auto-care",
        "print-pilot", "loutre-home", "network-guardian", "reverse-proxy-manager",
        "loutre-brain", "knowledge-graph", "trip-organizer",
    ):
        assert f"location ^~ /{slug}/ {{" in nginx


def test_last_three_ecosystem_apps_have_private_application_routes():
    nginx = (ROOT / "deploy" / "nginx" / "nginx.conf").read_text(encoding="utf-8")

    for slug, upstream in (
        ("loutre-brain", "http://loutre-brain:8000"),
        ("knowledge-graph", "http://knowledge-graph:8017"),
        ("trip-organizer", "http://trip-organizer:8000"),
    ):
        assert f"location = /{slug} {{ return 308 /{slug}/; }}" in nginx
        route = nginx.split(f"location ^~ /{slug}/ {{", 1)[1].split("}", 1)[0]
        assert f"proxy_pass {upstream};" in route
        assert "proxy_set_header Cookie $http_cookie;" in route
        assert 'proxy_set_header X-User "";' in route
        assert 'proxy_set_header X-Role "";' in route


def test_django_login_cookie_and_fastapi_session_bridge(tmp_path: Path):
    script = r'''import json
import os

import django
django.setup()

from pathlib import Path
from django.contrib.auth import get_user_model
from django.core.management import call_command
from django.test import Client
from django_ui.identity.models import ApplicationGrant
from fastapi.testclient import TestClient
from starlette.websockets import WebSocketDisconnect
from codexpi.config import Settings
from codexpi.django_session import DjangoSessionAuth
from codexpi.main import create_app

call_command("migrate", interactive=False, verbosity=0)
get_user_model().objects.create_superuser("admin", "admin@example.invalid", "a-very-long-test-password")
client = Client(enforce_csrf_checks=True)
page = client.get("/", secure=True, HTTP_HOST="codexpi.example-tailnet.ts.net")
assert page.status_code == 200, page.content
assert "img-src 'self' data: blob:" in page["Content-Security-Policy"]
bootstrap = client.get("/api/bootstrap", secure=True, HTTP_HOST="testserver")
assert bootstrap.status_code == 200, bootstrap.content
csrf = bootstrap.json()["csrf_token"]
login = client.post(
    "/api/auth/login",
    data=json.dumps({"username": "admin", "password": "a-very-long-test-password"}),
    content_type="application/json",
    secure=True,
    HTTP_HOST="testserver",
    HTTP_ORIGIN="https://testserver",
    HTTP_X_CSRF_TOKEN=csrf,
)
assert login.status_code == 200, login.content
ApplicationGrant.objects.create(user=get_user_model().objects.get(username="admin"), application="auto-care", role="admin")
identity = client.get(
    "/api/auth/applications/auto-care",
    secure=True,
    HTTP_HOST="testserver",
)
assert identity.status_code == 200, identity.content
assert identity.json()["permissions"] == ["read", "write", "admin"]
assert client.get(
    "/api/auth/applications/print-pilot",
    secure=True,
    HTTP_HOST="testserver",
).status_code == 403
assert client.get(
    "/api/auth/applications/not-an-app",
    secure=True,
    HTTP_HOST="testserver",
).status_code == 404
cookie = client.cookies["codexpi_session"]
assert cookie["secure"]
assert cookie["httponly"]
assert cookie["samesite"] == "Strict"
session = DjangoSessionAuth().session_user(cookie.value)
assert session is not None
assert session[0]["username"] == "admin"
assert DjangoSessionAuth.csrf_matches(login.json()["csrf_token"], session[1])
audit = (os.path.join(os.environ["CODEXPI_DATA_DIR"], "security.log"))
assert cookie.value not in open(audit, encoding="utf-8").read()
data_dir = Path(os.environ["CODEXPI_DATA_DIR"])
fastapi_app = create_app(Settings(
    base_dir=data_dir,
    data_dir=data_dir,
    database_path=data_dir / "fastapi.sqlite3",
    host="127.0.0.1",
    port=8000,
    cookie_name="codexpi_session",
    cookie_secure=True,
    session_hours=12,
    allowed_roots=(data_dir,),
    forbidden_roots=(),
    codex_bin="definitely-missing-codex",
    frontend_dist=data_dir / "missing-frontend",
    trusted_origins=("https://codexpi.example-tailnet.ts.net",),
    auth_backend="django",
    containerized=True,
))
with TestClient(fastapi_app) as api_client:
    api_client.cookies.set("codexpi_session", cookie.value)
    app_store = api_client.get("/loutre-app-store/")
    assert app_store.status_code == 200, app_store.content
    assert "Loutre App Store" in app_store.text
    try:
        with api_client.websocket_connect(
            "/ws/conversations/does-not-exist",
            headers={"origin": "https://codexpi.example-tailnet.ts.net"},
        ):
            raise AssertionError("Missing conversation socket unexpectedly opened")
    except WebSocketDisconnect as exc:
        assert exc.code == 4404, exc.code
logout = client.post(
    "/api/auth/logout",
    data="{}",
    content_type="application/json",
    secure=True,
    HTTP_HOST="testserver",
    HTTP_ORIGIN="https://testserver",
    HTTP_X_CSRF_TOKEN=login.json()["csrf_token"],
)
assert logout.status_code == 200, logout.content
'''
    env = os.environ.copy()
    env.update(
        {
            "DJANGO_SETTINGS_MODULE": "django_ui.settings",
            "CODEXPI_DATA_DIR": str(tmp_path),
            "CODEXPI_COOKIE_SECURE": "true",
            "CODEXPI_DJANGO_ALLOWED_HOSTS": "testserver,.ts.net",
            "PYTHONPATH": os.pathsep.join((str(ROOT), str(ROOT / "backend"))),
        }
    )
    completed = subprocess.run(
        [sys.executable, "-c", script],
        cwd=ROOT,
        env=env,
        capture_output=True,
        text=True,
        timeout=60,
        check=False,
    )
    assert completed.returncode == 0, completed.stdout + completed.stderr