Archived
443 lines
16 KiB
Python
443 lines
16 KiB
Python
import asyncio
|
|
import logging
|
|
import sqlite3
|
|
import time
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import karldns.app as app_module
|
|
from karldns import DdnsSettings, DdnsStore, make_app
|
|
from karldns.event_statuses import CLEANUP_PENDING_HOSTNAME_DELETED, CLEANUP_UNUSED_HOSTNAME_DELETED
|
|
from karldns.security import hash_lookup_token
|
|
|
|
|
|
def test_rate_limit_reject_does_not_increment_bucket(tmp_path: Path) -> None:
|
|
store = DdnsStore(tmp_path / "ddns.sqlite3")
|
|
|
|
assert store.allow_rate_limit("198.51.100.1:/", 1) is True
|
|
assert store.allow_rate_limit("198.51.100.1:/", 1) is False
|
|
store.flush_rate_limits()
|
|
|
|
window = int(time.time()) // 60
|
|
with store.connect() as conn:
|
|
count = conn.execute(
|
|
"SELECT count FROM rate_limits WHERE key = ? AND window = ?",
|
|
("198.51.100.1:/", window),
|
|
).fetchone()[0]
|
|
assert count == 1
|
|
|
|
|
|
def test_rate_limit_flush_persists_batched_counts(tmp_path: Path) -> None:
|
|
store = DdnsStore(tmp_path / "ddns.sqlite3")
|
|
|
|
assert store.allow_rate_limit("198.51.100.1:/api", 60) is True
|
|
assert store.allow_rate_limit("198.51.100.1:/api", 60) is True
|
|
store.flush_rate_limits()
|
|
|
|
window = int(time.time()) // 60
|
|
with store.connect() as conn:
|
|
count = conn.execute(
|
|
"SELECT count FROM rate_limits WHERE key = ? AND window = ?",
|
|
("198.51.100.1:/api", window),
|
|
).fetchone()[0]
|
|
assert count == 2
|
|
|
|
|
|
def test_cleanup_deletes_old_rate_limit_minute_buckets(tmp_path: Path) -> None:
|
|
store = DdnsStore(tmp_path / "ddns.sqlite3")
|
|
current_window = int(time.time()) // 60
|
|
|
|
with store.connect() as conn:
|
|
conn.execute(
|
|
"INSERT INTO rate_limits(key, window, count) VALUES (?, ?, ?)",
|
|
("old", current_window - 61, 1),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO rate_limits(key, window, count) VALUES (?, ?, ?)",
|
|
("recent", current_window - 30, 1),
|
|
)
|
|
|
|
store.cleanup(unused_account_hours=1)
|
|
|
|
with store.connect() as conn:
|
|
keys = {row[0] for row in conn.execute("SELECT key FROM rate_limits")}
|
|
assert keys == {"recent"}
|
|
|
|
|
|
def test_scheduled_cleanup_task_is_durable_and_lease_protected(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "ddns.sqlite3"
|
|
first = DdnsStore(database_path)
|
|
first.ensure_scheduled_task("cleanup", 3600)
|
|
|
|
second = DdnsStore(database_path)
|
|
task = second.list_scheduled_tasks()[0]
|
|
assert task["name"] == "cleanup"
|
|
assert task["interval_seconds"] == 3600
|
|
assert task["last_status"] == "pending"
|
|
|
|
assert first.acquire_due_task("cleanup", "worker-a", 60) is True
|
|
assert second.acquire_due_task("cleanup", "worker-b", 60) is False
|
|
|
|
first.complete_scheduled_task("cleanup", 3600, True)
|
|
completed = second.list_scheduled_tasks()[0]
|
|
assert completed["lease_owner"] is None
|
|
assert completed["lease_expires_at"] is None
|
|
assert completed["last_status"] == "ok"
|
|
assert completed["last_finished_at"]
|
|
assert completed["next_run_at"] > str(task["next_run_at"])
|
|
|
|
|
|
def test_first_run_creates_schema_and_repeated_startup_is_idempotent(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "nested" / "ddns.sqlite3"
|
|
settings = DdnsSettings(database_path=database_path, admin_password="admin")
|
|
|
|
make_app(settings)
|
|
make_app(settings)
|
|
|
|
store = DdnsStore(database_path)
|
|
with store.connect() as conn:
|
|
tables = {row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'")}
|
|
assert {
|
|
"accounts",
|
|
"cleanup_runs",
|
|
"rate_limits",
|
|
"records",
|
|
"scheduled_tasks",
|
|
"service_metrics",
|
|
"update_events",
|
|
}.issubset(tables)
|
|
assert len(store.list_scheduled_tasks()) == 1
|
|
assert store.list_scheduled_tasks()[0]["name"] == "cleanup"
|
|
|
|
|
|
def test_startup_migrates_legacy_cleanup_run_columns(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "ddns.sqlite3"
|
|
with sqlite3.connect(database_path) as conn:
|
|
conn.execute("""
|
|
CREATE TABLE cleanup_runs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
created_at TEXT NOT NULL,
|
|
domain_challenges_deleted INTEGER NOT NULL,
|
|
abandoned_setups_deleted INTEGER NOT NULL,
|
|
unused_accounts_deleted INTEGER NOT NULL
|
|
)
|
|
""")
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO cleanup_runs(
|
|
created_at,
|
|
domain_challenges_deleted,
|
|
abandoned_setups_deleted,
|
|
unused_accounts_deleted
|
|
)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
("2026-01-01T00:00:00+00:00", 3, 2, 1),
|
|
)
|
|
|
|
store = DdnsStore(database_path)
|
|
|
|
with store.connect() as conn:
|
|
columns = {row[1] for row in conn.execute("PRAGMA table_info(cleanup_runs)")}
|
|
row = conn.execute("""
|
|
SELECT unused_accounts_deleted
|
|
FROM cleanup_runs
|
|
""").fetchone()
|
|
assert columns == {"id", "created_at", "unused_accounts_deleted"}
|
|
assert tuple(row) == (1,)
|
|
|
|
|
|
def test_cleanup_records_only_unused_hostname_count(tmp_path: Path) -> None:
|
|
store = DdnsStore(tmp_path / "ddns.sqlite3")
|
|
|
|
result = store.cleanup(unused_account_hours=1)
|
|
|
|
assert result["unused_accounts"] == 0
|
|
assert set(store.list_cleanup_runs()[0]) == {"created_at", "unused_accounts_deleted"}
|
|
|
|
|
|
def test_cleanup_removes_pending_generated_setups_after_cutoff(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "ddns.sqlite3"
|
|
store = DdnsStore(database_path)
|
|
account = store.create_account("pending.ddns.example.net", "router", accepted=False)
|
|
|
|
with store.connect() as conn:
|
|
conn.execute(
|
|
"UPDATE accounts SET created_at = ? WHERE hostname = ?",
|
|
("2000-01-01T00:00:00+00:00", account["hostname"]),
|
|
)
|
|
|
|
result = store.cleanup(unused_account_hours=45 * 24)
|
|
|
|
assert result["unused_accounts"] == 1
|
|
assert store.get_account_by_slug(account["update_slug"]) is None
|
|
cleanup_events = store.list_update_events_for_hostname(account["hostname"])
|
|
assert cleanup_events[0]["status"] == CLEANUP_PENDING_HOSTNAME_DELETED
|
|
assert cleanup_events[0]["detail"] == "cleanup deleted abandoned hostname"
|
|
|
|
|
|
def test_cleanup_keeps_recent_pending_setup_until_pending_cutoff(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "ddns.sqlite3"
|
|
store = DdnsStore(database_path)
|
|
account = store.create_account("pending.ddns.example.net", "router", accepted=False)
|
|
|
|
recent = (datetime.now(UTC) - timedelta(hours=2)).isoformat()
|
|
with store.connect() as conn:
|
|
conn.execute(
|
|
"UPDATE accounts SET created_at = ? WHERE hostname = ?",
|
|
(recent, account["hostname"]),
|
|
)
|
|
|
|
result = store.cleanup(unused_account_hours=1, pending_account_hours=24)
|
|
|
|
assert result["unused_accounts"] == 0
|
|
assert store.get_account_by_slug(account["update_slug"]) is not None
|
|
|
|
|
|
def test_cleanup_logs_event_for_accepted_unused_hostname(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "ddns.sqlite3"
|
|
store = DdnsStore(database_path)
|
|
account = store.create_account("unused.ddns.example.net", "router", accepted=True)
|
|
|
|
with store.connect() as conn:
|
|
conn.execute(
|
|
"UPDATE accounts SET created_at = ? WHERE hostname = ?",
|
|
("2000-01-01T00:00:00+00:00", account["hostname"]),
|
|
)
|
|
|
|
result = store.cleanup(unused_account_hours=1)
|
|
|
|
assert result["unused_accounts"] == 1
|
|
assert store.get_account_by_slug(account["update_slug"]) is None
|
|
cleanup_events = store.list_update_events_for_hostname(account["hostname"])
|
|
assert cleanup_events[0]["status"] == CLEANUP_UNUSED_HOSTNAME_DELETED
|
|
assert cleanup_events[0]["detail"] == "cleanup deleted abandoned hostname"
|
|
|
|
|
|
def test_pending_generated_setup_stays_pending_after_restart(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "ddns.sqlite3"
|
|
store = DdnsStore(database_path)
|
|
account = store.create_account("pending.ddns.example.net", "router", accepted=False)
|
|
store.close()
|
|
|
|
restarted = DdnsStore(database_path)
|
|
|
|
assert restarted.get_account_by_slug(account["update_slug"])["accepted_at"] is None
|
|
|
|
|
|
def test_sqlite_startup_tuning_is_idempotent(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "ddns.sqlite3"
|
|
|
|
DdnsStore(database_path)
|
|
store = DdnsStore(database_path)
|
|
store.optimize()
|
|
store.checkpoint_wal()
|
|
|
|
with store.connect() as conn:
|
|
busy_timeout = conn.execute("PRAGMA busy_timeout").fetchone()[0]
|
|
journal_mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
|
|
synchronous = conn.execute("PRAGMA synchronous").fetchone()[0]
|
|
cache_size = conn.execute("PRAGMA cache_size").fetchone()[0]
|
|
temp_store = conn.execute("PRAGMA temp_store").fetchone()[0]
|
|
mmap_size = conn.execute("PRAGMA mmap_size").fetchone()[0]
|
|
user_version = conn.execute("PRAGMA user_version").fetchone()[0]
|
|
index_names = {row[1] for row in conn.execute("""
|
|
SELECT type, name
|
|
FROM sqlite_master
|
|
WHERE type = 'index'
|
|
""")}
|
|
|
|
assert busy_timeout == 10_000
|
|
assert journal_mode == "wal"
|
|
assert synchronous == 1
|
|
assert cache_size == -20_000
|
|
assert temp_store == 2
|
|
assert mmap_size == 256 * 1024 * 1024
|
|
assert user_version == 3
|
|
assert {
|
|
"idx_accounts_created_at",
|
|
"idx_rate_limits_window",
|
|
"idx_update_events_hostname_created_at",
|
|
"idx_update_event_rollups_hostname_last_seen_at",
|
|
}.issubset(index_names)
|
|
|
|
|
|
def test_accounts_schema_migration_removes_legacy_router_credentials(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "ddns.sqlite3"
|
|
with sqlite3.connect(database_path) as conn:
|
|
conn.execute("PRAGMA user_version = 2")
|
|
conn.execute("""
|
|
CREATE TABLE accounts (
|
|
hostname TEXT PRIMARY KEY,
|
|
username TEXT NOT NULL,
|
|
token_hash TEXT NOT NULL,
|
|
update_slug TEXT NOT NULL UNIQUE,
|
|
management_hash TEXT UNIQUE,
|
|
dashboard_password_hash TEXT,
|
|
created_at TEXT NOT NULL,
|
|
accepted_at TEXT,
|
|
disabled INTEGER NOT NULL DEFAULT 0
|
|
)
|
|
""")
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO accounts(
|
|
hostname, username, token_hash, update_slug, management_hash,
|
|
dashboard_password_hash, created_at, accepted_at, disabled
|
|
)
|
|
VALUES (?, ?, ?, ?, ?, NULL, ?, ?, 0)
|
|
""",
|
|
(
|
|
"home.ddns.example.net",
|
|
"router",
|
|
"old-token-hash",
|
|
"update-secret",
|
|
hash_lookup_token("dashboard-secret"),
|
|
"2026-05-21T00:00:00+00:00",
|
|
"2026-05-21T00:00:00+00:00",
|
|
),
|
|
)
|
|
|
|
store = DdnsStore(database_path)
|
|
account = store.get_account_by_slug("update-secret")
|
|
|
|
with store.connect() as conn:
|
|
columns = {row[1] for row in conn.execute("PRAGMA table_info(accounts)")}
|
|
user_version = conn.execute("PRAGMA user_version").fetchone()[0]
|
|
|
|
assert user_version == 3
|
|
assert "username" not in columns
|
|
assert "token_hash" not in columns
|
|
assert account
|
|
assert account["hostname"] == "home.ddns.example.net"
|
|
|
|
|
|
def test_rate_limit_flush_requeues_counts_after_database_failure(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
store = DdnsStore(tmp_path / "ddns.sqlite3")
|
|
assert store.allow_rate_limit("198.51.100.1:/api", 60) is True
|
|
|
|
original_connect = store.connect
|
|
|
|
def fail_connect():
|
|
raise RuntimeError("database temporarily unavailable")
|
|
|
|
monkeypatch.setattr(store, "connect", fail_connect)
|
|
with pytest.raises(RuntimeError, match="database temporarily unavailable"):
|
|
store.flush_rate_limits()
|
|
|
|
monkeypatch.setattr(store, "connect", original_connect)
|
|
store.flush_rate_limits()
|
|
|
|
window = int(time.time()) // 60
|
|
with store.connect() as conn:
|
|
count = conn.execute(
|
|
"SELECT count FROM rate_limits WHERE key = ? AND window = ?",
|
|
("198.51.100.1:/api", window),
|
|
).fetchone()[0]
|
|
assert count == 1
|
|
|
|
|
|
def test_cleanup_prunes_old_raw_update_events(tmp_path: Path) -> None:
|
|
store = DdnsStore(tmp_path / "ddns.sqlite3")
|
|
store.log_update_event("old.example.net", "203.0.113.1", None, "updated", "old", "198.51.100.1")
|
|
store.log_update_event("new.example.net", "203.0.113.2", None, "updated", "new", "198.51.100.2")
|
|
|
|
with store.connect() as conn:
|
|
conn.execute(
|
|
"UPDATE update_events SET created_at = ? WHERE hostname = ?",
|
|
("2000-01-01T00:00:00+00:00", "old.example.net"),
|
|
)
|
|
|
|
result = store.cleanup(unused_account_hours=1, event_history_hours=24)
|
|
|
|
assert result["update_events"] == 1
|
|
events = store.list_update_events()
|
|
assert [event["hostname"] for event in events] == ["new.example.net"]
|
|
|
|
|
|
def test_background_cleanup_removes_abandoned_hostnames(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "ddns.sqlite3"
|
|
app = make_app(
|
|
DdnsSettings(
|
|
database_path=database_path,
|
|
admin_password="admin",
|
|
cleanup_unused_account_hours=1,
|
|
cleanup_interval_seconds=1,
|
|
)
|
|
)
|
|
store = DdnsStore(database_path)
|
|
|
|
with TestClient(app):
|
|
abandoned_account = store.create_account("abandoned.ddns.example.net", "router")
|
|
active_account = store.create_account("active.ddns.example.net", "router")
|
|
store.upsert("active.ddns.example.net", "203.0.113.10", None)
|
|
|
|
old = "2000-01-01T00:00:00+00:00"
|
|
with store.connect() as conn:
|
|
conn.execute("UPDATE accounts SET created_at = ? WHERE hostname = ?", (old, abandoned_account["hostname"]))
|
|
conn.execute("UPDATE accounts SET created_at = ? WHERE hostname = ?", (old, active_account["hostname"]))
|
|
|
|
time.sleep(1.3)
|
|
|
|
assert store.get_account_by_slug(abandoned_account["update_slug"]) is None
|
|
assert store.get_account_by_slug(active_account["update_slug"]) is not None
|
|
scheduled_task = store.list_scheduled_tasks()[0]
|
|
assert scheduled_task["name"] == "cleanup"
|
|
assert scheduled_task["last_status"] == "ok"
|
|
assert scheduled_task["last_finished_at"]
|
|
|
|
|
|
def test_scheduled_cleanup_failure_is_logged(tmp_path: Path, monkeypatch, caplog) -> None:
|
|
database_path = tmp_path / "ddns.sqlite3"
|
|
settings = DdnsSettings(database_path=database_path, admin_password="admin", cleanup_interval_seconds=60)
|
|
store = DdnsStore(database_path)
|
|
store.ensure_scheduled_task(app_module.CLEANUP_TASK_NAME, settings.cleanup_interval_seconds)
|
|
|
|
def fail_cleanup(unused_account_hours, event_history_hours, pending_account_hours):
|
|
raise RuntimeError("cleanup exploded")
|
|
|
|
caplog.set_level(logging.ERROR, logger="karldns.app")
|
|
monkeypatch.setattr(store, "cleanup", fail_cleanup)
|
|
|
|
asyncio.run(app_module._run_cleanup_if_due(store, settings))
|
|
|
|
scheduled_task = store.list_scheduled_tasks()[0]
|
|
assert scheduled_task["last_status"] == "error"
|
|
assert "RuntimeError: cleanup exploded" in str(scheduled_task["last_error"])
|
|
assert "Scheduled cleanup failed" in caplog.text
|
|
|
|
|
|
def test_scheduler_loop_survives_iteration_exception(tmp_path: Path, monkeypatch, caplog) -> None:
|
|
settings = DdnsSettings(
|
|
database_path=tmp_path / "ddns.sqlite3",
|
|
admin_password="admin",
|
|
cleanup_interval_seconds=60,
|
|
)
|
|
store = DdnsStore(settings.database_path)
|
|
calls = 0
|
|
|
|
async def fail_once_then_cancel(inner_store, inner_settings):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 1:
|
|
raise RuntimeError("post-cleanup crash")
|
|
raise asyncio.CancelledError
|
|
|
|
async def no_sleep(_seconds: float) -> None:
|
|
return None
|
|
|
|
caplog.set_level(logging.ERROR, logger="karldns.app")
|
|
monkeypatch.setattr(app_module, "_run_cleanup_if_due", fail_once_then_cancel)
|
|
monkeypatch.setattr(app_module.asyncio, "sleep", no_sleep)
|
|
|
|
with pytest.raises(asyncio.CancelledError):
|
|
asyncio.run(app_module._scheduler_loop(store, settings))
|
|
|
|
assert calls == 2
|
|
assert "Scheduled cleanup loop crashed; continuing" in caplog.text
|