stackpilot/backend/tests/test_review_regressions.py

"""Smaller regressions found in the 0.67 review, one test each.

* NVIDIA cards were never detected: the image has no nvidia-smi and the DRI
  scan skips NVIDIA on purpose. The driver's /proc entries are read instead.
* A container's inspect view handed its environment — credentials, by
  convention — to the read-only role, which does not get the stack's .env.
* Git branch names and URLs go to git's command line; a leading "-" made them
  options.
* "Scan all" ran for longer than the proxy waits; it now runs in the
  background and reports through /scan-status.
"""
from __future__ import annotations

import asyncio

import pytest


def test_nvidia_gpus_are_read_from_the_driver_procfs(tmp_path, monkeypatch):
    from config import settings
    from services import gpu_service

    gpus = tmp_path / "driver" / "nvidia" / "gpus"
    (gpus / "0000:01:00.0").mkdir(parents=True)
    (gpus / "0000:01:00.0" / "information").write_text(
        "Model: \t\t NVIDIA GeForce RTX 3060\n"
        "IRQ:   \t\t 150\n"
        "GPU UUID: \t GPU-1234-abcd\n"
        "Device Minor: \t 0\n"
    )
    (tmp_path / "driver" / "nvidia" / "version").write_text(
        "NVRM version: NVIDIA UNIX x86_64 Kernel Module  550.120  Fri Sep 13 2024\n"
    )
    monkeypatch.setattr(settings, "HOST_PROC_PATH", str(tmp_path))
    monkeypatch.setattr(gpu_service, "_detect_nvidia", lambda: [])
    monkeypatch.setattr(gpu_service, "_detect_dri", lambda: [])

    (gpu,) = gpu_service.detect_gpus()
    assert gpu.vendor == "nvidia"
    assert gpu.name == "NVIDIA GeForce RTX 3060"
    assert gpu.uuid == "GPU-1234-abcd"
    assert gpu.index == 0
    assert gpu.driver == "550.120"


@pytest.fixture
def fake_container(monkeypatch):
    from services import container_service

    monkeypatch.setattr(
        container_service,
        "inspect_container",
        lambda container_id: {"id": container_id, "env": ["DB_PASSWORD=hunter2", "NOVALUE"]},
    )


def test_container_env_values_are_hidden_from_the_read_only_role(as_user, fake_container):
    env = as_user.get("/api/containers/abc").json()["env"]
    assert env == ["DB_PASSWORD=••••••", "NOVALUE"]


def test_admins_still_see_the_environment(as_admin, fake_container):
    assert as_admin.get("/api/containers/abc").json()["env"][0] == "DB_PASSWORD=hunter2"


def test_a_missing_container_is_a_404(as_admin, monkeypatch):
    from docker_client import DockerError
    from services import container_service

    def missing(container_id):
        raise DockerError("not_found", "No such container")

    monkeypatch.setattr(container_service, "inspect_container", missing)
    assert as_admin.get("/api/containers/nope").status_code == 404


@pytest.mark.parametrize("branch", ["-x", "--upload-pack=sh", "a..b", "x.lock", ""])
def test_branch_names_that_git_would_read_as_options_are_refused(branch):
    from services import git_service

    with pytest.raises(git_service.GitError):
        git_service.validate_branch(branch)


@pytest.mark.parametrize("url", ["-oProxyCommand=x", "ext::sh -c id", "fd::3", ""])
def test_repository_urls_that_run_commands_are_refused(url):
    from services import git_service

    with pytest.raises(git_service.GitError):
        git_service.validate_url(url)


def test_ordinary_branches_and_urls_pass():
    from services import git_service

    assert git_service.validate_branch("release/1.2") == "release/1.2"
    assert git_service.validate_url("git@forge:me/stack.git") == "git@forge:me/stack.git"


def test_scan_all_in_the_background_is_active_before_the_request_returns(monkeypatch, db):
    from services import scan_service

    release = asyncio.Event()

    async def slow_scan(session, image):
        await release.wait()

        class Row:
            error = None

        return Row()

    monkeypatch.setattr(scan_service, "scan", slow_scan)

    async def scenario():
        status = scan_service.start_sweep(["a", "b"])
        assert status["active"] is True and status["total"] == 2
        with pytest.raises(scan_service.ScanError):
            scan_service.start_sweep(["c"])
        release.set()
        for _ in range(100):
            if not scan_service.sweep_status()["active"]:
                break
            await asyncio.sleep(0.01)
        final = scan_service.sweep_status()
        assert final["active"] is False
        assert final["scanned"] == 2 and final["failed"] == 0

    asyncio.run(scenario())


def test_setup_refuses_an_empty_account(client, db, monkeypatch):
    import auth as auth_mod

    monkeypatch.setattr(auth_mod, "users_exist", lambda session: False)
    response = client.post("/api/auth/setup", json={"username": " ", "password": ""})
    assert response.status_code == 400