84 lines
3.6 KiB
Python
84 lines
3.6 KiB
Python
import hashlib
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from app.services import council_catalog as catalogs
|
|
|
|
|
|
@pytest.fixture()
|
|
def snapshot(monkeypatch, tmp_path):
|
|
data = json.loads(catalogs.CATALOGS[0].read_text(encoding="utf8"))
|
|
for specialty in data["specialties"]:
|
|
for passport in specialty["passports"]:
|
|
path = tmp_path / "passports" / (passport["id"] + ".pdf")
|
|
path.parent.mkdir(exist_ok=True)
|
|
path.write_bytes(b"test passport")
|
|
passport["local_pdf"] = str(path.relative_to(tmp_path))
|
|
passport["sha256"] = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
path = tmp_path / "catalog.json"
|
|
monkeypatch.setattr(catalogs, "DATA", tmp_path)
|
|
monkeypatch.setattr(catalogs, "CATALOGS", (path,))
|
|
def write():
|
|
path.write_text(json.dumps(data), encoding="utf8")
|
|
catalogs.catalog.cache_clear()
|
|
return catalogs.catalog()
|
|
yield data, write
|
|
catalogs.catalog.cache_clear()
|
|
|
|
|
|
@pytest.mark.parametrize("fault", ["schema", "duplicate_journal", "duplicate_link", "source", "passport", "point", "path", "hash", "quartile", "id"])
|
|
def test_damaged_snapshot_is_quarantined(snapshot, fault):
|
|
data, read = snapshot
|
|
if fault == "schema":
|
|
data["metadata"]["schema_version"] = "2.0.0"
|
|
elif fault == "duplicate_journal":
|
|
data["journals"].append(data["journals"][0])
|
|
elif fault == "duplicate_link":
|
|
data["specialty_links"].append(data["specialty_links"][0])
|
|
elif fault == "source":
|
|
data["journals"][0]["scopus"]["source_id"] = "missing"
|
|
elif fault == "passport":
|
|
data["specialty_links"][0]["passport_references"][0]["passport_id"] = "missing"
|
|
elif fault == "point":
|
|
data["specialty_links"][0]["passport_references"][0]["point_numbers"] = [99999]
|
|
elif fault == "path":
|
|
data["specialties"][0]["passports"][0]["local_pdf"] = "../../outside.pdf"
|
|
elif fault == "hash":
|
|
data["specialties"][0]["passports"][0]["sha256"] = "0" * 64
|
|
elif fault == "quartile":
|
|
data["journals"][0]["scopus"]["categories"][0]["quartile"] = "Q4"
|
|
elif fault == "id":
|
|
data["journals"][0]["scopus_source_id"] = 0
|
|
loaded = read()
|
|
assert loaded["issues"] and loaded["journals"] == {}
|
|
|
|
|
|
def test_q1_optional_q2_fields_are_not_required(snapshot):
|
|
data, read = snapshot
|
|
assert "selection_rule" not in data["metadata"]
|
|
loaded = read()
|
|
assert len(loaded["journals"]) == 107 and not loaded["issues"]
|
|
|
|
|
|
def test_unknown_journal_is_not_not_established_and_issn_conflicts():
|
|
records, issues = catalogs.publication_catalog(["0006-3835", "0020-0255"], [])
|
|
assert not records and issues
|
|
assert catalogs.publication_catalog([], ["A journal not in catalogs"], "1.3.8") == ([], [])
|
|
|
|
|
|
def test_overlap_preserves_both_records_and_best_q1_outside_q1_selection():
|
|
loaded = catalogs.catalog()
|
|
assert sum(len(records) == 2 for records in loaded["journals"].values()) == 2
|
|
q2_with_q1 = [r for records in loaded["journals"].values() for r in records
|
|
if r["catalog"] == "q2-journals" and r.get("also_meets_q1_rule") and not r.get("in_local_q1_catalog")]
|
|
assert len(q2_with_q1) == 3
|
|
assert all(any(m["quartile"] == "Q1" for m in r["metrics"]) for r in q2_with_q1)
|
|
|
|
|
|
def test_two_passports_keep_branch_and_point_numbering():
|
|
data = catalogs.catalog()
|
|
assert data["passports"]["1.3.8-technical"]["branch"] == "технические науки"
|
|
assert data["passports"]["1.3.8-physical-mathematical"]["branch"] == "физико-математические науки"
|
|
assert data["passports"]["1.3.8-technical"]["directions"] != data["passports"]["1.3.8-physical-mathematical"]["directions"]
|