miem_workers/tests/test_publication_import_postgres.py

"""Real PostgreSQL checks; opt in only inside tests/postgres/compose.yml."""

import copy
import csv
import json
import os
import subprocess
import sys
from contextlib import contextmanager
from pathlib import Path

import pytest
import requests
from sqlalchemy import create_engine, select, text
from sqlalchemy.orm import Session

from app.config import Settings
from app.db import Base
from app.models import CrawlError, CrawlRun, Employee, EmployeePublication
from app.services.crawler import _try_sync_employee_publications, run_crawl


ADMIN_URL = "postgresql+psycopg://test_admin:isolated_test_admin@postgres-test:5432/miem_import_test"
pytestmark = pytest.mark.skipif(
    os.environ.get("MIEM_POSTGRES_INTEGRATION") != "1", reason="Requires isolated PostgreSQL Compose project"
)


@pytest.fixture(autouse=True)
def isolated_environment(monkeypatch):
    assert os.environ.get("DATABASE_URL") == ADMIN_URL, "Use the dedicated test Compose project"

    def forbidden_request(*args, **kwargs):
        raise AssertionError("External HTTP requests are forbidden in PostgreSQL integration tests")

    monkeypatch.setattr(requests.sessions.Session, "request", forbidden_request)


@contextmanager
def database(name):
    assert name in {"miem_import_test", "miem_crawl_test"}
    engine = create_engine(ADMIN_URL.replace("/miem_import_test", "/" + name))
    try:
        with engine.connect() as connection:
            assert connection.scalar(text("SELECT current_database()")) == name
        Base.metadata.create_all(engine)
        with engine.begin() as connection:
            connection.execute(text("GRANT USAGE ON SCHEMA public TO import_test"))
            connection.execute(text("GRANT SELECT (id, full_name) ON public.employees TO import_test"))
            connection.execute(text("GRANT SELECT (id, employee_id, title, year, doi_url, raw_data) ON public.employee_publications TO import_test"))
            connection.execute(text("GRANT INSERT (employee_id, title, year, doi_url, raw_data, source_hash, created_at, updated_at) ON public.employee_publications TO import_test"))
            connection.execute(text("GRANT UPDATE (title, year, doi_url, raw_data, updated_at) ON public.employee_publications TO import_test"))
            # Created from the actual model: not an assumed sequence name.
            sequence = connection.scalar(text("SELECT pg_get_serial_sequence('public.employee_publications', 'id')"))
            assert sequence == "public.employee_publications_id_seq"
            connection.execute(text("GRANT USAGE ON SEQUENCE public.employee_publications_id_seq TO import_test"))
            permissions = connection.execute(text("""SELECT
                has_schema_privilege('import_test', 'public', 'CREATE'),
                has_table_privilege('import_test', 'public.employees', 'INSERT'),
                has_table_privilege('import_test', 'public.employee_publications', 'DELETE'),
                has_sequence_privilege('import_test', 'public.employee_publications_id_seq', 'UPDATE')
            """)).one()
            assert not any(permissions)
        with Session(engine) as session:
            yield session, name
    finally:
        # Only the dedicated ephemeral test DB; never DROP DATABASE or roles.
        Base.metadata.drop_all(engine)
        engine.dispose()


@pytest.fixture
def pg_import_db():
    with database("miem_import_test") as db:
        yield db


@pytest.fixture
def pg_crawl_db():
    with database("miem_crawl_test") as db:
        yield db


def employee(session, name="Synthetic One", key="one"):
    record = Employee(full_name=name, profile_key="staff:" + key,
                      canonical_url="https://synthetic.invalid/staff/" + key)
    session.add(record)
    session.commit()
    return record


def article(title="Synthetic article", identifier="10.1234/test", year="2024", author="Synthetic One", **metadata):
    return {"Название публикации": title, "Год": year, "DOI": identifier,
            "Авторы НИУ ВШЭ": author, **metadata}


def write_csv(path, rows):
    columns = list(dict.fromkeys(key for row in rows for key in row))
    with path.open("w", encoding="utf-8-sig", newline="") as output:
        writer = csv.DictWriter(output, fieldnames=columns, delimiter=";")
        writer.writeheader()
        writer.writerows(rows)


