guarddog-nexus/tests/conftest.py

"""Test fixtures for guarddog-nexus."""

import os
import sys

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

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

os.environ["DATABASE_PATH"] = ":memory:"
os.environ["NEXUS_URL"] = "http://nexus:8081"
os.environ["NEXUS_USERNAME"] = "admin"
os.environ["NEXUS_PASSWORD"] = "admin123"
os.environ["LOG_SYSLOG_HOST"] = ""
os.environ["TEMP_DIR"] = "/tmp/guarddog-nexus-test"

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


@pytest_asyncio.fixture
async def db_engine():
    engine = create_async_engine(
        "sqlite+aiosqlite:///file:guarddog_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 db_session(db_engine):
    maker = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False)
    async with maker() as session:
        yield session


@pytest_asyncio.fixture
async def client(db_engine):
    maker = async_sessionmaker(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_flagged_scan(db_session):
    scan = Scan(
        package_name="test-pkg",
        package_version="1.0",
        ecosystem=DEFAULT_ECOSYSTEM,
        repository="pypi-proxy",
        nexus_asset_url="http://nexus:8081/repository/pypi-proxy/packages/test-pkg/1.0/test-pkg-1.0.tar.gz",
        sha256="abc123",
        status=ScanStatus.COMPLETED.value,
        total_findings=1,
        flagged=True,
    )
    db_session.add(scan)
    await db_session.commit()
    await db_session.refresh(scan)

    finding = Finding(
        scan_id=scan.id,
        data={
            "rule": "test_rule",
            "severity": SEVERITY_WARNING,
            "message": "Test finding",
            "location": "test.py:1",
            "code": "print('test')",
        },
    )
    db_session.add(finding)
    await db_session.commit()
    await db_session.refresh(scan)
    return scan


@pytest.fixture
def sample_nexus_webhook():
    return {
        "timestamp": "2026-05-09T12:00:00.000+00:00",
        "nodeId": "test-node",
        "initiator": "admin",
        "action": "UPDATED",
        "repositoryName": "pypi-proxy",
        "asset": {
            "id": "abc123",
            "assetId": "dGVzdA==",
            "format": "pypi",
            "name": "/packages/requests/2.31.0/requests-2.31.0.tar.gz",
            "downloadUrl": "http://nexus:8081/repository/pypi-proxy/packages/requests/2.31.0/requests-2.31.0.tar.gz",
        },
    }


@pytest.fixture
def sample_nexus_component_webhook():
    return {
        "timestamp": "2026-05-09T12:00:00.000+00:00",
        "nodeId": "test-node",
        "initiator": "admin",
        "action": "UPDATED",
        "repositoryName": "pypi-proxy",
        "component": {
            "id": "comp1",
            "componentId": "dGVzdDI=",
            "format": "pypi",
            "name": "requests",
            "version": "2.31.0",
        },
    }


@pytest.fixture
def guarddog_output_clean():
    return {
        "package": "safe-pkg",
        "issues": 0,
        "errors": {},
        "results": {
            "obfuscation": {},
            "exec-base64": {},
            "shady-links": {},
            "typosquatting": None,
            "empty_information": None,
        },
    }


@pytest.fixture
def guarddog_output_flagged():
    return {
        "package": "bad-pkg",
        "issues": 3,
        "errors": {},
        "results": {
            "shady-links": [
                {
                    "message": "Package contains URL to suspicious domain",
                    "location": "setup.py:15",
                    "code": "url = 'http://evil.com'",
                }
            ],
            "exec-base64": [
                {
                    "message": "Base64-encoded code execution detected",
                    "location": "core.py:42",
                    "code": "exec(base64.b64decode(...))",
                }
            ],
            "empty_information": "Package description is empty",
            "obfuscation": {},
            "typosquatting": None,
        },
    }


@pytest.fixture
def guarddog_normalized_flagged():
    return {
        "findings": [
            {
                "rule": "shady-links",
                "severity": SEVERITY_WARNING,
                "message": "Package contains URL to suspicious domain",
                "location": "setup.py:15",
                "code": "url = 'http://evil.com'",
            },
            {
                "rule": "exec-base64",
                "severity": SEVERITY_WARNING,
                "message": "Base64-encoded code execution detected",
                "location": "core.py:42",
                "code": "exec(base64.b64decode(...))",
            },
            {
                "rule": "empty_information",
                "severity": SEVERITY_WARNING,
                "message": "Package description is empty",
                "location": "",
                "code": "",
            },
        ],
        "errors": [],
    }


@pytest.fixture
def guarddog_normalized_clean():
    return {
        "findings": [],
        "errors": [],
    }