128 lines
6.0 KiB
Python
128 lines
6.0 KiB
Python
import json
|
|
import sqlite3
|
|
from contextlib import contextmanager
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from app.crawler import ExtractedEvent, external_key
|
|
|
|
|
|
def insert_source(db, strategy_json='{}'):
|
|
cursor = db.execute("""INSERT INTO sources(name,url,venue_name,venue_address,timezone,strategy_json,item_selector,title_selector,date_selector)
|
|
VALUES('Test','https://example.org/events','Venue','1 Test St','Australia/Sydney',?,'.event','h2','time')""", (strategy_json,))
|
|
return cursor.lastrowid
|
|
|
|
|
|
def test_fresh_database_and_idempotent_initialization(temp_db):
|
|
import app.db as module
|
|
module.init_db()
|
|
with module.connect() as db:
|
|
tables = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table'")}
|
|
assert {"sources", "events", "geocode_cache"} <= tables
|
|
assert "strategy_json" in {row["name"] for row in db.execute("PRAGMA table_info(sources)")}
|
|
assert "hidden" in {row["name"] for row in db.execute("PRAGMA table_info(events)")}
|
|
|
|
|
|
def test_source_delete_cascades_events(temp_db):
|
|
import app.db as module
|
|
with module.connect() as db:
|
|
source_id = insert_source(db)
|
|
db.execute("INSERT INTO events(source_id,external_key,title,starts_at,venue_name,address,event_url) VALUES(?,?,?,?,?,?,?)", (source_id, "key", "One", "2026-09-07T10:00:00+00:00", "Venue", "Address", "https://example.org/one"))
|
|
db.execute("DELETE FROM sources WHERE id=?", (source_id,))
|
|
assert db.execute("SELECT count(*) FROM events").fetchone()[0] == 0
|
|
|
|
|
|
def test_legacy_empty_strategy_is_migrated(temp_db):
|
|
import app.db as module
|
|
with module.connect() as db:
|
|
source_id = insert_source(db, '{}')
|
|
module.init_db()
|
|
with module.connect() as db:
|
|
strategy = json.loads(db.execute("SELECT strategy_json FROM sources WHERE id=?", (source_id,)).fetchone()[0])
|
|
assert strategy["mode"] == "html"
|
|
assert strategy["item_selector"] == ".event"
|
|
|
|
|
|
def test_crawl_deduplicates_and_preserves_hidden(temp_db, monkeypatch):
|
|
import app.db as db_module
|
|
import app.main as main
|
|
monkeypatch.setattr(main, "connect", db_module.connect)
|
|
with db_module.connect() as db:
|
|
source_id = insert_source(db, '{"mode":"html"}')
|
|
start = datetime(2026, 9, 7, 19, 0, tzinfo=ZoneInfo("Australia/Sydney"))
|
|
extracted = ExtractedEvent("Trivia", start, description="First", url="https://example.org/trivia", address="1 Test St")
|
|
|
|
class ClientContext:
|
|
def __enter__(self): return object()
|
|
def __exit__(self, *args): return None
|
|
|
|
monkeypatch.setattr(main, "_http_client", lambda timeout=45: ClientContext())
|
|
monkeypatch.setattr(main, "execute_strategy", lambda source, client: [extracted, extracted])
|
|
assert main.crawl_source(source_id) == 2
|
|
with db_module.connect() as db:
|
|
rows = db.execute("SELECT * FROM events").fetchall()
|
|
assert len(rows) == 1
|
|
event_id = rows[0]["id"]
|
|
db.execute("UPDATE events SET hidden=1 WHERE id=?", (event_id,))
|
|
extracted.description = "Updated"
|
|
main.crawl_source(source_id)
|
|
with db_module.connect() as db:
|
|
row = db.execute("SELECT * FROM events WHERE id=?", (event_id,)).fetchone()
|
|
assert row["description"] == "Updated"
|
|
assert row["hidden"] == 1
|
|
|
|
|
|
def test_successful_empty_crawl_removes_all_stale_events(temp_db, monkeypatch):
|
|
import app.db as db_module
|
|
import app.main as main
|
|
monkeypatch.setattr(main, "connect", db_module.connect)
|
|
with db_module.connect() as db:
|
|
source_id = insert_source(db, '{"mode":"html"}')
|
|
db.execute("INSERT INTO events(source_id,external_key,title,starts_at,venue_name,address,event_url,hidden) VALUES(?,?,?,?,?,?,?,?)", (source_id, "stale", "Old event", "2026-09-07T10:00:00+00:00", "Venue", "Address", "https://example.org/old", 1))
|
|
|
|
class ClientContext:
|
|
def __enter__(self): return object()
|
|
def __exit__(self, *args): return None
|
|
|
|
monkeypatch.setattr(main, "_http_client", lambda timeout=45: ClientContext())
|
|
monkeypatch.setattr(main, "execute_strategy", lambda source, client: [])
|
|
assert main.crawl_source(source_id) == 0
|
|
with db_module.connect() as db:
|
|
assert db.execute("SELECT count(*) FROM events WHERE source_id=?", (source_id,)).fetchone()[0] == 0
|
|
|
|
|
|
def test_failed_crawl_preserves_existing_events(temp_db, monkeypatch):
|
|
import app.db as db_module
|
|
import app.main as main
|
|
monkeypatch.setattr(main, "connect", db_module.connect)
|
|
with db_module.connect() as db:
|
|
source_id = insert_source(db, '{"mode":"html"}')
|
|
db.execute("INSERT INTO events(source_id,external_key,title,starts_at,venue_name,address,event_url) VALUES(?,?,?,?,?,?,?)", (source_id, "existing", "Existing event", "2026-09-07T10:00:00+00:00", "Venue", "Address", "https://example.org/existing"))
|
|
|
|
class ClientContext:
|
|
def __enter__(self): return object()
|
|
def __exit__(self, *args): return None
|
|
|
|
monkeypatch.setattr(main, "_http_client", lambda timeout=45: ClientContext())
|
|
monkeypatch.setattr(main, "execute_strategy", lambda source, client: (_ for _ in ()).throw(RuntimeError("crawl failed")))
|
|
try:
|
|
main.crawl_source(source_id)
|
|
assert False, "Expected crawl failure"
|
|
except RuntimeError:
|
|
pass
|
|
with db_module.connect() as db:
|
|
assert db.execute("SELECT count(*) FROM events WHERE source_id=?", (source_id,)).fetchone()[0] == 1
|
|
|
|
|
|
def test_external_key_database_constraint(temp_db):
|
|
import app.db as module
|
|
with module.connect() as db:
|
|
source_id = insert_source(db)
|
|
values = (source_id, "same", "One", "2026-09-07T10:00:00+00:00", "Venue", "Address", "https://example.org/one")
|
|
db.execute("INSERT INTO events(source_id,external_key,title,starts_at,venue_name,address,event_url) VALUES(?,?,?,?,?,?,?)", values)
|
|
try:
|
|
db.execute("INSERT INTO events(source_id,external_key,title,starts_at,venue_name,address,event_url) VALUES(?,?,?,?,?,?,?)", values)
|
|
assert False, "Expected unique constraint"
|
|
except sqlite3.IntegrityError:
|
|
pass
|