`tests/test_cycle5_search.py`

106 lines
"""Cycle 4 tests — SearchService: masked results + REAL rate limiting + fail-closed projection.

Run:  python -m tests.test_cycle5_search

Proves (deterministic, no sleeps — the limiter clock is injected):
  1. search() returns masked, projection-safe rows (no raw PII).
  2. rate limiting is REAL — N allowed per window, N+1 raises RateLimited; budget refills
     after the window passes. (Closes Security 1.3, the last open High.)
  3. empty/punctuation queries raise EmptyQuery.
  4. rate-limit budgets are per-client-key (one abuser can't exhaust another's budget).
"""
from __future__ import annotations

import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from services.common.rate_limit import SlidingWindowRateLimiter
from services.common.sqlite_repo import FileObjectStore, SqliteRepository
from services.ingestion.ingest import FeedDefinition, ingest_authorized_feed
from services.search.service import EmptyQuery, RateLimited, SearchService

SAMPLE = Path(__file__).resolve().parents[1] / "data" / "sample" / "sample_state_feed.csv"


def ok(msg: str) -> None:
    print(f"  ✓ {msg}")


class FakeClock:
    def __init__(self) -> None:
        self.t = 0.0

    def __call__(self) -> float:
        return self.t


def _populated_repo(tmp: Path) -> SqliteRepository:
    store = FileObjectStore(tmp / "obj")
    (tmp / "obj").mkdir(parents=True, exist_ok=True)
    store.write_bytes("incoming/s.csv", SAMPLE.read_bytes())
    repo = SqliteRepository(str(tmp / "p.db"))
    ingest_authorized_feed(FeedDefinition("SAMPLE", "incoming/s.csv"), store, repo)
    return repo


def main() -> int:
    tmp = Path(tempfile.mkdtemp(prefix="upp-cycle5-"))
    repo = _populated_repo(tmp)
    clock = FakeClock()
    limiter = SlidingWindowRateLimiter(max_requests=3, window_seconds=60, clock=clock)
    svc = SearchService(repository=repo, limiter=limiter, max_results=20)

    print("1) masked, projection-safe results")
    rows = svc.search("Oneil", client_key="ip:10.0.0.1")
    assert rows, "expected a hit"
    assert all("owner_name_masked" in r and "owner_name_raw" not in r for r in rows), rows
    assert "ONEIL" not in (rows[0]["owner_name_masked"] or "").upper(), rows[0]
    ok(f"hit {rows[0]['owner_name_masked']} | {rows[0]['amount_band']} (no raw PII)")

    print("2) REAL rate limiting (3 per 60s window, deterministic clock)")
    limiter.reset()
    k = "ip:1.2.3.4"
    for i in range(3):
        svc.search("Oneil", client_key=k)          # 3 allowed at t=0
    raised = False
    try:
        svc.search("Oneil", client_key=k)          # 4th within window -> blocked
    except RateLimited as e:
        raised = True
        assert e.retry_after > 0, e.retry_after
    assert raised, "4th request in window should have been rate-limited"
    ok("4th request in window -> RateLimited (retry_after > 0)")

    clock.t = 61.0                                   # window elapsed
    rows_after = svc.search("Oneil", client_key=k)   # budget refilled
    assert rows_after, "should be allowed again after the window"
    ok("after window elapses -> allowed again")

    print("3) empty / punctuation queries rejected")
    for bad in ("", "   ", "!!!"):
        limiter.reset()
        try:
            svc.search(bad, client_key="ip:fresh")
            raise AssertionError(f"empty query {bad!r} should raise EmptyQuery")
        except EmptyQuery:
            pass
    ok("empty/punctuation queries -> EmptyQuery")

    print("4) per-client-key budgets are independent")
    limiter.reset()
    for _ in range(3):
        svc.search("Oneil", client_key="ip:aaa")     # exhaust aaa
    # a DIFFERENT client is unaffected
    assert svc.search("Oneil", client_key="ip:bbb"), "bbb should have its own budget"
    ok("one client's exhausted budget does not block another")

    print("\nALL CYCLE-4 SEARCH-SERVICE ASSERTIONS PASSED ✅")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())