752 lines
48 KiB
Python
752 lines
48 KiB
Python
import csv
|
||
from datetime import date
|
||
from io import BytesIO
|
||
from zipfile import ZipFile
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from openpyxl import Workbook, load_workbook
|
||
from openpyxl.worksheet.worksheet import Worksheet
|
||
from sqlalchemy.exc import OperationalError
|
||
|
||
from app.db import get_db
|
||
from app.main import app
|
||
from app.models import Employee, EmployeePublication
|
||
from app.services import council
|
||
from app.services.council_export import _append, council_xlsx
|
||
|
||
|
||
def selection_results(db, result):
|
||
branches = list(dict.fromkeys(b for item in result["specialties"] for b in item["branches"]))
|
||
return [result if b == result["branch"] and not result["q"] and not result["specialty"] else
|
||
council.council_candidates(db, result["end"], sort=result["sort"], direction=result["direction"], paginate=False, branch=b)
|
||
for b in branches]
|
||
|
||
|
||
@pytest.fixture()
|
||
def export(monkeypatch, tmp_path):
|
||
path = tmp_path / "articles.csv"
|
||
prepared = tmp_path / "articles.json"
|
||
def write(rows):
|
||
fields = ["Название публикации", "Год", "DOI", "EID (Scopus)", "UT (WoS)", "Название журнала",
|
||
"Авторы НИУ ВШЭ", "Квартиль scopus", "Квартиль wos"]
|
||
with path.open("w", encoding="utf-8-sig", newline="") as output:
|
||
writer = csv.DictWriter(output, fieldnames=fields, delimiter=";")
|
||
writer.writeheader()
|
||
writer.writerows(rows)
|
||
from app.services.council_source import prepare_reference
|
||
import json
|
||
prepared.write_text(json.dumps(prepare_reference(path), ensure_ascii=False), encoding="utf-8")
|
||
council._articles_by_author.cache_clear()
|
||
monkeypatch.setattr(council, "EXPORT", prepared)
|
||
write([])
|
||
yield write
|
||
council._articles_by_author.cache_clear()
|
||
|
||
|
||
def employee(db, name="Иванов Иван Иванович", degree="Доктор технических наук", status="active", key="doctor"):
|
||
person = Employee(profile_key=key, canonical_url="https://www.hse.ru/staff/test", full_name=name,
|
||
status=status, current_data={"contacts": {"emails": ["private@example.org"]},
|
||
"sections": [{"title": "Учёные степени", "items": [degree] if degree else []}]})
|
||
db.add(person)
|
||
db.flush()
|
||
return person
|
||
|
||
|
||
def publication(db, person, title="Shared work", doi=None, year=2024, kind=None, **extra):
|
||
item = EmployeePublication(employee_id=person.id, title=title, year=year, doi_url=doi,
|
||
publication_type=kind, source_hash=f"hash-{len(db.new)}-{title}-{doi}", **extra)
|
||
db.add(item)
|
||
return item
|
||
|
||
|
||
def row(title="Shared work", doi="10.1234/shared", year="2024", author="Иванов Иван Иванович", **extra):
|
||
return {"Название публикации": title, "Год": year, "DOI": doi, "Авторы НИУ ВШЭ": author,
|
||
"Квартиль scopus": "Q1", **extra}
|
||
|
||
|
||
def test_council_union_counts_provenance_and_types(db_session, export):
|
||
export([row(), row(**{"Квартиль scopus": "", "Квартиль wos": "Q2"}),
|
||
row(title="CSV only", doi="", year="2022"),
|
||
row(title="Not eligible", doi="10.1234/old", year="2019")])
|
||
person = employee(db_session, degree="Доктор технических наук")
|
||
publication(db_session, person, doi="https://dx.doi.org/10.1234/SHARED", kind="ARTICLE",
|
||
publication_id="100", url="https://publications.hse.ru/view/100")
|
||
publication(db_session, person, title=" shared WORK ", kind="ARTICLE")
|
||
publication(db_session, person, title="Book only", kind="BOOK", year=2020)
|
||
publication(db_session, person, title="Imported", doi="https://doi.org/10.1234/imported",
|
||
raw_data={"csv_enrichment": {"Квартиль wos": "Q2", "DOI": "10.1234/imported"}})
|
||
db_session.commit()
|
||
result = council.council_candidates(db_session, date(2026, 9, 30))
|
||
assert result["total"] == 1
|
||
candidate = result["candidates"][0]
|
||
assert (candidate["total_count"], candidate["db_count"], candidate["count"]) == (3, 2, 5)
|
||
shared = next(a for a in candidate["articles"] if a["doi"] == "10.1234/shared")
|
||
assert shared["type_label"] == "Статья"
|
||
assert shared["quartiles"] == ["Scopus Q1", "Web of Science Q2"]
|
||
assert shared["rows"] == [2, 3]
|
||
assert "Данные ВШЭ" in shared["origins"] and "ЛК сотрудника ВШЭ" in shared["origins"]
|
||
imported = next(a for a in candidate["articles"] if a["title"] == "Imported")
|
||
assert imported["origins"] == ["Данные ВШЭ"]
|
||
assert imported["qualifies"] and imported["q2"]
|
||
assert "Тип не указан" in imported["type_label"]
|
||
assert council.period(date(2024, 2, 29))[0] == date(2020, 1, 1)
|
||
|
||
|
||
def test_all_degree_levels_and_missing_publications_remain_visible(db_session, export):
|
||
export([row(author="Доктор", title="Single qualifying"),
|
||
row(author="Без степени", title="No degree"),
|
||
row(author="Кандидат", title="Candidate work"),
|
||
row(author="PhD", title="PhD work"),
|
||
row(author="Две степени", title="Mixed degrees", **{"Квартиль scopus": "Q3"}),
|
||
row(author="Уволен", title="Old", year="2010")])
|
||
employee(db_session, "Доктор", key="csv-only")
|
||
for name, degree in [("Без степени", None), ("Кандидат", "Кандидат технических наук"), ("PhD", "Ph.D.")]:
|
||
person = employee(db_session, name, degree, key=name)
|
||
publication(db_session, person)
|
||
employee(db_session, "Две степени", "Кандидат технических наук; Доктор физико-математических наук", key="mixed")
|
||
employee(db_session, "Уволен", status="dismissed", key="dismissed")
|
||
only_db = employee(db_session, "Только БД", key="database")
|
||
publication(db_session, only_db, kind="PATENT")
|
||
employee(db_session, "Без публикаций", key="empty")
|
||
db_session.commit()
|
||
items = council.council_candidates(db_session, date(2026, 9, 30))["candidates"]
|
||
assert {c["name"] for c in items} == {"Доктор", "Две степени", "Уволен", "Только БД", "Кандидат", "PhD", "Без публикаций"}
|
||
assert next(c for c in items if c["name"] == "Кандидат")["minimum"] == 20
|
||
assert next(c for c in items if c["name"] == "PhD")["clause"] == 6
|
||
assert next(c for c in items if c["name"] == "Без публикаций")["assessment_status"] == "Требует проверки"
|
||
assert next(c for c in items if c["name"] == "Доктор")["total_count"] == 1
|
||
assert next(c for c in items if c["name"] == "Только БД")["db_count"] == 1
|
||
assert next(c for c in items if c["name"] == "Две степени")["total_count"] == 0
|
||
assert next(c for c in items if c["name"] == "Уволен")["dismissed"]
|
||
|
||
|
||
def test_no_false_merge_when_title_has_conflicting_dois(db_session, export):
|
||
export([row(doi="10.1234/one"), row(doi="10.1234/two"), row(doi="")])
|
||
person = employee(db_session)
|
||
publication(db_session, person, title="Renamed work", doi="https://doi.org/10.1234/one")
|
||
db_session.commit()
|
||
item = council.council_candidates(db_session, date(2026, 9, 30))["candidates"][0]
|
||
assert item["count"] == 3
|
||
assert item["total_count"] == 3 and item["db_count"] == 1
|
||
|
||
|
||
def test_csv_same_name_does_not_attribute_to_wrong_employee(db_session, export):
|
||
export([row()])
|
||
for key in ["one", "two"]:
|
||
person = employee(db_session, key=key)
|
||
publication(db_session, person, title=key)
|
||
db_session.commit()
|
||
items = council.council_candidates(db_session, date(2026, 9, 30))["candidates"]
|
||
assert len(items) == 2
|
||
assert all(c["ambiguous"] and c["total_count"] == 0 and c["count"] == 1 for c in items)
|
||
|
||
|
||
def test_missing_database_year_merges_only_unambiguous_title_and_non_article_excluded(db_session, export):
|
||
export([row(title="Known book"), row(title="Article", doi="10.1234/article")])
|
||
person = employee(db_session)
|
||
publication(db_session, person, title="Known book", year=None, kind="BOOK")
|
||
publication(db_session, person, title="Article", year=None, kind="ARTICLE")
|
||
db_session.commit()
|
||
candidate = council.council_candidates(db_session, date(2026, 9, 30))["candidates"][0]
|
||
assert (candidate["count"], candidate["total_count"], candidate["db_count"]) == (2, 1, 2)
|
||
book = next(a for a in candidate["articles"] if a["type_label"] == "Книга")
|
||
assert not book["qualifies"]
|
||
assert "не включён в счётчики Q1/Q2/К1" in book["criteria"][-1]
|
||
|
||
|
||
def test_search_and_pagination_include_all_workers(db_session, export):
|
||
for number in range(28):
|
||
person = employee(db_session, f"Петров {number:02d}", key=f"person-{number}")
|
||
publication(db_session, person, title=f"Work {number}")
|
||
db_session.commit()
|
||
first = council.council_candidates(db_session)
|
||
last = council.council_candidates(db_session, offset=25)
|
||
assert first["total"] == 28 and len(first["candidates"]) == 25
|
||
assert len(last["candidates"]) == 3
|
||
assert {c["name"] for c in first["candidates"]}.isdisjoint(c["name"] for c in last["candidates"])
|
||
assert council.council_candidates(db_session, q="петров 27")["matched"] == 1
|
||
assert council.council_candidates(db_session, offset=-1)["offset"] == 0
|
||
assert council.council_candidates(db_session, offset=999)["offset"] == 25
|
||
descending = council.council_candidates(db_session, sort="full_name", direction="desc")
|
||
last_descending = council.council_candidates(db_session, sort="full_name", direction="desc", offset=25)
|
||
assert descending["candidates"][0]["name"] == "Петров 27"
|
||
assert [c["name"] for c in last_descending["candidates"]] == ["Петров 02", "Петров 01", "Петров 00"]
|
||
|
||
|
||
def test_sort_by_name_and_each_unique_publication_count(db_session, export):
|
||
export([row(author="Анна", title=f"CSV {number}", doi=f"10.1234/csv-{number}") for number in range(3)]
|
||
+ [row(author="Борис", title="CSV Boris", doi="10.1234/boris")])
|
||
anna = employee(db_session, "Анна", key="anna")
|
||
publication(db_session, anna, title="CSV 0", doi="https://doi.org/10.1234/csv-0")
|
||
for name, count in [("Борис", 2), ("Вера", 4)]:
|
||
person = employee(db_session, name, key=name)
|
||
for number in range(count):
|
||
publication(db_session, person, title=f"DB {name} {number}")
|
||
db_session.commit()
|
||
expected = {
|
||
"full_name": {"asc": ["Анна", "Борис", "Вера"], "desc": ["Вера", "Борис", "Анна"]},
|
||
"total_count": {"asc": ["Вера", "Борис", "Анна"], "desc": ["Анна", "Борис", "Вера"]},
|
||
"db_count": {"asc": ["Анна", "Борис", "Вера"], "desc": ["Вера", "Борис", "Анна"]},
|
||
"count": {"asc": ["Анна", "Борис", "Вера"], "desc": ["Вера", "Анна", "Борис"]},
|
||
}
|
||
for sort, directions in expected.items():
|
||
for direction, names in directions.items():
|
||
result = council.council_candidates(db_session, sort=sort, direction=direction)
|
||
assert [c["name"] for c in result["candidates"]] == names
|
||
invalid = council.council_candidates(db_session, sort="unknown", direction="unknown")
|
||
assert (invalid["sort"], invalid["direction"]) == ("total_count", "desc")
|
||
assert [c["name"] for c in invalid["candidates"]] == expected["total_count"]["desc"]
|
||
app.dependency_overrides[get_db] = lambda: db_session
|
||
try:
|
||
response = TestClient(app).get("/admin/council?sort=full_name&direction=desc")
|
||
assert response.status_code == 200
|
||
assert response.text.index("Вера") < response.text.index("Борис") < response.text.index("Анна")
|
||
assert '<option value="full_name" selected>' in response.text
|
||
assert '<option value="desc" selected>' in response.text
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
def test_public_page_data_minimization_and_safe_links(db_session, export):
|
||
export([row(title='<script>alert("csv")</script>', doi="javascript:bad")])
|
||
person = employee(db_session)
|
||
person.canonical_url = "javascript:bad"
|
||
publication(db_session, person, url="javascript:alert(1)", doi="javascript:bad", kind="BOOK")
|
||
db_session.commit()
|
||
app.dependency_overrides[get_db] = lambda: db_session
|
||
try:
|
||
client = TestClient(app)
|
||
response = client.get("/admin/council", follow_redirects=False)
|
||
assert response.status_code == 200
|
||
assert "Иванов Иван Иванович" in response.text and "Доктор технических наук" in response.text
|
||
assert "Используется высшая обнаруженная степень; её документальное подтверждение проверяется отдельно." in response.text
|
||
assert "CSV по критериям" not in response.text and "ВСЕГО" in response.text and "Все публикации" in response.text
|
||
assert 'class="council__article"' not in response.text
|
||
fragment = client.get(f"/admin/council/employees/{person.id}/articles")
|
||
assert fragment.status_code == 200
|
||
assert "Книга" in fragment.text and "не указан тип материала" in fragment.text
|
||
assert "<script>" in fragment.text and "<script>" not in fragment.text
|
||
assert "javascript:" not in fragment.text and "private@example.org" not in fragment.text
|
||
assert "javascript:" not in response.text and "private@example.org" not in response.text
|
||
assert "/admin/employees/" not in response.text and "/admin/logout" not in response.text
|
||
assert "Backend 2.0.0 · Frontend 2.0.0" in response.text
|
||
assert "Вход в систему" in response.text
|
||
assert client.get("/admin/directory", follow_redirects=False).status_code in (302, 303, 401, 403)
|
||
assert "Сотрудники не найдены" in client.get("/admin/council?q=missing").text
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
def test_public_error_state_missing_export_and_database_failure(db_session, export, monkeypatch):
|
||
app.dependency_overrides[get_db] = lambda: db_session
|
||
try:
|
||
council.EXPORT.unlink()
|
||
response = TestClient(app).get("/admin/council")
|
||
assert response.status_code == 503 and "Повторить загрузку" in response.text
|
||
def fail(*args, **kwargs):
|
||
raise OperationalError("query", {}, Exception("private error"))
|
||
monkeypatch.setattr("app.admin.council_candidates", fail)
|
||
response = TestClient(app).get("/admin/council")
|
||
assert response.status_code == 503 and "private error" not in response.text
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
def test_xlsx_exports_all_filtered_workers_in_order_and_deduplicated_evidence(db_session, export):
|
||
export([row(author="Петров 27"), row(author="Петров 27", **{"Квартиль wos": "Q2"})])
|
||
for number in range(28):
|
||
person = employee(db_session, f"Петров {number:02d}", key=f"xlsx-{number}")
|
||
publication(db_session, person, title="Shared work" if number == 27 else f"Work {number}",
|
||
doi="https://doi.org/10.1234/shared" if number == 27 else None,
|
||
kind="ARTICLE", publication_id=str(number), url=f"https://publications.hse.ru/view/{number}")
|
||
excluded = employee(db_session, "Без степени", degree=None, key="excluded")
|
||
publication(db_session, excluded)
|
||
db_session.commit()
|
||
app.dependency_overrides[get_db] = lambda: db_session
|
||
try:
|
||
client = TestClient(app)
|
||
response = client.get("/admin/council/export.xlsx?q=Петров&sort=full_name&direction=desc&offset=25")
|
||
assert response.status_code == 200
|
||
assert response.headers["content-type"] == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||
assert response.headers["content-disposition"] == 'attachment; filename="council.xlsx"'
|
||
book = load_workbook(BytesIO(response.content))
|
||
assert book.sheetnames == ["Подбор диссовета", "Сотрудники", "Публикации", "Параметры", "_Данные подбора", "_Справочники подбора"]
|
||
people, works = book["Сотрудники"], book["Публикации"]
|
||
assert [cell.value for cell in people[1]] == [
|
||
"№", "ФИО", "Научная степень", "Q1", "Q2", "К1", "К2", "Всего", "Всего из ЛК",
|
||
"Общее количество всех публикаций без дублей", "Специальности ВАК", "Профиль ВШЭ"
|
||
]
|
||
assert [cell.value for cell in works[1]] == [
|
||
"№", "ФИО", "Название", "Год", "Тип материала", "Критерии", "Q1", "Q2", "К1", "К2",
|
||
"RSCI", "Специальность ВАК", "DOI", "Ссылка на сайт"
|
||
]
|
||
assert people["L2"].value == "https://www.hse.ru/staff/test"
|
||
assert people.max_column == 12 and works.max_column == 14
|
||
assert people.max_row == 29 and works.max_row == 29
|
||
assert people["B2"].value == "Петров 27" and people["B29"].value == "Петров 00"
|
||
assert [people.cell(2, n).value for n in range(4, 11)] == [1, 1, 0, 0, 1, 1, 1]
|
||
assert works["C2"].value == "Shared work" and works["E2"].value == "Статья"
|
||
assert "Scopus Q1" in works["F2"].value and "Web of Science Q2" in works["F2"].value
|
||
assert works["M2"].value == "https://doi.org/10.1234/shared"
|
||
assert works["N2"].value == "https://publications.hse.ru/view/27"
|
||
assert people["D2"].data_type == "n" and people.freeze_panes == "A2"
|
||
assert works.auto_filter.ref == works.dimensions
|
||
text = " ".join(str(cell.value) for sheet in book for row in sheet for cell in row)
|
||
assert "private@example.org" not in text and "Без степени" not in text
|
||
focused = load_workbook(BytesIO(client.get("/admin/council/export.xlsx?q=Петров%2027").content))
|
||
assert focused["Сотрудники"].max_row == 29
|
||
assert focused["Подбор диссовета"]["C5"].value == "Петров 27"
|
||
empty = load_workbook(BytesIO(client.get("/admin/council/export.xlsx?q=missing").content))
|
||
assert empty["Сотрудники"].max_row == 29 and empty["Публикации"].max_row == 29
|
||
assert empty["Подбор диссовета"]["C5"].value == "missing"
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
def test_xlsx_treats_profile_text_as_text_and_rejects_silent_truncation(db_session, export):
|
||
person = employee(db_session, '=HYPERLINK("https://example.org","name")')
|
||
publication(db_session, person, title="=1+1\x01")
|
||
publication(db_session, person, title="#N/A")
|
||
db_session.commit()
|
||
app.dependency_overrides[get_db] = lambda: db_session
|
||
try:
|
||
client = TestClient(app)
|
||
response = client.get("/admin/council/export.xlsx")
|
||
assert response.status_code == 200
|
||
book = load_workbook(BytesIO(response.content))
|
||
assert book["Сотрудники"]["B2"].value == person.full_name
|
||
assert book["Сотрудники"]["B2"].data_type == "s"
|
||
assert book["_Данные подбора"]["B2"].value == person.full_name
|
||
assert book["_Данные подбора"]["B2"].data_type == "s"
|
||
titles = {book["Публикации"].cell(n, 3).value: book["Публикации"].cell(n, 3).data_type for n in [2, 3]}
|
||
assert titles == {"=1+1": "s", "#N/A": "s"}
|
||
with ZipFile(BytesIO(response.content)) as archive:
|
||
assert all(b"<f>" not in archive.read(f"xl/worksheets/sheet{n}.xml") for n in (2, 3, 4))
|
||
person.full_name = "Long " + "x" * 32768
|
||
db_session.commit()
|
||
failure = client.get("/admin/council/export.xlsx")
|
||
assert failure.status_code == 503 and "Не удалось создать XLSX" in failure.text
|
||
assert "content-disposition" not in failure.headers
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
def test_xlsx_selection_contains_full_data_and_preserves_initial_filters(db_session, export):
|
||
matching = employee(db_session)
|
||
publication(db_session, matching, kind="ARTICLE", raw_data={"journal": "Бизнес-информатика"})
|
||
other = employee(db_session, "Другой сотрудник", key="other")
|
||
publication(db_session, other, title="Other", kind="ARTICLE")
|
||
db_session.commit()
|
||
app.dependency_overrides[get_db] = lambda: db_session
|
||
try:
|
||
client = TestClient(app)
|
||
response = client.get('/admin/council/export.xlsx?specialty=1.2.2&q=Иванов&sort=confirmed_count&direction=asc')
|
||
assert response.status_code == 200
|
||
book = load_workbook(BytesIO(response.content))
|
||
dashboard, source, options = book.worksheets[0], book['_Данные подбора'], book['_Справочники подбора']
|
||
assert dashboard.title == 'Подбор диссовета' and book.active == dashboard
|
||
assert dashboard['C3'].value.startswith('1.2.2 —')
|
||
assert dashboard['C4'].value == 'технические' and dashboard['C5'].value == 'Иванов'
|
||
assert dashboard['K3'].value == 'Подтверждено' and dashboard['K4'].value == 'По возрастанию'
|
||
assert source.max_row == 5 and source.sheet_state == options.sheet_state == 'hidden'
|
||
assert set(source.cell(row, 2).value for row in range(2, 6)) == {matching.full_name, other.full_name}
|
||
assert {source.cell(row, 18).value for row in range(2, 6)} == {'технические', 'физико-математические'}
|
||
assert {source.cell(row, 11).value for row in range(2, 6)} == {9, 11}
|
||
assert book['Сотрудники'].max_row == book['Публикации'].max_row == 3
|
||
assert len(dashboard.data_validations.dataValidation) == 6
|
||
for row in range(3, options.max_row + 1):
|
||
code = options.cell(row, 1).value.split(' ')[0]
|
||
expected = ['физико-математические'] if code == '1.1.2' else ['технические'] if code == '2.3.2' else ['технические', 'физико-математические']
|
||
assert [options.cell(row, column).value for column in (4, 5) if options.cell(row, column).value] == expected
|
||
assert options.cell(row, 2).value in book.defined_names
|
||
assert '_xlfn.AGGREGATE' in dashboard['S10'].value
|
||
assert '$C$3' in dashboard['F4'].value and 'IFERROR' in dashboard['F4'].value
|
||
memo_address = dashboard['A7'].hyperlink.location.split('!', 1)[1]
|
||
assert dashboard[memo_address].value == 'Памятка: как пользоваться подбором'
|
||
assert dashboard[memo_address].row > 11
|
||
memo_text = ' '.join(str(dashboard.cell(row, 1).value or '').replace('\n', ' ')
|
||
for row in range(dashboard[memo_address].row, dashboard.max_row + 1))
|
||
for explanation in ('Для кандидата/PhD', 'Подтверждено', '9 подтверждённых',
|
||
'11 по физико-математической', '20', 'Не подтверждены', 'все работы сотрудника',
|
||
'Уволенные', 'автоматический расчёт', 'не отправляются на сайт'):
|
||
assert explanation in memo_text
|
||
assert any(cell.hyperlink and cell.hyperlink.location == "'Подбор диссовета'!A1"
|
||
for row in dashboard for cell in row)
|
||
assert book.calculation.calcMode == 'auto' and book.calculation.forceFullCalc
|
||
with ZipFile(BytesIO(response.content)) as archive:
|
||
assert not any('vbaProject' in name or 'externalLinks' in name for name in archive.namelist())
|
||
db_session.query(EmployeePublication).delete()
|
||
db_session.query(Employee).delete()
|
||
db_session.commit()
|
||
empty = load_workbook(BytesIO(client.get('/admin/council/export.xlsx').content))
|
||
assert empty['Сотрудники'].max_row == 1 and empty['Публикации'].max_row == 1
|
||
assert empty['_Данные подбора'].max_row == 1
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
def test_xlsx_appending_rows_does_not_scan_growing_sheet(monkeypatch):
|
||
def fail_on_scan(sheet):
|
||
pytest.fail("Appending rows must not recompute worksheet bounds by scanning existing cells")
|
||
monkeypatch.setattr(Worksheet, "max_row", property(fail_on_scan))
|
||
monkeypatch.setattr(Worksheet, "max_column", property(fail_on_scan))
|
||
sheet = Workbook().active
|
||
for number in range(1000):
|
||
_append(sheet, [number, "text"])
|
||
assert sheet.cell(1000, 1).value == 999
|
||
assert sheet.cell(1000, 2).value == "text"
|
||
|
||
|
||
def test_category_counts_union_k1_journal_matching_and_period(db_session, export):
|
||
export([
|
||
row(title="All categories", doi="10.1234/all", **{"Квартиль wos": "Q2", "Название журнала": "Бизнес-информатика"}),
|
||
row(title="All categories", doi="10.1234/all"),
|
||
row(title="Q2 only", doi="10.1234/q2", **{"Квартиль scopus": "", "Квартиль wos": "Q2"}),
|
||
row(title="Old K1", doi="10.1234/old", year="2020", **{"Квартиль scopus": "", "Название журнала": "Бизнес-информатика"}),
|
||
row(title="Unknown year", doi="10.1234/unknown", year="", **{"Название журнала": "Бизнес-информатика"}),
|
||
row(title="Future", doi="10.1234/future", year="2027"),
|
||
row(title="Known book", doi="10.1234/book", **{"Название журнала": "Бизнес-информатика"}),
|
||
])
|
||
person = employee(db_session)
|
||
publication(db_session, person, title="All categories", doi="https://doi.org/10.1234/all", kind="ARTICLE")
|
||
publication(db_session, person, title="Known book", doi="https://doi.org/10.1234/book", kind="BOOK")
|
||
publication(db_session, person, title="K1 only", kind="ARTICLE",
|
||
description={"short": {"ru": "<b>Бизнес-информатика</b>. 2024. № 2."}})
|
||
publication(db_session, person, title="Similar journal", kind="ARTICLE",
|
||
description={"short": {"ru": "Бизнес-информатика и технологии. 2024."}})
|
||
publication(db_session, person, title="Mention in title is not journal Бизнес-информатика", kind="ARTICLE")
|
||
db_session.commit()
|
||
candidate = council.council_candidates(db_session, date(2026, 10, 8))["candidates"][0]
|
||
assert [candidate[f"{key}_count"] for key in ("q1", "q2", "k1", "total")] == [1, 2, 2, 3]
|
||
assert candidate["count"] == 9
|
||
works = {article["title"]: article for article in candidate["articles"]}
|
||
assert works["All categories"]["q1"] and works["All categories"]["q2"] and works["All categories"]["k1"]
|
||
assert works["K1 only"]["k1"] and "К1 · Бизнес-информатика (справочник ВАК)" in works["K1 only"]["criteria"]
|
||
for title in ("Old K1", "Unknown year", "Future", "Known book", "Similar journal"):
|
||
assert not works[title]["qualifies"]
|
||
|
||
|
||
def test_k1_reference_integrity():
|
||
import json
|
||
data = json.loads(council.K1_JOURNALS.read_text(encoding="utf-8"))
|
||
assert data["source_rows"] == 3134
|
||
assert sum(data["category_counts"].values()) == 3134
|
||
assert data["category_counts"]["К1"] == len(data["journals"]) == 773
|
||
assert len(council._k1_journals()) == 774 # 773 journals and one documented previous name.
|
||
assert "вестник российских университетов математика" not in council._k1_journals()
|
||
assert "бизнес информатика" in council._k1_journals()
|
||
|
||
|
||
@pytest.mark.parametrize("journal", [
|
||
"Ученые записки Казанской академии ветеринарной медицины им. Н.Э. Баумана",
|
||
"Ученые записки Казанской государственной академии ветеринарной медицины им. Н.Э. Баумана",
|
||
])
|
||
@pytest.mark.parametrize("source", ["csv", "journal", "description"])
|
||
def test_renamed_k1_journal_matches_current_and_documented_previous_name(db_session, export, journal, source):
|
||
person = employee(db_session)
|
||
if source == "csv":
|
||
export([row(**{"Квартиль scopus": "", "Название журнала": journal})])
|
||
else:
|
||
fields = {"raw_data": {"journal": journal}} if source == "journal" else {
|
||
"description": {"short": {"ru": f"{journal}. 2024. № 1."}}
|
||
}
|
||
publication(db_session, person, kind="ARTICLE", **fields)
|
||
db_session.commit()
|
||
result = council.council_candidates(db_session, date(2026, 10, 8))
|
||
candidate = result["candidates"][0]
|
||
assert [candidate[f"{key}_count"] for key in ("q1", "q2", "k1", "total")] == [0, 0, 1, 1]
|
||
canonical = "Ученые записки Казанской академии ветеринарной медицины им. Н.Э. Баумана"
|
||
assert f"К1 · {canonical} (справочник ВАК)" in candidate["articles"][0]["criteria"]
|
||
book = load_workbook(BytesIO(council_xlsx(result, selection_results(db_session, result))))
|
||
assert [book["Сотрудники"].cell(2, column).value for column in (6, 8)] == [1, 1]
|
||
|
||
|
||
def test_k1_rename_note_is_preserved_separately():
|
||
import json
|
||
data = json.loads(council.K1_JOURNALS.read_text(encoding="utf-8"))
|
||
journal = next(item for item in data["journals"] if item["source_row"] == 699)
|
||
assert "наименование в Перечне" not in journal["name"]
|
||
assert journal["note"].startswith("до 10.04.2026 г. наименование в Перечне")
|
||
assert len(journal["aliases"]) == 1
|
||
assert council._k1_journals()[council._journal_key(journal["aliases"][0])] == journal["name"]
|
||
bilingual = next(item for item in data["journals"] if item["source_row"] == 2)
|
||
assert bilingual["name"] == "Acta Biomedica Scientifica (East Siberian Biomedical Journal)"
|
||
|
||
|
||
def test_ambiguous_k1_alias_is_rejected(monkeypatch, tmp_path):
|
||
import json
|
||
reference = tmp_path / "k1.json"
|
||
reference.write_text(json.dumps({"journals": [
|
||
{"name": "One", "aliases": ["Previous"]}, {"name": "Two", "aliases": ["Previous"]}
|
||
]}), encoding="utf-8")
|
||
monkeypatch.setattr(council, "K1_JOURNALS", reference)
|
||
council._k1_journals.cache_clear()
|
||
try:
|
||
with pytest.raises(ValueError, match="Ambiguous K1 journal alias"):
|
||
council._k1_journals()
|
||
finally:
|
||
council._k1_journals.cache_clear()
|
||
|
||
|
||
@pytest.mark.parametrize("reverse", [False, True])
|
||
def test_conflicting_source_years_keep_categories_with_their_year(db_session, export, reverse):
|
||
rows = [row(year="2020", **{"Название журнала": "Бизнес-информатика"}),
|
||
row(year="2024", **{"Квартиль scopus": "Q2"})]
|
||
export(list(reversed(rows)) if reverse else rows)
|
||
person = employee(db_session)
|
||
# Evidence with no year must not inherit an arbitrary conflicting year.
|
||
publication(db_session, person, doi="https://doi.org/10.1234/shared", year=None,
|
||
raw_data={"csv_enrichment": {"Квартиль scopus": "Q1", "Журнал": "Бизнес-информатика"}})
|
||
db_session.commit()
|
||
candidate = council.council_candidates(db_session, date(2026, 10, 8))["candidates"][0]
|
||
assert [candidate[f"{key}_count"] for key in ("q1", "q2", "k1", "total")] == [0, 1, 0, 1]
|
||
assert candidate["count"] == 1
|
||
article = candidate["articles"][0]
|
||
assert article["year"] is None
|
||
assert any("Годы источников расходятся: 2020, 2024" in criterion for criterion in article["criteria"])
|
||
|
||
|
||
@pytest.mark.parametrize("sort", ["q1_count", "q2_count", "k1_count", "total_count"])
|
||
@pytest.mark.parametrize("direction", ["asc", "desc"])
|
||
def test_sort_category_counters_and_xlsx(db_session, export, sort, direction):
|
||
export([row(author="Анна", **{"Квартиль wos": "Q2", "Название журнала": "Бизнес-информатика"})])
|
||
employee(db_session, "Анна", key="anna")
|
||
person = employee(db_session, "Борис", key="boris")
|
||
publication(db_session, person, title="Without criteria")
|
||
db_session.commit()
|
||
result = council.council_candidates(db_session, date(2026, 10, 8), sort=sort, direction=direction)
|
||
expected = ["Борис", "Анна"] if direction == "asc" else ["Анна", "Борис"]
|
||
assert [candidate["name"] for candidate in result["candidates"]] == expected
|
||
book = load_workbook(BytesIO(council_xlsx(result, selection_results(db_session, result))))
|
||
headers = [cell.value for cell in book["Сотрудники"][1]]
|
||
assert headers[3:8] == ["Q1", "Q2", "К1", "К2", "Всего"]
|
||
assert "CSV по критериям" not in headers
|
||
anna_row = 3 if direction == "asc" else 2
|
||
assert [book["Сотрудники"].cell(anna_row, column).value for column in range(4, 9)] == [1, 1, 1, 0, 1]
|
||
|
||
|
||
def test_specialty_filters_use_eligible_source_journal_not_merged_evidence(db_session, export):
|
||
export([
|
||
row(title="Matching", doi="10.1234/matching", **{"Название журнала": "Бизнес-информатика"}),
|
||
row(title="Matching", doi="10.1234/matching"),
|
||
row(title="Old journal, new Q1", doi="10.1234/conflict", year="2020", **{"Название журнала": "Бизнес-информатика"}),
|
||
row(title="Old journal, new Q1", doi="10.1234/conflict", year="2024"),
|
||
row(title="Book", doi="10.1234/book", **{"Название журнала": "Бизнес-информатика"}),
|
||
row(title="Unknown year", doi="10.1234/unknown", year="", **{"Название журнала": "Бизнес-информатика"}),
|
||
])
|
||
person = employee(db_session)
|
||
publication(db_session, person, title="Book", doi="10.1234/book", kind="BOOK")
|
||
publication(db_session, person, title="Unknown year", doi="10.1234/unknown", year=None)
|
||
other = employee(db_session, name="Другой доктор", key="other")
|
||
publication(db_session, other, title="Q1 alone", raw_data={"csv_enrichment": {"Квартиль scopus": "Q1"}})
|
||
db_session.commit()
|
||
result = council.council_candidates(db_session, date(2026, 10, 8), specialty="1.2.2")
|
||
assert (result["total"], result["matched"]) == (2, 1)
|
||
candidate = result["candidates"][0]
|
||
assert candidate["specialty_count"] == 1
|
||
assert [a["title"] for a in candidate["articles"] if a["matches_specialty"]] == ["Matching"]
|
||
assert "По специальности 1.2.2" in " ".join(candidate["articles"][0]["criteria"])
|
||
assert council.council_candidates(db_session, specialty="2.2.14")["matched"] == 0
|
||
with pytest.raises(council.CouncilInputError):
|
||
council.council_candidates(db_session, specialty="<script>")
|
||
|
||
|
||
def test_specialty_page_pagination_search_and_export_agree(db_session, export):
|
||
for number in range(28):
|
||
person = employee(db_session, name=f"Доктор {number:02d}", key=str(number))
|
||
publication(db_session, person, title=f"Work {number}", kind="ARTICLE",
|
||
raw_data={"journal": "Бизнес-информатика"})
|
||
person = employee(db_session, name="Не подходит", key="unrelated")
|
||
publication(db_session, person, raw_data={"journal": "Неизвестный журнал"})
|
||
db_session.commit()
|
||
app.dependency_overrides[get_db] = lambda: db_session
|
||
try:
|
||
client = TestClient(app)
|
||
response = client.get("/admin/council?specialty=1.2.2&sort=full_name&direction=asc")
|
||
assert response.status_code == 200
|
||
from bs4 import BeautifulSoup
|
||
selected = BeautifulSoup(response.text, "html.parser").select_one('#council-specialty option[selected]')
|
||
assert selected["value"] == "1.2.2"
|
||
assert response.text.count('<option value="2.3.5"') == 1
|
||
assert "Публикаций по выбранной специальности: 1" in client.get("/admin/council/employees/1/articles?specialty=1.2.2").text
|
||
assert "Не подходит" not in response.text
|
||
from bs4 import BeautifulSoup
|
||
soup = BeautifulSoup(response.text, "html.parser")
|
||
next_page = soup.find("a", string="Далее")["href"]
|
||
assert "specialty=1.2.2" in next_page
|
||
assert "Доктор 27" in client.get("/admin/council" + next_page).text
|
||
filtered = client.get("/admin/council?specialty=1.2.2&q=27")
|
||
assert "Показаны 1–1 из 1" in filtered.text
|
||
book = load_workbook(BytesIO(client.get("/admin/council/export.xlsx?specialty=1.2.2&sort=full_name&direction=asc").content))
|
||
assert book["Сотрудники"].max_row == 30
|
||
assert book["Подбор диссовета"]["C3"].value.startswith("1.2.2 —")
|
||
params = dict(book["Параметры"].values)
|
||
assert params["Специальность ВАК"].startswith("1.2.2.")
|
||
assert "Перечень ВАК" in book["Публикации"]["F2"].value or "справочник ВАК" in book["Публикации"]["F2"].value
|
||
empty = client.get("/admin/council?specialty=2.2.14")
|
||
assert "Сотрудники не найдены" in empty.text
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
def test_specialty_reference_integrity():
|
||
import hashlib
|
||
import json
|
||
from scripts.extract_k1_specialties import active_codes
|
||
data = json.loads(council.K1_SPECIALTIES.read_text(encoding="utf-8"))
|
||
k1 = json.loads(council.K1_JOURNALS.read_text(encoding="utf-8"))
|
||
assert data["k1_source_sha256"] == hashlib.sha256(council.K1_JOURNALS.read_bytes().replace(b"\r\n", b"\n")).hexdigest()
|
||
assert [j["name"] for j in data["journals"]] == [j["name"] for j in k1["journals"]]
|
||
assert len(data["journals"]) == 773 and len(data["specialties"]) == 12
|
||
assert data["source_rows"] == 3231
|
||
for journal in data["journals"]:
|
||
assert journal["vak_entries"]
|
||
if journal["status"] == "matched":
|
||
active = [entry for entry in journal["vak_entries"] if active_codes(entry, data["as_of"])]
|
||
assert len(active) == 1
|
||
assert journal["specialty_codes"] == active_codes(active[0], data["as_of"])
|
||
else:
|
||
assert journal["status"] == "ambiguous" and not journal["specialty_codes"]
|
||
for specialty in data["specialties"]:
|
||
assert any(specialty["code"] in j["specialty_codes"] for j in data["journals"])
|
||
|
||
|
||
def test_xlsx_specialties_are_explicit_with_and_without_filter(db_session, export):
|
||
export([row(title="Old journal", doi="10.1234/conflict", year="2020", **{"Название журнала": "Бизнес-информатика"}),
|
||
row(title="Old journal", doi="10.1234/conflict", year="2024")])
|
||
person = employee(db_session)
|
||
for title in ['Matching', 'Second matching']:
|
||
publication(db_session, person, title=title, kind='ARTICLE', raw_data={'journal': 'Бизнес-информатика'})
|
||
publication(db_session, person, title='Book', kind='BOOK', raw_data={'journal': 'Бизнес-информатика'})
|
||
publication(db_session, person, title='Unknown journal', kind='ARTICLE', raw_data={'journal': 'Other'})
|
||
other = employee(db_session, name='Без соответствия', key='other')
|
||
publication(db_session, other, title='Unknown', kind='ARTICLE')
|
||
db_session.commit()
|
||
for specialty in ('', '1.2.2'):
|
||
result = council.council_candidates(db_session, date(2026, 10, 8), specialty=specialty, paginate=False)
|
||
book = load_workbook(BytesIO(council_xlsx(result, selection_results(db_session, result))))
|
||
staff = list(book['Сотрудники'].values)
|
||
works = list(book['Публикации'].values)
|
||
candidate = next(dict(zip(staff[0], values)) for values in staff[1:] if values[1] == person.full_name)
|
||
assert candidate['Специальности ВАК'].count('1.2.2.') == 1
|
||
assert 'Математическое моделирование' in candidate['Специальности ВАК']
|
||
assert '2.3.5.' in candidate['Специальности ВАК']
|
||
assert '5.2.2.' not in candidate['Специальности ВАК'] # Only the site's twelve specialties.
|
||
assert 'Работ по выбранной специальности' not in staff[0]
|
||
records = {values[2]: dict(zip(works[0], values)) for values in works[1:]}
|
||
assert '1.2.2.' in records['Matching']['Специальность ВАК']
|
||
assert 'По выбранной специальности' not in works[0]
|
||
for title in ('Book', 'Unknown journal', 'Old journal'):
|
||
assert records[title]['Специальность ВАК'] == 'Не подтверждены'
|
||
if not specialty:
|
||
unmatched = next(dict(zip(staff[0], values)) for values in staff[1:] if values[1] == other.full_name)
|
||
assert unmatched['Специальности ВАК'] == 'Не подтверждены'
|
||
assert book['Сотрудники'].auto_filter.ref.endswith(f'L{len(staff)}')
|
||
assert book['Публикации'].auto_filter.ref.endswith(f'N{len(works)}')
|
||
|
||
|
||
def test_employee_without_degree_publications_are_not_parsed(db_session, export, monkeypatch):
|
||
person = employee(db_session, degree='Профессор')
|
||
publication(db_session, person, kind='ARTICLE', description={'short': {'ru': '<b>Journal</b>. 2024.'}})
|
||
db_session.commit()
|
||
def unexpected(publication):
|
||
pytest.fail('Ineligible employee publications must not be parsed')
|
||
monkeypatch.setattr(council, '_database_article', unexpected)
|
||
assert council.council_candidates(db_session)['total'] == 0
|
||
|
||
|
||
def test_xlsx_compact_total_current_degree_and_normalized_criteria(db_session, export):
|
||
export([row(**{'Название журнала': 'Бизнес-информатика', 'Квартиль wos': 'Q2'})])
|
||
person = employee(db_session)
|
||
person.current_data = {'sections': [{'title': 'Учёные степени', 'items': [
|
||
'Кандидат технических наук', 'Доктор технических наук']} ]}
|
||
publication(db_session, person, title='Shared work', doi='https://doi.org/10.1234/shared', kind='ARTICLE')
|
||
publication(db_session, person, title='Other publication', kind=None)
|
||
db_session.commit()
|
||
result = council.council_candidates(db_session, date(2026, 10, 8))
|
||
candidate = result['candidates'][0]
|
||
other = next(a for a in candidate['articles'] if a['title'] == 'Other publication')
|
||
# Export the same total as the site, including doctor-only K2 and RSCI.
|
||
other.update(k2=True, rsci=True, qualifies=True)
|
||
candidate.update(k2_count=1, rsci_count=1, total_count=2)
|
||
shared = next(a for a in candidate['articles'] if a['title'] == 'Shared work')
|
||
shared['criteria'].extend(['В пределах пяти лет', 'В пределах пяти лет по году; точную дату нужно проверить'])
|
||
book = load_workbook(BytesIO(council_xlsx(result, selection_results(db_session, result))))
|
||
staff = dict(zip([c.value for c in book['Сотрудники'][1]], [c.value for c in book['Сотрудники'][2]]))
|
||
assert staff['Научная степень'] == 'Доктор технических наук'
|
||
assert staff['Q1'] == staff['Q2'] == staff['К1'] == staff['К2'] == 1
|
||
assert staff['Всего'] == candidate['total_count'] == 2
|
||
assert staff['Всего из ЛК'] == staff['Общее количество всех публикаций без дублей'] == 2
|
||
rows = {values[2]: dict(zip(next(book['Публикации'].values), values)) for values in list(book['Публикации'].values)[1:]}
|
||
assert rows['Other publication']['Тип материала'] == 'не указан'
|
||
assert rows['Other publication']['К2'] == rows['Other publication']['RSCI'] == 'Да'
|
||
assert rows['Shared work']['Критерии'].count('В пределах пяти лет') == 1
|
||
assert 'точную дату' not in rows['Shared work']['Критерии']
|
||
params = dict(book['Параметры'].values)
|
||
assert 'Q1/Q2/К1/RSCI' in params['Всего'] and 'для доктора наук также К2' in params['Всего']
|
||
assert 'источники указаны на листе' not in ' '.join(str(v) for v in params.values())
|
||
assert not any('Работ / По выбранной специальности' == key for key in params)
|
||
|
||
|
||
def test_council_lazy_loading_selects_only_matching_publications(db_session, export, monkeypatch):
|
||
from sqlalchemy import event
|
||
first = employee(db_session, name='Первый сотрудник', key='first')
|
||
second = employee(db_session, name='Второй сотрудник', key='second')
|
||
publication(db_session, first, title='Первая статья', kind='ARTICLE')
|
||
publication(db_session, second, title='Вторая статья', kind='ARTICLE')
|
||
db_session.commit()
|
||
parsed = []
|
||
original = council._database_article
|
||
def tracked(publication):
|
||
parsed.append(publication.employee_id)
|
||
return original(publication)
|
||
monkeypatch.setattr(council, '_database_article', tracked)
|
||
result = council.council_candidates(db_session, q='первый')
|
||
assert result['total'] == 2 and result['matched'] == 1
|
||
assert parsed == [first.id]
|
||
parsed.clear()
|
||
statements = []
|
||
def sql(connection, cursor, statement, parameters, context, executemany):
|
||
statements.append(statement)
|
||
event.listen(db_session.bind, 'before_cursor_execute', sql)
|
||
try:
|
||
result = council.council_candidates(db_session, q='нет совпадений')
|
||
assert result['total'] == 2 and result['matched'] == 0
|
||
assert not any('employee_publications' in statement for statement in statements)
|
||
finally:
|
||
event.remove(db_session.bind, 'before_cursor_execute', sql)
|
||
assert parsed == []
|
||
app.dependency_overrides[get_db] = lambda: db_session
|
||
try:
|
||
client = TestClient(app)
|
||
page = client.get('/admin/council')
|
||
assert 'data-articles-url' in page.text and 'class="council__article"' not in page.text
|
||
detail = client.get(f'/admin/council/employees/{first.id}/articles')
|
||
assert detail.status_code == 200
|
||
assert 'Первая статья' in detail.text and 'Вторая статья' not in detail.text
|
||
assert parsed == [first.id, second.id, first.id]
|
||
assert client.get('/admin/council/employees/99999/articles').status_code == 404
|
||
assert client.get(f'/admin/council/employees/{first.id}/articles?specialty=wrong').status_code == 422
|
||
finally:
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
def test_lazy_loading_preserves_global_duplicate_name_check(db_session, export):
|
||
export([row(author='Одно ФИО', title='Не присваивать автоматически')])
|
||
first = employee(db_session, name='Одно ФИО', key='first')
|
||
employee(db_session, name='Одно ФИО', key='second')
|
||
publication(db_session, first, title='Своя статья')
|
||
db_session.commit()
|
||
result = council.council_candidates(db_session, employee_id=first.id)
|
||
assert result['total'] == 2 and result['matched'] == 1
|
||
assert result['candidates'][0]['ambiguous']
|
||
assert [a['title'] for a in result['candidates'][0]['articles']] == ['Своя статья']
|
||
|
||
|
||
def test_plain_description_fast_path_and_html_entities():
|
||
from app.services.council_source import _description_journals
|
||
description = {'short': {'ru': ' Journal Name. 2024. № 1. ', 'en': '<b>Journal & Name</b>. 2024.'},
|
||
'shortLeft': {'en': 'Journal & Name. 2024.'}}
|
||
assert _description_journals(description) == ['Journal Name', 'Journal & Name', 'Journal & Name']
|