def cli(tmp_path, db, rows, *, apply=False, expected_code=0):
    _, name = db
    source = tmp_path / "articles.csv"
    report = tmp_path / "report.json"
    write_csv(source, rows)
    command = [sys.executable, str(Path(__file__).resolve().parents[1] / "scripts/import_publications.py"),
               str(source), "--report", str(report)]
    if apply:
        command.append("--apply")
    # Explicit whitelist: no production env, .env, pg_service or ~/.pgpass.
    environment = {"PATH": os.environ["PATH"], "LANG": "C.UTF-8", "PGHOST": "postgres-test",
                   "PGPORT": "5432", "PGDATABASE": name, "PGUSER": "import_test",
                   "PGPASSWORD": "isolated_test_import", "PGPASSFILE": "/dev/null",
                   "PGSERVICEFILE": "/dev/null", "PGCONNECT_TIMEOUT": "5"}
    result = subprocess.run(command, env=environment, cwd=tmp_path, capture_output=True, text=True, timeout=60)
    assert result.returncode == expected_code, result.stdout + result.stderr
    assert "Synthetic One" not in result.stdout and "isolated_test_import" not in result.stdout
    return json.loads(report.read_text()) if report.exists() else None


def snapshot(session):
    session.expire_all()
    tables = {table.name: session.execute(select(table).order_by(*table.primary_key.columns)).all()
              for table in Base.metadata.sorted_tables}
    sequences = session.execute(text("SELECT sequencename, last_value FROM pg_sequences WHERE schemaname = 'public' ORDER BY sequencename")).all()
    session.commit()
    return tables, sequences


def test_postgres_dry_run_changes_neither_rows_nor_sequences(pg_import_db, tmp_path):
    session, _ = pg_import_db
    employee(session)
    before = snapshot(session)
    report = cli(tmp_path, pg_import_db, [article()])
    assert report["summary"]["counts"] == {"insert": 1, "update": 0, "skip": 0}
    assert snapshot(session) == before


def test_postgres_apply_and_reimport(pg_import_db, tmp_path):
    session, _ = pg_import_db
    person = employee(session)
    existing = EmployeePublication(employee_id=person.id, title="Existing", year=2024,
                                   source_hash="existing", raw_data={"hse": "keep"})
    session.add(existing)
    session.commit()
    rows = [article(title="Existing"), article(title="New", identifier="10.1234/new"),
            article(title="Unknown author", identifier="10.1234/unknown", author="Synthetic Missing")]
    report = cli(tmp_path, pg_import_db, rows, apply=True)
    assert report["summary"]["counts"] == {"insert": 1, "update": 1, "skip": 1}
    session.expire_all()
    assert session.query(Employee).count() == 1
    assert session.query(EmployeePublication).count() == 2
    assert existing.doi_url == "https://doi.org/10.1234/test" and existing.raw_data["hse"] == "keep"
    before = snapshot(session)
    repeated = cli(tmp_path, pg_import_db, rows, apply=True)
    assert repeated["summary"]["counts"] == {"insert": 0, "update": 0, "skip": 3}
    assert snapshot(session) == before


def test_postgres_apply_failure_rolls_back_update_and_insert(pg_import_db, tmp_path):
    session, _ = pg_import_db
    person = employee(session)
    existing = EmployeePublication(employee_id=person.id, title="Existing", year=2024,
                                   source_hash="existing", raw_data={"hse": "keep"})
    session.add(existing)
    session.commit()
    # Test-only constraint forces a real server error after update + successful INSERT.
    session.execute(text("ALTER TABLE public.employee_publications ADD CONSTRAINT test_failure CHECK (title <> 'Reject')"))
    session.commit()
    before, _ = snapshot(session)
    rows = [article(title="Existing"), article(title="Accepted", identifier="10.1234/accepted"),
            article(title="Reject", identifier="10.1234/reject")]
    assert cli(tmp_path, pg_import_db, rows, apply=True, expected_code=1) is None
    after, _ = snapshot(session)
    assert after == before  # nextval may advance on rollback; rows must not change.
    assert (tmp_path / "report.json.pending").exists()


def test_postgres_year_only_update_preserves_json(pg_import_db, tmp_path):
    session, _ = pg_import_db
    person = employee(session)
    row = article()
    raw = {"hse": {"keep": True}, "csv_enrichment": row.copy()}
    record = EmployeePublication(employee_id=person.id, title=row["Название публикации"],
                                 year=None, doi_url="https://doi.org/10.1234/test",
                                 source_hash="existing", raw_data=raw)
    session.add(record)
    session.commit()
    report = cli(tmp_path, pg_import_db, [row], apply=True)
    session.expire_all()
    assert report["summary"]["counts"]["update"] == 1
    assert record.year == 2024 and record.raw_data == raw


