Testing

vega-core tests live under tests/ and are split into unit tests, integration tests, and a manual debug runner.

Test structure

tests/
  conftest.py                    # shared pytest fixtures

  unit/
    test_agent_runner.py         # AgentRunner adapter selection, timeout, cancel
    test_artifacts.py            # LocalArtifactStore read/write, path safety
    test_base_stage.py           # BaseStage lifecycle, plugin hooks, artifact validation
    test_codex_adapter.py        # CodexAdapter subprocess wiring, cancel, schema
    test_core_contracts.py       # CoreResult, StageError, request/result dataclasses
    test_imports_and_stages.py   # import smoke tests for all stage classes
    test_lifecycle_cancellation.py  # CancellationManager checkpoints, SIGTERM flow
    test_plan_stage.py           # PlanStage agent call, artifact output, events
    test_runtime_config.py       # RuntimeConfig.from_env() with mocked env
    test_scan_strategies.py      # Threat model and audit strategy selection
    test_schema_overrides.py     # apply_schema_enum_overrides validation rules
    test_source_scope.py         # SourceView, path normalization, include_paths
    test_stage_orchestrator.py   # StageOrchestrator run_scan, plan_repo flows
    test_verify_patch_stages.py  # VerifyStage and PatchStage with fake agent

  integration/
    test_plan_repo.py            # end-to-end plan_repo with real source dir
    (test_run_scan.py)           # end-to-end run_scan (planned)
    (test_verify_patch.py)       # end-to-end verify_and_patch (planned)

  manual/
    e2e_scan_debug.py            # local debug runner; calls plan_repo and run_scan

Test infrastructure

FakeAgentAdapter

The primary mock for agent calls:

@dataclass
class FakeAgentAdapter:
    name: str = "fake"
    responses: dict[str, Mapping[str, Any]] = field(default_factory=dict)
    errors: dict[str, str] = field(default_factory=dict)
    delay_seconds: float = 0

Pre-populate responses with {task_purpose: output_dict}. Stages that run agent tasks will receive those outputs without spawning a subprocess. Purpose strings follow the stage naming convention ("plan", "threat_model.codex_direct", "audit.default", "verify.finding", "patch.generate", etc.).

fake = FakeAgentAdapter(
    responses={
        "plan": {
            "audit_priority": [{"module_name": "auth", "module_paths": ["src/auth/"]}],
            "modules": {"auth": {"components": [
                {"component_name": "login", "scope": ["src/auth/login.py"]}
            ]}}
        },
        "threat_model.codex_direct": {"content": "# Threat Model\n\n..."},
        "audit.default": {"findings": [...]},
    }
)
runner = AgentRunner((fake,))

testing/fixtures.py

Shared pytest fixtures for common test setups:

• tmp_source_dir — temporary directory with minimal source files
• plan_artifact_ref — pre-built ArtifactRef for a vega.plan artifact
• fake_agent_runner — AgentRunner with a pre-configured FakeAgentAdapter
• local_artifact_store — LocalArtifactStore pointing to a temp directory
• event_collector — event sink that collects CoreEvent values into a list

testing/fakes.py

Additional fake implementations used by tests:

• FakeCancellationToken — toggleable cancellation signal
• FakeSharedTriageService — returns identity deduplication (no-op triage)
• FakePluginResolver — always resolves to DefaultPlugin

Wiring tests with configure_core_services

from api import configure_core_services, reset_core_services, plan_repo
from api.requests import PlanRequest
from agents.adapters.fake import FakeAgentAdapter
from agents.runner import AgentRunner

def test_plan_repo(tmp_path):
    events = []
    fake = FakeAgentAdapter(responses={
        "plan": {
            "audit_priority": [{"module_name": "auth", "module_paths": ["src/"]}],
            "modules": {"auth": {"components": [{"component_name": "handler", "scope": ["src/"]}]}},
        }
    })
    configure_core_services(
        event_bus=EventBus(sink=events.append),
        artifact_store=LocalArtifactStore(root=tmp_path),
        cancellation=CancellationManager(),
        agent_runner=AgentRunner((fake,)),
        plugin_resolver=FakePluginResolver(),
        triage=FakeSharedTriageService(),
    )
    try:
        result = plan_repo(PlanRequest(
            repo_id="test_repo",
            source_snapshot=ArtifactRef(uri=f"file://{tmp_path}", kind="source_snapshot", content_type="application/json"),
        ))
        assert result.status.value == "completed"
        assert "plan" in result.artifacts
    finally:
        reset_core_services()

Running tests

cd /path/to/vega-core
python -m pytest tests/unit/ -v
python -m pytest tests/integration/ -v
python -m pytest tests/ -v        # all tests

Run a single test file:

python -m pytest tests/unit/test_plan_stage.py -v

Manual debug runner

tests/manual/e2e_scan_debug.py runs a real plan + scan against a local source directory. It requires a real Codex binary and API key:

export OPENAI_API_KEY=sk-...
export VEGA_CODEX_BIN=codex
export VEGA_ARTIFACT_STORE=local
export VEGA_ARTIFACT_ROOT=./debug_artifacts

python tests/manual/e2e_scan_debug.py --repo-path /path/to/source

It calls plan_repo then run_scan and prints events and the final CoreResult to stdout.

What tests cover

Area	Coverage
BaseStage lifecycle	Plugin hooks, cancellation checkpoints, artifact validation, error mapping
AgentRunner	Adapter selection, can_run ordering, timeout enforcement, cancel
CodexAdapter	Subprocess wiring, --json flag, SIGTERM cancel path, schema pass-through
PlanStage	Agent output parsing, artifact write, domain events
ThreatModelStage	Cache fingerprint validation, static threat model hook, strategies
AuditStage	Parallel workers, resume logic, finding parsing, schema overrides
VerifyStage	Per-bug verification, threat model simplification, parallel per-bug agents, evidence artifact
PatchStage	Per-finding pipeline, minimality gate, recheck, integration test
PluginSelectionService	Three-tier selection, events, fallback
StageOrchestrator	Workflow wiring, plugin selection, strategy selection, result normalization
CancellationManager	checkpoint raises StageCancelled, cancel_active_agent_calls
LocalArtifactStore	Read/write, path safety, exists
apply_schema_enum_overrides	Replace/extend modes, validation rules
SourceView	glob, walk, include_paths scoping
RuntimeConfig.from_env	All env var mappings