roboco/tests/unit/runtime/test_spawn_agent_path_segment.py

"""spawn_agent rejects an unsafe agent_id before any filesystem op.

``agent_id`` flows from request-facing call sites into path building (log
dirs, settings files, container names). A traversal-shaped id (``../evil``,
``a/b``, NUL, empty) must be rejected at the spawn chokepoint — before it
reaches any path join — via ``_safe_agent_path_segment``. The guard is the
first statement in ``spawn_agent`` and only consults a pure static method,
so a bare (un-initialized) orchestrator suffices (same shape as the
human-role guard tests).
"""

from __future__ import annotations

from typing import Any
from unittest.mock import MagicMock

import pytest
from roboco.runtime.orchestrator import AgentOrchestrator, AgentReadinessError


def _orch() -> Any:
    return object.__new__(AgentOrchestrator)


@pytest.mark.parametrize(
    "bad_id",
    ["../evil", "a/b", "a\\b", ".", "..", "", "be\x00dev"],
)
@pytest.mark.asyncio
async def test_spawn_agent_rejects_unsafe_path_id(bad_id: str) -> None:
    """A traversal-shaped agent_id must raise ValueError before any fs op."""
    orch = _orch()
    with pytest.raises(ValueError, match="unsafe agent id"):
        await orch.spawn_agent(bad_id)


@pytest.mark.asyncio
async def test_spawn_agent_accepts_normal_slug_reaches_later_gate() -> None:
    """A normal slug must NOT trip the path guard — it reaches the downstream
    readiness gate (a *different* refusal), proving the guard doesn't
    over-block the fleet."""
    orch = _orch()

    async def _ready(_aid: str, _tid: str | None) -> str | None:
        return "stubbed-not-ready"

    orch._readiness_gate = _ready
    # Stub any later attribute touch so a bare orch survives past the guard.
    orch._instances = MagicMock()

    with pytest.raises(AgentReadinessError) as exc_info:
        await orch.spawn_agent("be-dev-1", task_id="t-1")

    assert "unsafe agent id" not in str(exc_info.value)
    assert "human-only role" not in str(exc_info.value)