miem_workers/tests/test_db_schema.py

from pathlib import Path
import sqlite3

from sqlalchemy import create_engine, inspect, text
from sqlalchemy.dialects.sqlite import dialect
from sqlalchemy.schema import CreateTable

from app.db import _ensure_runtime_schema
from app.models import SystemUser


def test_runtime_schema_adds_skipped_count_to_existing_crawl_runs_table(monkeypatch):
    engine = create_engine("sqlite:///:memory:")
    with engine.begin() as connection:
        connection.execute(
            text(
                """
                CREATE TABLE crawl_runs (
                  id INTEGER PRIMARY KEY,
                  source_url TEXT NOT NULL,
                  status VARCHAR(32) NOT NULL DEFAULT 'running',
                  found_count INTEGER NOT NULL DEFAULT 0,
                  parsed_count INTEGER NOT NULL DEFAULT 0
                )
                """
            )
        )
    monkeypatch.setattr("app.db.engine", engine)

    _ensure_runtime_schema()

    columns = {column["name"] for column in inspect(engine).get_columns("crawl_runs")}
    assert "skipped_count" in columns


def test_runtime_schema_creates_employee_publications_table_when_employees_exist(monkeypatch):
    engine = create_engine("sqlite:///:memory:")
    with engine.begin() as connection:
        connection.execute(
            text(
                """
                CREATE TABLE employees (
                  id INTEGER PRIMARY KEY,
                  profile_key VARCHAR(255) NOT NULL UNIQUE,
                  canonical_url TEXT NOT NULL,
                  status VARCHAR(32) NOT NULL DEFAULT 'active',
                  first_seen_at DATETIME NOT NULL,
                  last_seen_at DATETIME NOT NULL,
                  created_at DATETIME NOT NULL,
                  updated_at DATETIME NOT NULL
                )
                """
            )
        )
        connection.execute(
            text(
                """
                CREATE TABLE crawl_runs (
                  id INTEGER PRIMARY KEY,
                  source_url TEXT NOT NULL,
                  status VARCHAR(32) NOT NULL DEFAULT 'running',
                  found_count INTEGER NOT NULL DEFAULT 0,
                  parsed_count INTEGER NOT NULL DEFAULT 0,
                  skipped_count INTEGER NOT NULL DEFAULT 0
                )
                """
            )
        )
    monkeypatch.setattr("app.db.engine", engine)

    _ensure_runtime_schema()
    _ensure_runtime_schema()

    inspector = inspect(engine)
    assert "employee_publications" in inspector.get_table_names()
    columns = {column["name"] for column in inspector.get_columns("employee_publications")}
    assert {"employee_id", "publication_id", "doi_url", "authors", "raw_data", "source_hash"}.issubset(columns)


def test_runtime_schema_creates_employee_news_links_table_when_employees_exist(monkeypatch):
    engine = create_engine("sqlite:///:memory:")
    with engine.begin() as connection:
        connection.execute(
            text(
                """
                CREATE TABLE employees (
                  id INTEGER PRIMARY KEY,
                  profile_key VARCHAR(255) NOT NULL UNIQUE,
                  canonical_url TEXT NOT NULL,
                  status VARCHAR(32) NOT NULL DEFAULT 'active',
                  first_seen_at DATETIME NOT NULL,
                  last_seen_at DATETIME NOT NULL,
                  created_at DATETIME NOT NULL,
                  updated_at DATETIME NOT NULL
                )
                """
            )
        )
        connection.execute(
            text(
                """
                CREATE TABLE crawl_runs (
                  id INTEGER PRIMARY KEY,
                  source_url TEXT NOT NULL,
                  status VARCHAR(32) NOT NULL DEFAULT 'running',
                  found_count INTEGER NOT NULL DEFAULT 0,
                  parsed_count INTEGER NOT NULL DEFAULT 0,
                  skipped_count INTEGER NOT NULL DEFAULT 0
                )
                """
            )
        )
    monkeypatch.setattr("app.db.engine", engine)

    _ensure_runtime_schema()
    _ensure_runtime_schema()

    inspector = inspect(engine)
    assert "employee_news_links" in inspector.get_table_names()
    columns = {column["name"] for column in inspector.get_columns("employee_news_links")}
    assert {"employee_id", "title", "url", "summary", "published_at", "published_year", "source_hash", "raw_data"}.issubset(columns)


def test_runtime_schema_adds_profile_verification_fields(monkeypatch):
    engine = create_engine("sqlite:///:memory:")
    with engine.begin() as connection:
        connection.execute(
            text(
                """
                CREATE TABLE employees (
                  id INTEGER PRIMARY KEY,
                  profile_key VARCHAR(255) NOT NULL UNIQUE,
                  canonical_url TEXT NOT NULL,
                  status VARCHAR(32) NOT NULL DEFAULT 'active',
                  first_seen_at DATETIME NOT NULL,
                  last_seen_at DATETIME NOT NULL,
                  created_at DATETIME NOT NULL,
                  updated_at DATETIME NOT NULL
                )
                """
            )
        )
    monkeypatch.setattr("app.db.engine", engine)

    _ensure_runtime_schema()

    inspector = inspect(engine)
    columns = {column["name"] for column in inspector.get_columns("employees")}
    assert {"profile_unavailable_streak", "last_profile_check_at", "has_academic_degree"}.issubset(columns)
    assert "employee_profile_urls" in inspector.get_table_names()


def test_sqlite_council_role_migration_preserves_users_and_history():
    original = str(CreateTable(SystemUser.__table__).compile(dialect=dialect())).replace(", 'council'", "")
    with sqlite3.connect(":memory:") as connection:
        connection.execute(original)
        connection.execute("CREATE TABLE user_actions (user_id INTEGER REFERENCES system_users(id))")
        connection.execute("""
            INSERT INTO system_users
                (id, username, full_name, role, is_active, password_hash, must_change_password,
                 session_version, failed_login_attempts, created_at)
            VALUES (17, 'reader', '', 'user', 1, 'fixture', 0, 4, 0, '2026-11-12')
        """)
        connection.execute("INSERT INTO user_actions VALUES (17)")
        before = connection.execute("SELECT * FROM system_users").fetchall()
        migration = Path("migrations/sqlite/014_system_user_council_role.sql").read_text(encoding="utf-8")
        connection.executescript(migration)
        assert connection.execute("SELECT * FROM system_users").fetchall() == before
        assert connection.execute("SELECT * FROM user_actions").fetchall() == [(17,)]
        connection.execute("UPDATE system_users SET role = 'council' WHERE id = 17")
        assert connection.execute("SELECT role FROM system_users").fetchone() == ("council",)