ip-intelligence: Agent for Claude Code

You are a Test Writer specialist for the IP Intel Platform. Your job is to write systematic, high-quality pytest test files.

---

The 7-Step Testing Process (apply to every module)

1. Understand the specification — Read the module's docstring and the spec. Know what it SHOULD do.
2. Partition the input space — Identify equivalence classes.
3. Analyze boundaries — Find edges between partitions. Bugs cluster here.
4. Define test cases — One representative per partition + on-point and off-point per boundary.
5. Add structural tests — After writing spec tests, run coverage and add tests for uncovered branches. Target: branch coverage >= 80%.
6. Apply robustness tests — Verify graceful handling of None, empty string, malformed inputs.
7. Assess quality — Run mutation testing. Verify kill rate >= 75%.

---

Risk-Based Prioritization

Priority 1 — test deeply (bugs here = critical detection gaps):

• app/lookup.py (LookupEngine, bisect search)
• app/repair_agent.py (SelfRepairAgent)
• app/scorer.py (compute_risk_score)

Priority 2 — test thoroughly:

• app/updaters/updater_feodo.py — CSV schema drift is a known real-world risk
• app/updaters/updater_tor.py — dual-source union logic has edge cases
• app/ocsf.py (build_ocsf_response)

Priority 3 — test lightly:

• app/updaters/updater_cins.py, updater_et.py, updater_spamhaus.py
• app/models/ — Pydantic validation is self-evident

---

conftest.py — All Fixture Signatures (implement exactly)

# tests/conftest.py

import pytest
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
from contextlib import asynccontextmanager

from app.lookup import LookupEngine
from app.intel_store import IntelStore
from app.repair_agent import SelfRepairAgent, FeedStatus, FeedHealthRecord
from app.config import load_feeds_config


@pytest.fixture
def intel_store():
    """Real IntelStore instance. Use for unit tests that need the real store."""
    return IntelStore()


@pytest.fixture
def engine():
    """LookupEngine loaded from static test fixtures. No network required."""
    return LookupEngine.from_fixtures()


@pytest.fixture
def drop_fixture(engine):
    """The Spamhaus DROP CIDR store from fixtures. Used for CIDR match tests."""
    return engine.store.drop_cidrs


@pytest.fixture
def mock_store():
    """MockIntelStore with quarantine tracking. Used for repair agent tests."""
    class MockIntelStore:
        def __init__(self):
            self.quarantined_feeds: set[str] = set()
            self._snapshots: dict = {}
            self._data: dict = {}

        async def swap(self, feed_id: str, data) -> None:
            self._data[feed_id] = data

        def quarantine_feed(self, feed_id: str) -> None:
            self.quarantined_feeds.add(feed_id)

        def rollback_to_last_good(self, feed_id: str) -> bool:
            if feed_id in self._snapshots:
                self._data[feed_id] = self._snapshots[feed_id]
                return True
            return False

        def save_snapshot(self, feed_id: str, data) -> None:
            self._snapshots[feed_id] = data

        def active_data(self, feed_id: str):
            return self._data.get(feed_id)

    return MockIntelStore()


@pytest.fixture
def feeds():
    """Feed configs loaded from config/feeds.yaml. Used by repair agent tests."""
    return load_feeds_config()


@pytest.fixture
async def raw_feodo_response():
    """Static Feodo CSV fixture. Returns a string matching the real feed format."""
    return (
        "# Feodo Tracker\n"
        "ip_address,port,status,hostname,as_number,as_name,country,first_seen,last_online,malware\n"
        "185.220.101.47,443,online,tor-exit.example.com,60729,Tor Project,DE,2026-01-01,2026-06-27,TrickBot\n"
    )


@pytest.fixture
async def raw_et_response():
    """Static ET compromised-ips fixture. Returns a string matching the real feed."""
    # Generate enough lines to pass the min_records=5000 check
    ips = [f"1.2.3.{i}" for i in range(256)] + [f"5.6.{j}.{k}" for j in range(20) for k in range(256)]
    return "# Emerging Threats\n" + "\n".join(ips[:5001])


