234 lines
9.6 KiB
Python
234 lines
9.6 KiB
Python
"""Offline checks: synthetic identities only, no settings or database access."""
|
|
from scripts.import_publications import plan, read_rows
|
|
from pathlib import Path
|
|
|
|
|
|
def row(title="Test", doi="10.1234/test", year="2024", authors="Test Person"):
|
|
return {"Название публикации": title, "DOI": doi, "Год": year,
|
|
"Авторы НИУ ВШЭ": authors}
|
|
|
|
|
|
EMPLOYEES = [{"id": 1, "full_name": "Test Person"}]
|
|
|
|
|
|
def test_repeat_and_no_new_employees():
|
|
rows = [(2, row()), (3, row()), (4, row(authors="Missing Person"))]
|
|
actions, audit = plan(rows, EMPLOYEES, [])
|
|
assert len(actions) == 1 and actions[0]["result"] == "insert"
|
|
assert [a["reason"] for a in audit] == ["", "csv_duplicate", "employee_not_found"]
|
|
stored = [{"id": 11, **actions[0]["values"]}]
|
|
repeat, _ = plan(rows, EMPLOYEES, stored)
|
|
assert repeat == []
|
|
|
|
|
|
def test_ambiguity_and_conflicts():
|
|
assert not plan([(2, row())], EMPLOYEES * 2, [])[0]
|
|
stored = [{"id": 1, "employee_id": 1, "title": "Test", "year": 2024,
|
|
"doi_url": "https://doi.org/10.1234/other", "raw_data": {}}]
|
|
assert plan([(2, row())], EMPLOYEES, stored)[1][0]["reason"] == "doi_conflict"
|
|
stored[0]["doi_url"] = "https://doi.org/10.1234/test"
|
|
assert plan([(2, row(year="2025"))], EMPLOYEES, stored)[1][0]["reason"] == "year_conflict"
|
|
assert plan([(2, row())], EMPLOYEES, stored * 2)[0] # same id is one identity
|
|
stored.append({**stored[0], "id": 2})
|
|
assert plan([(2, row())], EMPLOYEES, stored)[1][0]["reason"] == "ambiguous_publication"
|
|
|
|
|
|
def test_fill_only_preserves_metadata_and_empty_values():
|
|
stored = [{"id": 1, "employee_id": 1, "title": "Test", "year": 2024,
|
|
"doi_url": None, "raw_data": {"hse": "keep", "csv_enrichment": {"extra": "keep"}}}]
|
|
actions, _ = plan([(2, {**row(), "extra": "", "new": "added"})], EMPLOYEES, stored)
|
|
values = actions[0]["values"]
|
|
assert set(values) == {"doi_url", "raw_data"}
|
|
assert values["raw_data"]["hse"] == "keep"
|
|
assert values["raw_data"]["csv_enrichment"]["extra"] == "keep"
|
|
assert values["raw_data"]["csv_enrichment"]["new"] == "added"
|
|
assert "new" not in stored[0]["raw_data"]["csv_enrichment"]
|
|
|
|
|
|
def test_missing_doi_then_doi_does_not_update_synthetic_id():
|
|
actions, _ = plan([(2, row(doi="")), (3, row())], EMPLOYEES, [])
|
|
assert len(actions) == 1 and actions[0]["result"] == "insert"
|
|
assert actions[0]["values"]["doi_url"] == "https://doi.org/10.1234/test"
|
|
|
|
|
|
def test_invalid_rows_and_owner_file():
|
|
assert not plan([(2, row(year="oops")), (3, row(doi="bad"))], EMPLOYEES, [])[0]
|
|
path = Path(__file__).resolve().parents[2] / "Статьи.csv"
|
|
if path.exists():
|
|
rows = read_rows(path)
|
|
assert len(rows) == 1797
|
|
actions, audit = plan(rows, [], [])
|
|
assert not actions and audit and all(a["result"] == "skip" for a in audit)
|
|
|
|
|
|
def test_cli_read_only_and_atomic_failure(tmp_path, monkeypatch):
|
|
import sys
|
|
import psycopg
|
|
from scripts import import_publications
|
|
|
|
source = tmp_path / "input.csv"
|
|
source.write_text("Название публикации;Год;DOI;Авторы НИУ ВШЭ\nFirst;2024;10.1234/first;Test Person\nSecond;2024;10.1234/second;Test Person\n", encoding="utf-8")
|
|
|
|
class Result:
|
|
rowcount = 1
|
|
|
|
def __init__(self, rows):
|
|
self.rows = rows
|
|
|
|
def fetchall(self):
|
|
return self.rows
|
|
|
|
def fetchone(self):
|
|
return self.rows[0]
|
|
|
|
class Connection:
|
|
def __init__(self, fail=False):
|
|
self.queries, self.writes = [], []
|
|
self.fail, self.rolled_back, self.committed = fail, False, False
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, kind, *_):
|
|
if kind:
|
|
self.rollback()
|
|
elif not self.rolled_back:
|
|
self.committed = True
|
|
|
|
def rollback(self):
|
|
self.rolled_back = True
|
|
self.writes.clear()
|
|
|
|
def execute(self, query, params=None):
|
|
query = query if isinstance(query, str) else query.as_string()
|
|
self.queries.append(query)
|
|
if query.startswith("SELECT id, full_name"):
|
|
return Result(EMPLOYEES)
|
|
if query.startswith("SELECT id, employee_id"):
|
|
return Result([])
|
|
if "pg_try_advisory" in query:
|
|
return Result([{"locked": True}])
|
|
if query.startswith("INSERT"):
|
|
self.writes.append(params)
|
|
if self.fail and len(self.writes) == 2:
|
|
raise RuntimeError("Synthetic second-write failure")
|
|
return Result([])
|
|
|
|
report = tmp_path / "dry.json"
|
|
dry = Connection()
|
|
monkeypatch.setattr(psycopg, "connect", lambda *a, **k: dry)
|
|
monkeypatch.setattr(sys, "argv", ["import_publications", str(source), "--report", str(report)])
|
|
import_publications.main()
|
|
assert dry.rolled_back and not dry.committed and not dry.writes
|
|
assert not any(q.startswith(("INSERT", "UPDATE", "DELETE")) for q in dry.queries)
|
|
assert "READ ONLY" in dry.queries[0]
|
|
assert report.exists()
|
|
|
|
failed = Connection(fail=True)
|
|
monkeypatch.setattr(psycopg, "connect", lambda *a, **k: failed)
|
|
report = tmp_path / "apply.json"
|
|
monkeypatch.setattr(sys, "argv", ["import_publications", str(source), "--report", str(report), "--apply"])
|
|
import pytest
|
|
with pytest.raises(RuntimeError):
|
|
import_publications.main()
|
|
assert failed.rolled_back and not failed.committed and not failed.writes
|
|
assert not report.exists()
|
|
|
|
success = Connection()
|
|
monkeypatch.setattr(psycopg, "connect", lambda *a, **k: success)
|
|
monkeypatch.setattr(sys, "argv", ["import_publications", str(source), "--report", str(tmp_path / "success.json"), "--apply"])
|
|
import_publications.main()
|
|
assert success.committed and len(success.writes) == 2
|
|
assert not any(q.startswith(("DELETE", "ALTER", "CREATE")) for q in success.queries)
|
|
|
|
|
|
def test_report_cannot_overwrite_source(tmp_path, monkeypatch):
|
|
import sys
|
|
import pytest
|
|
from scripts import import_publications
|
|
source = tmp_path / "input.csv"
|
|
source.write_text("unchanged", encoding="utf-8")
|
|
monkeypatch.setattr(sys, "argv", ["import_publications", str(source), "--report", str(source)])
|
|
with pytest.raises(ValueError):
|
|
import_publications.main()
|
|
assert source.read_text() == "unchanged"
|
|
|
|
|
|
def test_duplicate_rows_fill_late_metadata_and_report_conflicts():
|
|
rows = [(2, {**row(), "EID (Scopus)": "", "Квартиль scopus": ""}),
|
|
(3, {**row(), "EID (Scopus)": "new-id", "Квартиль scopus": "Q1"}),
|
|
(4, {**row(), "EID (Scopus)": "different-id", "Квартиль scopus": "Q2", "UT (WoS)": "late-wos"})]
|
|
actions, audit = plan(rows, EMPLOYEES, [])
|
|
assert len(actions) == 1 and actions[0]["result"] == "insert"
|
|
metadata = actions[0]["values"]["raw_data"]["csv_enrichment"]
|
|
assert metadata["EID (Scopus)"] == "new-id" and metadata["Квартиль scopus"] == "Q1"
|
|
assert metadata["UT (WoS)"] == "late-wos"
|
|
assert audit[1]["reason"] == "merged_csv_duplicate"
|
|
assert audit[2]["conflict_fields"] == ["EID (Scopus)", "Квартиль scopus"]
|
|
assert audit[2]["result"] == "skip"
|
|
repeated, repeated_audit = plan(rows, EMPLOYEES, [{"id": 11, **actions[0]["values"]}])
|
|
assert repeated == []
|
|
assert repeated_audit[2]["reason"] == "metadata_conflict_first_value_kept"
|
|
assert "different-id" not in str(audit)
|
|
|
|
|
|
def test_cli_apply_year_only_preserves_existing_json(tmp_path, monkeypatch):
|
|
import json
|
|
import sys
|
|
import psycopg
|
|
from scripts import import_publications
|
|
source = tmp_path / "input.csv"
|
|
fields = row()
|
|
source.write_text(";".join(fields) + "\n" + ";".join(fields.values()) + "\n", encoding="utf-8")
|
|
original_raw = {"hse": "keep", "csv_enrichment": fields.copy()}
|
|
stored = {"id": 11, "employee_id": 1, "title": "Test", "year": None,
|
|
"doi_url": "https://doi.org/10.1234/test", "raw_data": original_raw}
|
|
actions, _ = plan([(2, fields)], EMPLOYEES, [stored])
|
|
assert actions[0]["values"] == {"year": 2024}
|
|
|
|
class Result:
|
|
rowcount = 1
|
|
|
|
def __init__(self, rows):
|
|
self.rows = rows
|
|
|
|
def fetchall(self):
|
|
return self.rows
|
|
|
|
def fetchone(self):
|
|
return self.rows[0]
|
|
|
|
class Connection:
|
|
committed = False
|
|
updates = []
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, error, *_):
|
|
self.committed = error is None
|
|
|
|
def execute(self, query, params=None):
|
|
query = query if isinstance(query, str) else query.as_string()
|
|
if query.startswith("SELECT id, full_name"):
|
|
return Result(EMPLOYEES)
|
|
if query.startswith("SELECT id, employee_id"):
|
|
return Result([stored])
|
|
if "pg_try_advisory" in query:
|
|
return Result([{"locked": True}])
|
|
if query.startswith("UPDATE"):
|
|
self.updates.append((query, params))
|
|
return Result([])
|
|
|
|
connection = Connection()
|
|
report = tmp_path / "report.json"
|
|
monkeypatch.setattr(psycopg, "connect", lambda *a, **k: connection)
|
|
monkeypatch.setattr(sys, "argv", ["import_publications", str(source), "--report", str(report), "--apply"])
|
|
import_publications.main()
|
|
assert connection.committed and len(connection.updates) == 1
|
|
query, params = connection.updates[0]
|
|
assert '"year" = %s' in query and "raw_data" not in query
|
|
assert params == [2024, 11]
|
|
assert stored["raw_data"] == original_raw and stored["year"] is None
|
|
assert json.loads(report.read_text())["summary"]["counts"] == {"insert": 0, "update": 1, "skip": 0}
|