add sqlite database

tests/test_model.py

@@ -0,0 +1,171 @@
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+
+import pytest
+from peewee import IntegrityError
+
+from repub.model import (
+    Job,
+    Source,
+    database,
+    initialize_database,
+    resolve_database_path,
+)
+
+
+def test_resolve_database_path_defaults_to_republisher_db(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.delenv("REPUBLISHER_DB_PATH", raising=False)
+
+    assert resolve_database_path() == tmp_path / "republisher.db"
+
+
+def test_resolve_database_path_prefers_environment_variable(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    db_path = tmp_path / "env-configured.db"
+    monkeypatch.setenv("REPUBLISHER_DB_PATH", str(db_path))
+
+    assert resolve_database_path() == db_path
+
+
+def test_initialize_database_bootstraps_schema_from_sql_files(tmp_path: Path) -> None:
+    db_path = tmp_path / "bootstrap.db"
+
+    initialize_database(db_path)
+
+    connection = sqlite3.connect(db_path)
+    try:
+        table_names = {
+            row[0]
+            for row in connection.execute(
+                """
+                SELECT name
+                FROM sqlite_master
+                WHERE type = 'table' AND name NOT LIKE 'sqlite_%'
+                """
+            )
+        }
+        assert table_names == {
+            "job",
+            "job_execution",
+            "settings",
+            "source",
+            "source_feed",
+            "source_pangea",
+        }
+
+        defaults = {
+            row[1]: row[4]
+            for row in connection.execute("PRAGMA table_info('source_pangea')")
+        }
+        assert defaults["content_type"] is None
+        assert defaults["only_newest"] is None
+        assert defaults["max_articles"] is None
+        assert defaults["oldest_article"] is None
+        assert defaults["include_authors"] is None
+        assert defaults["exclude_media"] is None
+        assert defaults["include_content"] is None
+        assert defaults["content_format"] is None
+    finally:
+        connection.close()
+
+
+def test_initialize_database_configures_sqlite_pragmas(tmp_path: Path) -> None:
+    db_path = tmp_path / "pragmas.db"
+
+    initialize_database(db_path)
+
+    database.connect(reuse_if_open=True)
+    try:
+        pragma_values = {
+            "cache_size": database.execute_sql("PRAGMA cache_size").fetchone()[0],
+            "page_size": database.execute_sql("PRAGMA page_size").fetchone()[0],
+            "journal_mode": database.execute_sql("PRAGMA journal_mode").fetchone()[0],
+            "synchronous": database.execute_sql("PRAGMA synchronous").fetchone()[0],
+            "temp_store": database.execute_sql("PRAGMA temp_store").fetchone()[0],
+            "foreign_keys": database.execute_sql("PRAGMA foreign_keys").fetchone()[0],
+            "busy_timeout": database.execute_sql("PRAGMA busy_timeout").fetchone()[0],
+        }
+        assert pragma_values == {
+            "cache_size": 15625,
+            "page_size": 4096,
+            "journal_mode": "wal",
+            "synchronous": 1,
+            "temp_store": 2,
+            "foreign_keys": 1,
+            "busy_timeout": 5000,
+        }
+    finally:
+        database.close()
+
+
+def test_initialize_database_creates_scheduler_and_execution_indexes(
+    tmp_path: Path,
+) -> None:
+    db_path = tmp_path / "indexes.db"
+
+    initialize_database(db_path)
+
+    connection = sqlite3.connect(db_path)
+    try:
+        index_names = {
+            row[0]
+            for row in connection.execute(
+                """
+                SELECT name
+                FROM sqlite_master
+                WHERE type = 'index'
+                  AND name IN (
+                    'job_enabled_idx',
+                    'job_execution_job_created_at_idx',
+                    'job_execution_status_started_at_idx',
+                    'job_execution_status_ended_at_idx'
+                  )
+                """
+            )
+        }
+        assert index_names == {
+            "job_enabled_idx",
+            "job_execution_job_created_at_idx",
+            "job_execution_status_started_at_idx",
+            "job_execution_status_ended_at_idx",
+        }
+    finally:
+        connection.close()
+
+
+def test_job_table_allows_exactly_one_job_per_source(tmp_path: Path) -> None:
+    initialize_database(tmp_path / "jobs.db")
+
+    source = Source.create(
+        name="Guardian feed mirror",
+        slug="guardian-feed",
+        source_type="feed",
+    )
+    Job.create(
+        source=source,
+        enabled=True,
+        spider_arguments="",
+        cron_minute="15",
+        cron_hour="*",
+        cron_day_of_month="*",
+        cron_day_of_week="*",
+        cron_month="*",
+    )
+
+    with pytest.raises(IntegrityError):
+        Job.create(
+            source=source,
+            enabled=True,
+            spider_arguments="language=en",
+            cron_minute="30",
+            cron_hour="*",
+            cron_day_of_month="*",
+            cron_day_of_week="*",
+            cron_month="*",
+        )