pyr0ball
bbb4605829
Add blocklist_candidates table and indexes to _SCHEMA in pipeline.py. Add TestSchema tests verifying table existence, column set, and status/hit_count defaults. All 193 tests pass.
43 lines
1.8 KiB
Python
43 lines
1.8 KiB
Python
"""Tests for blocklist service — schema, extraction, candidate management."""
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import pytest
|
|
from pathlib import Path
|
|
|
|
|
|
class TestSchema:
|
|
def test_blocklist_candidates_table_exists(self, tmp_path):
|
|
from app.ingest.pipeline import ensure_schema
|
|
db = tmp_path / "test.db"
|
|
ensure_schema(db)
|
|
conn = sqlite3.connect(str(db))
|
|
tables = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
|
|
assert "blocklist_candidates" in tables
|
|
|
|
def test_blocklist_candidates_columns(self, tmp_path):
|
|
from app.ingest.pipeline import ensure_schema
|
|
db = tmp_path / "test.db"
|
|
ensure_schema(db)
|
|
conn = sqlite3.connect(str(db))
|
|
cols = {r[1] for r in conn.execute("PRAGMA table_info(blocklist_candidates)").fetchall()}
|
|
assert cols >= {
|
|
"id", "domain_or_ip", "source_device_ip", "source_device_name",
|
|
"first_seen", "last_seen", "hit_count", "status", "pushed_at",
|
|
"log_evidence", "matched_rule", "llm_score", "llm_reason",
|
|
}
|
|
|
|
def test_status_default_is_pending(self, tmp_path):
|
|
from app.ingest.pipeline import ensure_schema
|
|
import uuid
|
|
db = tmp_path / "test.db"
|
|
ensure_schema(db)
|
|
conn = sqlite3.connect(str(db))
|
|
conn.execute(
|
|
"INSERT INTO blocklist_candidates (id, domain_or_ip, first_seen, last_seen) VALUES (?, ?, ?, ?)",
|
|
(str(uuid.uuid4()), "samsungads.com", "2026-05-14T00:00:00+00:00", "2026-05-14T00:00:00+00:00"),
|
|
)
|
|
conn.commit()
|
|
row = conn.execute("SELECT status, hit_count FROM blocklist_candidates").fetchone()
|
|
assert row[0] == "pending"
|
|
assert row[1] == 1
|