guarddog-nexus/tests/e2e/conftest.py

"""E2E test fixtures for GuardDog Nexus end-to-end tests."""

import os
import sys
from pathlib import Path

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

# Add project root to path
sys.path.insert(0, str(Path(__file__).parent.parent.parent))

# Set environment for testing
os.environ["DATABASE_PATH"] = ":memory:"
os.environ["NEXUS_URL"] = "http://nexus:8081"
os.environ["LOG_SYSLOG_HOST"] = ""
os.environ["TEMP_DIR"] = "/tmp/guarddog-nexus-e2e"
os.environ["LLM_ENABLED"] = "0"
os.environ["LLM_AUTO_ANALYZE"] = "0"
os.environ["LLM_API_KEY"] = ""

from guarddog_nexus.constants import DEFAULT_ECOSYSTEM, SEVERITY_WARNING  # noqa: E402
from guarddog_nexus.db.engine import Base, get_session  # noqa: E402
from guarddog_nexus.db.models import Finding, Scan, ScanStatus  # noqa: E402
from guarddog_nexus.main import app  # noqa: E402


@pytest_asyncio.fixture
async def e2e_db_engine():
    """Create shared database engine for e2e tests."""
    engine = create_async_engine(
        "sqlite+aiosqlite:///file:e2e_test?mode=memory&cache=shared&uri=true"
    )
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()


@pytest_asyncio.fixture
async def e2e_db_session(e2e_db_engine):
    """Create database session for e2e tests."""
    maker = async_sessionmaker(e2e_db_engine, class_=AsyncSession, expire_on_commit=False)
    async with maker() as session:
        yield session


@pytest_asyncio.fixture
async def e2e_client(e2e_db_engine):
    """Create HTTP client for e2e tests."""
    maker = async_sessionmaker(e2e_db_engine, class_=AsyncSession, expire_on_commit=False)

    async def override_get_session():
        async with maker() as session:
            yield session

    app.dependency_overrides[get_session] = override_get_session

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

    app.dependency_overrides.clear()


@pytest_asyncio.fixture
async def sample_e2e_scan(e2e_db_session):
    """Create a sample scan with findings for e2e tests."""
    scan = Scan(
        package_name="test-e2e-pkg",
        package_version="1.0.0",
        ecosystem=DEFAULT_ECOSYSTEM,
        repository="pypi-proxy",
        nexus_asset_url="http://nexus:8081/repository/pypi-proxy/packages/test-e2e-pkg/1.0.0/test-e2e-pkg-1.0.0.tar.gz",
        sha256="e2e1234567890abcdef",
        status=ScanStatus.COMPLETED.value,
        total_findings=2,
        flagged=True,
    )
    e2e_db_session.add(scan)
    await e2e_db_session.commit()
    await e2e_db_session.refresh(scan)

    # Add findings
    for i, rule in enumerate(["shady-links", "exec-base64"]):
        finding = Finding(
            scan_id=scan.id,
            data={
                "rule": rule,
                "severity": SEVERITY_WARNING,
                "message": f"E2E test finding {i + 1}",
                "location": f"test.py:{i + 1}",
                "code": f"print('test {i + 1}')",
            },
        )
        e2e_db_session.add(finding)
    await e2e_db_session.commit()
    await e2e_db_session.refresh(scan)
    return scan


@pytest.fixture
def e2e_webhook_payload():
    """Create a sample Nexus webhook payload."""
    return {
        "timestamp": "2026-05-11T12:00:00.000+00:00",
        "nodeId": "e2e-test-node",
        "initiator": "e2e-test",
        "action": "UPDATED",
        "repositoryName": "pypi-proxy",
        "asset": {
            "id": "e2e123",
            "assetId": "dGVzdGUyZTFFMjM=",
            "format": "pypi",
            "name": "/packages/e2e-test-pkg/1.0.0/e2e-test-pkg-1.0.0.tar.gz",
            "downloadUrl": "http://nexus:8081/repository/pypi-proxy/packages/e2e-test-pkg/1.0.0/e2e-test-pkg-1.0.0.tar.gz",
        },
    }


@pytest.fixture
def e2e_go_webhook_payload():
    """Create a sample Go webhook payload."""
    return {
        "timestamp": "2026-05-11T12:00:00.000+00:00",
        "nodeId": "e2e-test-node",
        "initiator": "e2e-test",
        "action": "UPDATED",
        "repositoryName": "go-proxy",
        "asset": {
            "id": "e2ego123",
            "assetId": "Z29lMjFFMjM=",
            "format": "go",
            "name": "/packages/github.com/e2e/test-go/@v/v1.0.0.zip",
            "downloadUrl": "http://nexus:8081/repository/go-proxy/github.com/e2e/test-go/@v/v1.0.0.zip",
        },
    }


@pytest.fixture
def e2e_npm_webhook_payload():
    """Create a sample npm webhook payload."""
    return {
        "timestamp": "2026-05-11T12:00:00.000+00:00",
        "nodeId": "e2e-test-node",
        "initiator": "e2e-test",
        "action": "UPDATED",
        "repositoryName": "npm-proxy",
        "asset": {
            "id": "e2enpm123",
            "assetId": "bnBtZTJFRTIz",
            "format": "npm",
            "name": "/packages/e2e-test-npm/-/e2e-test-npm-1.0.0.tgz",
            "downloadUrl": "http://nexus:8081/repository/npm-proxy/e2e-test-npm/-/e2e-test-npm-1.0.0.tgz",
        },
    }