300 lines
18 KiB
Python
300 lines
18 KiB
Python
from copy import deepcopy
|
||
from datetime import date
|
||
|
||
import pytest
|
||
from sqlalchemy import create_engine, inspect, text, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.db import _ensure_runtime_schema, Base
|
||
from app.models import EmployeePublication, Employee
|
||
from app.services import publication_quartiles as quartiles
|
||
from app.services.crawler import _apply_publication
|
||
from scripts.enrich_publication_quartiles import resolve_doi, load_sjr, main
|
||
|
||
|
||
@pytest.mark.parametrize("scopus,wos", [("Q1", "Q2"), ("q3", "Q4"), ("Q4", "")])
|
||
def test_hse_backfill_keeps_systems_q1_to_q4_and_unknown_metric_year(scopus, wos):
|
||
raw = {"other": "preserve", "csv_enrichment": {"Квартиль scopus": scopus, "Квартиль wos": wos, "Год": "2024"}}
|
||
before = deepcopy(raw)
|
||
result = quartiles.hse_entries(raw, 2024)
|
||
assert raw == before
|
||
assert [e["quartile"] for e in result] == [v.upper() for v in (scopus, wos) if v]
|
||
assert all(e["metric_year"] is None and e["publication_year"] == 2024 for e in result)
|
||
assert [e["system"] for e in result] == (["Scopus", "Web of Science"] if wos else ["Scopus"])
|
||
|
||
|
||
@pytest.mark.parametrize("value", ["Q5", "1", "", None])
|
||
def test_invalid_hse_value_does_not_invent_a_quartile(value):
|
||
assert quartiles.hse_entries({"csv_enrichment": {"Квартиль scopus": value}}, 2024) == []
|
||
|
||
|
||
def test_merge_is_idempotent_and_preserves_conflicting_evidence():
|
||
first = quartiles.hse_entries({"csv_enrichment": {"Квартиль scopus": "Q1", "DOI": "10.1234/work"}}, 2024)[0]
|
||
payload = quartiles.merge(None, [first])
|
||
assert quartiles.merge(payload, [{**first, "checked_on": "2099-01-01"}]) == payload
|
||
assert len(quartiles.merge(payload, [{**first, "quartile": "Q2"}])["entries"]) == 2
|
||
assert quartiles.labels(payload, 2024, doi="10.1234/work") == ["Scopus Q1"]
|
||
assert quartiles.labels(payload, 2023) == []
|
||
|
||
|
||
def test_catalog_metric_requires_exact_year_unless_latest_explicitly_allowed(monkeypatch):
|
||
monkeypatch.setattr(quartiles, "publication_catalog", lambda ids, names: ([{
|
||
"name": "Journal", "metrics": [{"year": 2025, "system": "Scopus", "quartile": "Q1", "metric": "CiteScore",
|
||
"category": "Mathematics", "source": "Snapshot", "source_url": "https://example.org/metrics"}]}], []))
|
||
metadata = {"type": "journal-article", "years": [2024], "issns": ["0006-3835"]}
|
||
assert quartiles.catalog_entries(metadata, 2024)[0] == []
|
||
found, status = quartiles.catalog_entries(metadata, 2024, True)
|
||
assert status == "found_latest" and found[0]["metric_year"] == 2025 and found[0]["publication_year"] == 2024
|
||
assert quartiles.labels(quartiles.merge(None, found), 2024) == []
|
||
assert quartiles.catalog_entries(metadata, 2023, True)[1] == "publication_year_conflict"
|
||
assert quartiles.catalog_entries({**metadata, "type": "book-chapter"}, 2024, True)[1] == "not_journal_article"
|
||
|
||
|
||
def sjr_record(year, categories="Analysis (Q1); Applied Mathematics (Q2)"):
|
||
return {"journal": "Journal", "issns": ["0006-3835"], "quartile": "Q1", "categories": categories,
|
||
"source_url": f"https://opendata.unex.es/recurso/ciencia-tecnologia/investigacion/revistas/Revista/{year}-1"}
|
||
|
||
|
||
def test_sjr_keeps_historical_year_categories_and_system_distinct():
|
||
metadata = {"type": "journal-article", "years": [2020], "issns": ["0006-3835"]}
|
||
found, status = quartiles.sjr_entries(metadata, 2020, quartiles.index_sjr({2020: [sjr_record(2020)], 2025: [sjr_record(2025)]}), True)
|
||
assert status == "found" and [e["quartile"] for e in found] == ["Q1", "Q2"]
|
||
assert all(e["system"] == "SCImago" and e["metric_year"] == 2020 for e in found)
|
||
assert quartiles.labels(quartiles.merge(None, found), 2020) == []
|
||
latest, status = quartiles.sjr_entries(metadata, 2020, quartiles.index_sjr({2025: [sjr_record(2025)]}), True)
|
||
assert status == "found_latest" and latest[0]["metric_year"] == 2025
|
||
assert quartiles.sjr_entries(metadata, 2020, quartiles.index_sjr({2025: [sjr_record(2025)]}))[0] == []
|
||
|
||
|
||
def test_sjr_identity_and_category_conflicts_do_not_resolve_by_order():
|
||
metadata = {"type": "journal-article", "years": [2020], "issns": ["0006-3835"]}
|
||
first = sjr_record(2020)
|
||
second = {**first, "source_url": first["source_url"] + "2"}
|
||
assert quartiles.sjr_entries(metadata, 2020, quartiles.index_sjr({2020: [first, second]}))[1] == "journal_identity_conflict"
|
||
assert quartiles.sjr_entries(metadata, 2020, quartiles.index_sjr({2020: [sjr_record(2020, "Analysis (Q3)")]} ))[1] == "ranking_category_conflict"
|
||
|
||
|
||
def test_schema_addition_is_repeatable_and_preserves_existing_data(monkeypatch):
|
||
engine = create_engine("sqlite://")
|
||
with engine.begin() as connection:
|
||
connection.execute(text("CREATE TABLE employee_publications (id INTEGER PRIMARY KEY, title TEXT NOT NULL)"))
|
||
connection.execute(text("INSERT INTO employee_publications (id,title) VALUES (1,'Keep')"))
|
||
monkeypatch.setattr("app.db.engine", engine)
|
||
_ensure_runtime_schema()
|
||
_ensure_runtime_schema()
|
||
assert "quartiles" in {c["name"] for c in inspect(engine).get_columns("employee_publications")}
|
||
with engine.connect() as connection:
|
||
assert connection.execute(text("SELECT title,quartiles FROM employee_publications")).one() == ("Keep", None)
|
||
|
||
|
||
def test_crawler_preserves_quartiles_while_replacing_source_owned_raw_data():
|
||
payload = quartiles.merge(None, quartiles.hse_entries({"csv_enrichment": {"Квартиль scopus": "Q4"}}, 2024))
|
||
publication = EmployeePublication(title="Old", source_hash="old", quartiles=deepcopy(payload))
|
||
publication.vak_categories = {"schema_version": 1, "entries": [{"category": "К1"}], "status": "found"}
|
||
_apply_publication(publication, {"id": "42", "title": "Updated", "year": 2024, "raw_data": {"title": "Updated"}}, "new")
|
||
assert publication.quartiles == payload
|
||
assert publication.vak_categories["entries"] == [{"category": "К1"}]
|
||
assert publication.title == "Updated" and publication.source_hash == "new"
|
||
|
||
|
||
def test_council_reads_new_field_without_raw_enrichment_and_keeps_historical_filter():
|
||
from app.services.council import _database_article
|
||
values = quartiles.hse_entries({"csv_enrichment": {"Квартиль scopus": "Q2", "DOI": "10.1234/work"}}, 2024)
|
||
publication = EmployeePublication(title="Work", year=2024, publication_type="ARTICLE", source_hash="hash",
|
||
quartiles=quartiles.merge(None, values), raw_data={}, doi_url="https://doi.org/10.1234/work")
|
||
assert _database_article(publication)["quartiles"] == ["Scopus Q2"]
|
||
publication.quartiles = quartiles.merge(None, [{**values[0], "metric_year": 2025}])
|
||
assert _database_article(publication)["quartiles"] == []
|
||
|
||
|
||
def test_doi_cache_uses_verified_identity_and_does_not_need_network_on_repeat(monkeypatch, tmp_path):
|
||
calls = []
|
||
class Response:
|
||
status_code = 200
|
||
def raise_for_status(self):
|
||
pass
|
||
def json(self):
|
||
return {"message": {"DOI": "10.1234/work", "type": "journal-article", "container-title": ["Journal"],
|
||
"ISSN": ["0006-3835"], "published": {"date-parts": [[2024, 2, 1]]}}}
|
||
def get(url, **kwargs):
|
||
calls.append(url)
|
||
assert kwargs["allow_redirects"] is False
|
||
return Response()
|
||
monkeypatch.setattr("scripts.enrich_publication_quartiles.requests.get", get)
|
||
result = resolve_doi("10.1234/work", tmp_path)
|
||
assert result["status"] == "resolved" and result["years"] == [2024]
|
||
assert resolve_doi("10.1234/work", tmp_path) == result and len(calls) == 1
|
||
assert calls[0] == "https://api.crossref.org/works/10.1234%2Fwork"
|
||
|
||
|
||
def test_sjr_source_combines_multiple_categories_of_the_same_resource(monkeypatch, tmp_path):
|
||
url = "https://opendata.unex.es/recurso/ciencia-tecnologia/investigacion/revistas/Revista/2020-1"
|
||
def binding(category, quartile="Q1"):
|
||
return {key: {"value": value} for key, value in {
|
||
"uri": url, "title": "Journal", "issn": "0006-3835", "quartile": quartile, "categories": category}.items()}
|
||
class Response:
|
||
def raise_for_status(self):
|
||
pass
|
||
def json(self):
|
||
return {"results": {"bindings": [binding("Analysis (Q1)"), binding("Applied Mathematics (Q2)")]}}
|
||
monkeypatch.setattr("scripts.enrich_publication_quartiles.requests.get", lambda *args, **kwargs: Response())
|
||
rankings, errors = load_sjr(["0006-3835"], [2020], tmp_path)
|
||
assert not errors
|
||
metadata = {"type": "journal-article", "years": [2020], "issns": ["0006-3835"]}
|
||
found, status = quartiles.sjr_entries(metadata, 2020, rankings)
|
||
assert status == "found" and {e["quartile"] for e in found} == {"Q1", "Q2"}
|
||
|
||
|
||
def test_doi_identity_conflict_does_not_supply_journal_metrics(monkeypatch, tmp_path):
|
||
class Response:
|
||
status_code = 200
|
||
def raise_for_status(self):
|
||
pass
|
||
def json(self):
|
||
return {"message": {"DOI": "10.1234/another", "type": "journal-article", "ISSN": ["0006-3835"]}}
|
||
monkeypatch.setattr("scripts.enrich_publication_quartiles.requests.get", lambda *args, **kwargs: Response())
|
||
result = resolve_doi("10.1234/work", tmp_path)
|
||
assert result["status"] == "doi_identity_conflict" and "issns" not in result
|
||
assert not list(tmp_path.glob("*.json"))
|
||
|
||
|
||
def test_enrichment_cli_dry_run_apply_and_repeat_preserve_original_data(monkeypatch, tmp_path):
|
||
url = "sqlite:///" + str(tmp_path / "publications.db")
|
||
engine = create_engine(url)
|
||
Base.metadata.create_all(engine)
|
||
raw = {"original": "keep", "csv_enrichment": {"Квартиль scopus": "Q3", "Год": "2024"}}
|
||
with Session(engine) as session:
|
||
employee = Employee(profile_key="test", canonical_url="https://example.org/person")
|
||
session.add(employee)
|
||
session.flush()
|
||
session.add(EmployeePublication(employee_id=employee.id, title="Work", year=2024, source_hash="hash", raw_data=raw))
|
||
session.commit()
|
||
monkeypatch.setenv("DATABASE_URL", url)
|
||
arguments = ["enrich", "--cache-dir", str(tmp_path / "cache"), "--report", str(tmp_path / "report.json")]
|
||
monkeypatch.setattr("sys.argv", arguments)
|
||
main()
|
||
with Session(engine) as session:
|
||
assert session.scalar(select(EmployeePublication)).quartiles is None
|
||
monkeypatch.setattr("sys.argv", arguments + ["--apply"])
|
||
main()
|
||
with Session(engine) as session:
|
||
publication = session.scalar(select(EmployeePublication))
|
||
payload = deepcopy(publication.quartiles)
|
||
assert publication.raw_data == raw and payload["entries"][0]["quartile"] == "Q3"
|
||
main()
|
||
with Session(engine) as session:
|
||
publication = session.scalar(select(EmployeePublication))
|
||
assert publication.quartiles == payload and publication.raw_data == raw
|
||
|
||
|
||
def test_issn_succession_requires_explicit_relation_and_unique_doi_journal_title():
|
||
old = {"history": "Continued as", "related": ["Physical Review B"]}
|
||
new = {"issns": {"2469-9950", "2469-9969"}}
|
||
metadata = {"issns": ["1098-0121", "1550-235X"], "journals": ["Physical Review B"]}
|
||
result = quartiles.with_title_history(metadata, {"10980121": [old]}, {"physical review b": [new]}, "https://example.org/source-titles")
|
||
assert result["issns"] == metadata["issns"] and "2469-9950" in result["ranking_issns"]
|
||
assert metadata == {"issns": ["1098-0121", "1550-235X"], "journals": ["Physical Review B"]}
|
||
assert quartiles.with_title_history(metadata, {"10980121": [old]}, {"physical review b": [new, new]}, "url") == metadata
|
||
assert quartiles.with_title_history({**metadata, "journals": ["Other journal"]}, {"10980121": [old]}, {"physical review b": [new]}, "url") == {**metadata, "journals": ["Other journal"]}
|
||
|
||
|
||
def test_crossref_not_found_falls_back_to_datacite_journal_metadata(monkeypatch, tmp_path):
|
||
class Response:
|
||
def __init__(self, status):
|
||
self.status_code = status
|
||
def raise_for_status(self):
|
||
pass
|
||
def json(self):
|
||
return {"data": {"attributes": {"doi": "10.1234/work", "publicationYear": 2024,
|
||
"types": {"resourceTypeGeneral": "JournalArticle"},
|
||
"container": {"title": "Journal", "identifierType": "ISSN", "identifier": "0006-3835"}}}}
|
||
calls = []
|
||
def get(url, **kwargs):
|
||
calls.append(url)
|
||
return Response(404 if "crossref.org" in url else 200)
|
||
monkeypatch.setattr("scripts.enrich_publication_quartiles.requests.get", get)
|
||
result = resolve_doi("10.1234/work", tmp_path)
|
||
assert result["status"] == "resolved" and result["type"] == "journal-article"
|
||
assert result["issns"] == ["0006-3835"] and result["years"] == [2024]
|
||
assert result["datacite_checked"] and len(calls) == 2
|
||
|
||
|
||
@pytest.mark.parametrize("new_doi,new_journal", [("10.1234/new", "Different journal"), ("10.1234/old", "Different journal")])
|
||
def test_crawler_doi_or_journal_change_excludes_preserved_quartile(new_doi, new_journal):
|
||
from app.services.council import _database_article
|
||
evidence = {"system": "Scopus", "quartile": "Q1", "publication_year": 2024, "metric_year": 2024,
|
||
"journal": "Old journal", "issns": [], "source": "Checked"}
|
||
payload = {"schema_version": 1, "entries": [evidence], "lookup": {"doi": "10.1234/old"}}
|
||
publication = EmployeePublication(title="Work", year=2024, source_hash="old", publication_type="ARTICLE",
|
||
doi_url="https://doi.org/10.1234/old", quartiles=deepcopy(payload), raw_data={"journal": "Old journal"})
|
||
assert _database_article(publication)["quartiles"] == ["Scopus Q1"]
|
||
_apply_publication(publication, {"id": "42", "title": "Work", "year": 2024, "type": "ARTICLE",
|
||
"doi_url": "https://doi.org/" + new_doi, "raw_data": {"journal": new_journal}}, "new")
|
||
article = _database_article(publication)
|
||
assert publication.quartiles == payload
|
||
assert article["quartiles"] == [] and article["quartile_issues"]
|
||
assert all("перепроверка" in issue or "Уточните журнал статьи" in issue for issue in article["quartile_issues"])
|
||
|
||
|
||
def test_updating_lookup_freezes_legacy_binding_instead_of_rebinding_old_entries():
|
||
old = {"system": "Scopus", "quartile": "Q1", "publication_year": 2024, "metric_year": 2024,
|
||
"journal": "Old journal", "issns": [], "source": "Checked"}
|
||
payload = {"schema_version": 1, "entries": [old], "lookup": {"doi": "10.1234/old"}}
|
||
updated = quartiles.merge(payload, [], lookup={"doi": "10.1234/new"})
|
||
assert updated["entries"][0]["doi"] == "10.1234/old"
|
||
assert quartiles.labels(updated, 2024, "10.1234/new", ["Old journal"]) == []
|
||
fresh = {**old, "doi": "10.1234/new", "quartile": "Q2"}
|
||
refreshed = quartiles.merge(updated, [fresh], lookup={"doi": "10.1234/new"})
|
||
assert quartiles.labels(refreshed, 2024, "10.1234/new", ["Old journal"]) == ["Scopus Q2"]
|
||
assert len(refreshed["entries"]) == 2
|
||
|
||
|
||
def test_unbound_hse_fallback_does_not_bypass_quartile_identity_checks():
|
||
from app.services.council import _database_article
|
||
publication = EmployeePublication(title="Work", year=2024, source_hash="hash", publication_type="ARTICLE",
|
||
doi_url="10.1234/new", raw_data={"journal": "New journal", "csv_enrichment": {
|
||
"DOI": "10.1234/old", "Название журнала": "Old journal", "Квартиль scopus": "Q1"}})
|
||
article = _database_article(publication)
|
||
assert article["quartiles"] == [] and article["quartile_issues"]
|
||
assert "Old journal" not in article["journals"]
|
||
|
||
|
||
def test_journal_issn_allows_explicit_rename_but_conflicting_issn_is_rejected():
|
||
value = {"system": "Scopus", "quartile": "Q1", "publication_year": 2024, "metric_year": 2024,
|
||
"doi": "10.1234/work", "journal": "Old title", "issns": ["0006-3835"]}
|
||
payload = quartiles.merge(None, [value])
|
||
assert quartiles.labels(payload, 2024, "10.1234/work", ["Renamed title"], ["00063835"]) == ["Scopus Q1"]
|
||
assert quartiles.labels(payload, 2024, "10.1234/work", ["Old title"], ["11111111"]) == []
|
||
|
||
|
||
def test_default_enrichment_rechecks_filled_evidence_after_doi_change(monkeypatch, tmp_path):
|
||
url = "sqlite:///" + str(tmp_path / "stale.db")
|
||
engine = create_engine(url)
|
||
Base.metadata.create_all(engine)
|
||
old = {"system": "Scopus", "quartile": "Q1", "publication_year": 2024, "metric_year": 2024,
|
||
"journal": "Old journal", "issns": [], "source": "Checked"}
|
||
payload = {"schema_version": 1, "entries": [old], "lookup": {"doi": "10.1234/old", "status": "found"}}
|
||
with Session(engine) as session:
|
||
person = Employee(profile_key="test", canonical_url="https://example.org/person")
|
||
session.add(person)
|
||
session.flush()
|
||
session.add(EmployeePublication(employee_id=person.id, title="Work", year=2024, source_hash="hash",
|
||
doi_url="10.1234/new", raw_data={"journal": "New journal"}, quartiles=payload))
|
||
session.commit()
|
||
calls = []
|
||
def resolve(doi, cache):
|
||
calls.append(doi)
|
||
return {"status": "resolved", "doi": doi, "type": "journal-article", "journals": ["New journal"]}
|
||
monkeypatch.setattr("scripts.enrich_publication_quartiles.resolve_doi", resolve)
|
||
monkeypatch.setattr("scripts.enrich_publication_quartiles.catalog_entries", lambda metadata, year, latest: ([
|
||
{**old, "doi": metadata["doi"], "journal": "New journal", "quartile": "Q2"}], "found"))
|
||
monkeypatch.setenv("DATABASE_URL", url)
|
||
monkeypatch.setattr("sys.argv", ["enrich", "--apply", "--lookup-doi", "--cache-dir", str(tmp_path / "cache"),
|
||
"--report", str(tmp_path / "report.json")])
|
||
main()
|
||
with Session(engine) as session:
|
||
stored = session.scalar(select(EmployeePublication)).quartiles
|
||
assert calls == ["10.1234/new"] and len(stored["entries"]) == 2
|
||
assert stored["entries"][0]["doi"] == "10.1234/old"
|
||
assert quartiles.labels(stored, 2024, "10.1234/new", ["New journal"]) == ["Scopus Q2"]
|