test_finding_query_runtime.py

"""Tests for finding query runtime orchestration."""

from __future__ import annotations

from security_scanner.core.finding.model import Disposition, Finding, Verdict
from security_scanner.runtime.finding_query import (
    FindingQueryRequest,
    read_findings,
)
from security_scanner.storage.adapters.nosql_db.transport import DynamoDbCompatibleConfig


def _finding(line_start: int, verdict: str) -> Finding:
    return Finding.create(
        repo_full_name="fake-org/fake-repo",
        rule_id="aws-access-key-id",
        file_path="config/settings.py",
        line_start=line_start,
        raw_secret="AKIAFAKEEXAMPLE000000",
        source_tool="gitleaks",
        scan_run_id="scan_abc12345",
        rule_pack_version="secret-rules-0.1.0",
        triage_verdict=verdict,
    )


def test_read_findings_reads_all_when_scan_run_id_absent():
    """Query runtime reads the full store when no scan-run filter is requested."""
    findings = [object()]

    class FakeReader:
        def __init__(self) -> None:
            self.calls: list[tuple[str, str | None]] = []

        def read_all(self):
            self.calls.append(("read_all", None))
            return findings

        def read_for_scan_run(self, scan_run_id):
            self.calls.append(("read_for_scan_run", scan_run_id))
            return []

    reader = FakeReader()

    result = read_findings(
        FindingQueryRequest(storage_backend="jsonl", jsonl_path="findings.jsonl"),
        store=reader,
    )

    assert result == findings
    assert reader.calls == [("read_all", None)]


def test_read_findings_uses_scan_run_filter_when_present():
    """Query runtime delegates scan-run filtering to the storage reader seam."""
    findings = [object()]

    class FakeReader:
        def __init__(self) -> None:
            self.scan_run_ids: list[str] = []

        def read_all(self):
            raise AssertionError("scan-run query should not call read_all")

        def read_for_scan_run(self, scan_run_id):
            self.scan_run_ids.append(scan_run_id)
            return findings

    reader = FakeReader()

    result = read_findings(
        FindingQueryRequest(
            storage_backend="dynamodb",
            scan_run_id="scan_test0001",
            dynamodb_config=DynamoDbCompatibleConfig(table_name="SecurityScannerLocal"),
        ),
        store=reader,
    )

    assert result == findings
    assert reader.scan_run_ids == ["scan_test0001"]


def test_read_findings_builds_jsonl_reader_from_factory():
    """JSONL query requests pass the findings path to the storage factory."""
    findings = [object()]
    calls = []

    class FakeReader:
        def read_all(self):
            return findings

        def read_for_scan_run(self, scan_run_id):
            raise AssertionError(scan_run_id)

    def fake_store_factory(backend, **kwargs):
        calls.append((backend, kwargs))
        return FakeReader()

    result = read_findings(
        FindingQueryRequest(storage_backend="jsonl", jsonl_path="custom.jsonl"),
        store_factory=fake_store_factory,
    )

    assert result == findings
    assert calls == [("jsonl", {"jsonl_path": "custom.jsonl"})]


def test_read_findings_builds_nosql_reader_from_factory():
    """DynamoDB-compatible query requests pass explicit config to the factory."""
    findings = [object()]
    config = DynamoDbCompatibleConfig(table_name="SecurityScannerLocal")
    calls = []

    class FakeReader:
        def read_all(self):
            return findings

        def read_for_scan_run(self, scan_run_id):
            raise AssertionError(scan_run_id)

    def fake_store_factory(backend, **kwargs):
        calls.append((backend, kwargs))
        return FakeReader()

    result = read_findings(
        FindingQueryRequest(storage_backend="dynamodb", dynamodb_config=config),
        store_factory=fake_store_factory,
    )

    assert result == findings
    assert calls == [("dynamodb", {"dynamodb_config": config})]


def test_read_findings_normalizes_jsonl_backend_alias_for_factory():
    findings = [object()]
    calls = []

    class FakeReader:
        def read_all(self):
            return findings

        def read_for_scan_run(self, scan_run_id):
            raise AssertionError(scan_run_id)

    def fake_store_factory(backend, **kwargs):
        calls.append((backend, kwargs))
        return FakeReader()

    result = read_findings(
        FindingQueryRequest(storage_backend="JSONL", jsonl_path="custom.jsonl"),
        store_factory=fake_store_factory,
    )

    assert result == findings
    assert calls == [("jsonl", {"jsonl_path": "custom.jsonl"})]


def test_read_findings_normalizes_nosql_backend_alias_for_factory():
    findings = [object()]
    calls = []

    class FakeReader:
        def read_all(self):
            return findings

        def read_for_scan_run(self, scan_run_id):
            raise AssertionError(scan_run_id)

    def fake_store_factory(backend, **kwargs):
        calls.append((backend, kwargs))
        return FakeReader()

    result = read_findings(
        FindingQueryRequest(storage_backend="dynamodb_compatible"),
        store_factory=fake_store_factory,
    )

    assert result == findings
    assert calls == [("dynamodb-compatible", {"dynamodb_config": None})]


def test_read_findings_filters_by_disposition_when_requested():
    """The read-API seam slices findings by disposition for the dashboard (FR-11)."""
    unreviewed = _finding(1, Verdict.NEEDS_REVIEW.value)
    verified = _finding(2, Verdict.TRUE_POSITIVE.value)
    false_positive = _finding(3, Verdict.FALSE_POSITIVE.value)

    class FakeReader:
        def read_all(self):
            return [unreviewed, verified, false_positive]

        def read_for_scan_run(self, scan_run_id):
            raise AssertionError(scan_run_id)

    result = read_findings(
        FindingQueryRequest(
            storage_backend="jsonl",
            jsonl_path="findings.jsonl",
            dispositions=[Disposition.UNREVIEWED.value, Disposition.VERIFIED.value],
        ),
        store=FakeReader(),
    )

    assert result == [unreviewed, verified]


def test_read_findings_disposition_filter_applies_to_scan_run_reads():
    """Disposition filtering composes with the scan-run filter."""
    verified = _finding(2, Verdict.TRUE_POSITIVE.value)
    false_positive = _finding(3, Verdict.FALSE_POSITIVE.value)

    class FakeReader:
        def read_all(self):
            raise AssertionError("scan-run query should not call read_all")

        def read_for_scan_run(self, scan_run_id):
            return [verified, false_positive]

    result = read_findings(
        FindingQueryRequest(
            storage_backend="dynamodb",
            scan_run_id="scan_abc12345",
            dynamodb_config=DynamoDbCompatibleConfig(table_name="SecurityScannerLocal"),
            dispositions=[Disposition.FALSE_POSITIVE.value],
        ),
        store=FakeReader(),
    )

    assert result == [false_positive]