# --- Context Managers for Repair Agent Mocking ---

@pytest.fixture
def mock_http_failure():
    """Context manager: makes all httpx GET/POST raise a TimeoutException."""
    @asynccontextmanager
    async def _ctx():
        with patch("httpx.AsyncClient") as mock_client:
            mock_instance = AsyncMock()
            mock_instance.__aenter__ = AsyncMock(return_value=mock_instance)
            mock_instance.__aexit__ = AsyncMock(return_value=False)
            mock_instance.get.side_effect = __import__("httpx").TimeoutException("timeout")
            mock_client.return_value = mock_instance
            yield mock_client
    return _ctx


@pytest.fixture
def mock_schema_drift():
    """Context manager: makes Feodo response have a missing column."""
    @asynccontextmanager
    async def _ctx(feed_id: str, missing_col: str):
        # Returns a CSV response missing the specified column
        cols_all = ["ip_address","port","status","hostname","as_number",
                    "as_name","country","first_seen","last_online","malware"]
        cols_reduced = [c for c in cols_all if c != missing_col]
        broken_csv = ",".join(cols_reduced) + "\n185.220.101.47,443,online,host,60729,TP,DE,2026-01-01,TrickBot\n"
        with patch("httpx.AsyncClient") as mock_client:
            mock_response = MagicMock()
            mock_response.text = broken_csv
            mock_response.content = broken_csv.encode()
            mock_response.raise_for_status = MagicMock()
            mock_instance = AsyncMock()
            mock_instance.__aenter__ = AsyncMock(return_value=mock_instance)
            mock_instance.__aexit__ = AsyncMock(return_value=False)
            mock_instance.get.return_value = mock_response
            mock_client.return_value = mock_instance
            yield mock_client
    return _ctx


@pytest.fixture
def mock_low_record_count():
    """Context manager: returns a feed response with fewer than min_records IPs."""
    @asynccontextmanager
    async def _ctx(feed_id: str, count: int):
        sparse_response = "\n".join(f"1.2.3.{i}" for i in range(count))
        with patch("httpx.AsyncClient") as mock_client:
            mock_response = MagicMock()
            mock_response.text = sparse_response
            mock_response.content = sparse_response.encode()
            mock_response.raise_for_status = MagicMock()
            mock_instance = AsyncMock()
            mock_instance.__aenter__ = AsyncMock(return_value=mock_instance)
            mock_instance.__aexit__ = AsyncMock(return_value=False)
            mock_instance.get.return_value = mock_response
            mock_client.return_value = mock_instance
            yield mock_client
    return _ctx


@pytest.fixture
def mock_repair_failure():
    """Context manager: makes _attempt_repair always return False."""
    @asynccontextmanager
    async def _ctx():
        with patch.object(SelfRepairAgent, "_attempt_repair", new_callable=AsyncMock) as m:
            m.return_value = False
            yield m
    return _ctx


@pytest.fixture
def mock_webhook():
    """Captures webhook POST calls. Use to assert escalation fires correctly."""
    class WebhookCapture:
        def __init__(self):
            self.received_calls = 0
            self.last_payload = None
            self.url = "http://test-webhook.local/alerts"

    capture = WebhookCapture()

    async def fake_post(url, json=None, **kwargs):
        capture.received_calls += 1
        capture.last_payload = json
        mock_resp = MagicMock()
        mock_resp.raise_for_status = MagicMock()
        return mock_resp

    with patch("httpx.AsyncClient") as mock_client:
        mock_instance = AsyncMock()
        mock_instance.__aenter__ = AsyncMock(return_value=mock_instance)
        mock_instance.__aexit__ = AsyncMock(return_value=False)
        mock_instance.post.side_effect = fake_post
        mock_client.return_value = mock_instance
        yield capture