def test_postgres_late_csv_metadata_and_conflicts(pg_import_db, tmp_path):
    session, _ = pg_import_db
    employee(session)
    rows = [article(), article(**{"EID (Scopus)": "late-id", "Квартиль scopus": "Q1"}),
            article(**{"EID (Scopus)": "conflict-id", "UT (WoS)": "late-wos"})]
    report = cli(tmp_path, pg_import_db, rows, apply=True)
    session.expire_all()
    assert session.query(EmployeePublication).count() == 1
    metadata = session.query(EmployeePublication).one().raw_data["csv_enrichment"]
    assert metadata["EID (Scopus)"] == "late-id" and metadata["Квартиль scopus"] == "Q1"
    assert metadata["UT (WoS)"] == "late-wos"
    assert report["summary"]["metadata_conflicts"] == 1
    assert report["records"][2]["conflict_fields"] == ["EID (Scopus)"]
    before = snapshot(session)
    repeated = cli(tmp_path, pg_import_db, rows, apply=True)
    assert repeated["summary"]["counts"]["insert"] == repeated["summary"]["counts"]["update"] == 0
    assert snapshot(session) == before


def test_postgres_crawler_preserves_import_and_adopts_hse_id(pg_crawl_db, tmp_path):
    session, _ = pg_crawl_db
    person = employee(session)
    cli(tmp_path, pg_crawl_db, [article(), article(title="CSV only", identifier="10.1234/only")], apply=True)
    session.expire_all()
    ids = {p.title: p.id for p in session.query(EmployeePublication).all()}
    run = CrawlRun(source_url="https://synthetic.invalid/persons", status="running")
    session.add(run)
    session.commit()
    parsed = {"sections": [{"type": "publications", "publications": [
        {"publication_id": "hse-one", "title": "Synthetic article", "year": 2024,
         "doi_url": "https://doi.org/10.1234/test", "raw_data": {"id": "hse-one"}}]}]}
    for _ in range(2):
        _try_sync_employee_publications(session, run, person, copy.deepcopy(parsed))
        session.commit()
        session.expire_all()
    assert session.query(EmployeePublication).count() == 2
    kept = session.get(EmployeePublication, ids["Synthetic article"])
    assert kept.publication_id == "hse-one" and "csv_enrichment" in kept.raw_data
    assert session.get(EmployeePublication, ids["CSV only"]).raw_data["csv_enrichment"]
    assert session.query(CrawlError).count() == 0


@pytest.mark.parametrize("bound_id", [None, "hse-A"])
@pytest.mark.parametrize("incoming_ids", [("hse-A", "hse-B"), ("hse-B", "hse-A")])
def test_postgres_hse_id_conflict_does_not_block_other_employee(pg_crawl_db, tmp_path, monkeypatch, bound_id, incoming_ids):
    session, _ = pg_crawl_db
    first = employee(session)
    second = employee(session, "Synthetic Two", "two")
    cli(tmp_path, pg_crawl_db, [article(identifier="")], apply=True)
    session.expire_all()
    original = session.query(EmployeePublication).filter_by(employee_id=first.id).one()
    original.publication_id = bound_id
    original.authors = [{"id": "original-author"}]
    session.commit()
    fields = ("publication_id", "source_hash", "raw_data", "authors", "updated_at")
    before = {field: copy.deepcopy(getattr(original, field)) for field in fields}
    session.commit()

    def profile(person, publications):
        return {"source_url": person.canonical_url, "profile_type": "staff",
                "profile_id": person.profile_key.split(":")[1], "full_name": person.full_name,
                "parser_version": "integration-test", "tabs": [],
                "sections": [{"type": "publications", "publications": publications}]}

    profiles = {
        first.canonical_url: profile(first, [{"publication_id": key, "title": "Synthetic article", "year": 2024,
                                             "authors": [{"id": key}], "raw_data": {"id": key}} for key in incoming_ids]),
        second.canonical_url: profile(second, [{"publication_id": "second-article", "title": "Other employee article",
                                               "year": 2024, "raw_data": {"id": "second-article"}}]),
    }
    monkeypatch.setattr("app.services.crawler.collect_profile_links", lambda *a, **k: list(profiles))
    monkeypatch.setattr("app.services.crawler.parse_person_profile", lambda _, url, *a, **k: copy.deepcopy(profiles[url]))
    settings = Settings(_env_file=None, database_url=ADMIN_URL.replace("miem_import_test", "miem_crawl_test"),
                        source_url="https://synthetic.invalid/persons", request_delay_seconds=0)
    run = run_crawl(session, settings)
    session.expire_all()
    assert run.status == "completed" and run.parsed_count == 2
    kept = session.query(EmployeePublication).filter_by(employee_id=first.id).one()
    assert {field: getattr(kept, field) for field in fields} == before
    assert session.query(EmployeePublication).filter_by(employee_id=second.id).one().publication_id == "second-article"
    error = session.query(CrawlError).filter_by(crawl_run_id=run.id).one()
    assert "Ambiguous HSE IDs" in error.message