miem_workers/tests/test_council_assessment.py

import itertools
import json
from datetime import date, datetime
from io import BytesIO

import pytest
from fastapi.testclient import TestClient
from openpyxl import load_workbook

from app.db import get_db
from app.main import app
from app.models import Employee, EmployeePublication
from app.services import council, council_assessment as assessment
from app.services.council_catalog import catalog, citescore_quartile, publication_catalog
from app.services.council_export import council_xlsx


SOURCE = {"source": "Проверенная запись", "checked_on": "2026-10-08"}


@pytest.fixture()
def proofs(monkeypatch, tmp_path):
    path = tmp_path / "evidence.json"
    monkeypatch.setattr(assessment, "EVIDENCE", path)
    monkeypatch.setattr(council, "_articles_by_author", lambda: {})
    def write(degrees=None, publications=None, policy=None):
        path.write_text(json.dumps({"schema_version": 1, "policy": policy or {}, "employees": {
            "test": {"degrees": degrees or [], "publications": publications or []}}}), encoding="utf8")
        assessment.evidence.cache_clear()
    write()
    yield write
    assessment.evidence.cache_clear()


def person(db, degrees):
    value = Employee(profile_key="test", canonical_url="https://www.hse.ru/staff/test", full_name="Тестовый сотрудник",
                     current_data={"sections": [{"title": "Учёные степени", "items": degrees}]})
    db.add(value)
    db.flush()
    return value


def work(db, employee, number=0, year=2024, kind="ARTICLE", basis="К1", **proof_fields):
    doi = f"10.1234/work-{number}"
    db.add(EmployeePublication(employee_id=employee.id, title=f"Work {number}", year=year, publication_type=kind,
                               source_hash=f"hash-{number}", doi_url=f"https://doi.org/{doi}"))
    return {"doi": doi, "kind": "article", "kind_evidence": SOURCE,
            "bases": [{**SOURCE, "basis": basis, "applicable_year": year}], **proof_fields}


def result(db, branch="технические", assessment_branch=None, **kwargs):
    db.commit()
    value = council.council_candidates(db, today=date(2026, 10, 8), branch=branch, **kwargs)["candidates"][0]
    if assessment_branch:
        return assessment.assess_candidate(value["degree"].split("; "), value["articles"],
                                           assessment.evidence()["employees"]["test"], assessment_branch, "",
                                           date(2022, 1, 1), date(2026, 12, 31))
    return value


@pytest.mark.parametrize("branch,rule", assessment.BRANCHES.items())
@pytest.mark.parametrize("delta", [-1, 0, 1])
def test_each_doctor_threshold_and_monograph_requirement(db_session, proofs, branch, rule, delta):
    minimum, mono = rule
    degree = "Доктор технических наук"
    employee = person(db_session, [degree])
    records = [work(db_session, employee, n) for n in range(minimum + delta)]
    proofs([{**SOURCE, "degree": degree}], records)
    value = result(db_session, assessment_branch=branch)
    assert value["minimum"] == minimum and value["monograph_minimum"] == mono
    assert value["confirmed_count"] == minimum + delta
    assert value["assessment_status"].startswith("Соответствует") == (delta >= 0 and not mono)


@pytest.mark.parametrize("degree", ["Кандидат технических наук", "Ph.D."])
@pytest.mark.parametrize("count", [19, 20, 21])
@pytest.mark.parametrize("basis", ["К1", "К2", "RSCI"])
def test_candidate_phd_20_and_k2_exclusion(db_session, proofs, degree, count, basis):
    employee = person(db_session, [degree])
    proofs([{**SOURCE, "degree": degree}], [work(db_session, employee, n, basis=basis) for n in range(count)])
    value = result(db_session)
    assert value["clause"] == 6 and value["minimum"] == 20
    assert value["confirmed_count"] == (0 if basis == "К2" else count)
    assert value["assessment_status"].startswith("Соответствует") == (basis != "К2" and count >= 20)


@pytest.mark.parametrize("degree,total", [("Доктор технических наук", 3), ("Кандидат технических наук", 2), ("Ph.D.", 2)])
def test_xlsx_total_matches_site_for_k2_rsci_and_overlapping_bases(db_session, proofs, degree, total):
    employee = person(db_session, [degree])
    records = [work(db_session, employee, number, basis=basis) for number, basis in enumerate(("К2", "RSCI", "К1"))]
    records[-1]["bases"].append({**SOURCE, "basis": "К2", "applicable_year": 2024})
    proofs([{**SOURCE, "degree": degree}], records)
    db_session.commit()
    selections = [council.council_candidates(db_session, date(2026, 10, 8), branch=branch, paginate=False)
                  for branch in ("технические", "физико-математические")]
    candidate = selections[0]["candidates"][0]
    assert candidate["k2_count"] == 2 and candidate["rsci_count"] == 1
    assert candidate["total_count"] == total
    book = load_workbook(BytesIO(council_xlsx(selections[0], selections)))
    assert book["Сотрудники"]["H2"].value == total
    assert book["_Данные подбора"]["I2"].value == total
    assert book["_Данные подбора"]["I3"].value == total


@pytest.mark.parametrize("degrees", list(itertools.permutations(["Кандидат технических наук", "PhD", "Доктор технических наук"])))
def test_history_always_chooses_doctor_without_fallback(db_session, proofs, degrees):
    person(db_session, list(degrees) + ["Профессор", "Неизвестная квалификация"])
    proofs([{**SOURCE, "degree": "Кандидат технических наук"}])
    value = result(db_session)
    assert value["clause"] == 5 and value["selected_degree"] == "Доктор технических наук"
    assert not value["degree_confirmed"] and value["count"] == 0
    assert value["assessment_status"] == "Требует проверки"


def test_equal_candidate_and_phd_history(db_session, proofs):
    person(db_session, ["PhD", "Кандидат технических наук"])
    value = result(db_session)
    assert value["clause"] == 6 and "PhD" in value["selected_degree"] and "Кандидат" in value["selected_degree"]


@pytest.mark.parametrize("branch", [b for b, rule in assessment.BRANCHES.items() if rule[1]])
def test_monograph_explicit_evidence_and_period_policy(db_session, proofs, branch):
    degree = "Доктор технических наук"
    employee = person(db_session, [degree])
    records = [work(db_session, employee, n) for n in range(8)]
    records.append(work(db_session, employee, 8, kind="BOOK", kind_evidence=SOURCE))
    proofs([{**SOURCE, "degree": degree}], records)
    assert result(db_session, assessment_branch=branch)["monograph_count"] == 0
    records[-1]["kind"] = "monograph"
    proofs([{**SOURCE, "degree": degree}], records, {"monograph_period": {**SOURCE, "value": "five_years"}})
    value = result(db_session, assessment_branch=branch)
    assert value["confirmed_count"] == 8 and value["monograph_count"] == 1
    assert value["assessment_status"].startswith("Соответствует")


@pytest.mark.parametrize("year,exact,expected", [(2021, None, False), (2026, None, True),
    (2022, None, True), (2021, "2021-10-09", False), (2021, "2021-10-08", False),
    (2026, "2026-10-07", True), (2026, "2026-10-08", True), (2026, "2026-10-09", True),
    (2024, "2025-01-01", False), (2024, "invalid", True)])
def test_boundary_dates_and_conflicting_year(db_session, proofs, year, exact, expected):
    employee = person(db_session, ["Доктор технических наук"])
    record = work(db_session, employee, year=year)
    if exact:
        record.update(publication_date=exact, date_evidence=SOURCE)
    proofs(publications=[record])
    value = result(db_session)
    assert value["confirmed_count"] == int(expected)


def test_no_proof_from_quartile_csv_and_current_k1(db_session, proofs):
    employee = person(db_session, ["Доктор технических наук"])
    db_session.add(EmployeePublication(employee_id=employee.id, title="Preliminary", year=2024, source_hash="prelim",
        raw_data={"journal": "Бизнес-информатика", "csv_enrichment": {"Квартиль scopus": "Q1"}}))
    value = result(db_session)
    assert value["total_count"] == 1 and value["confirmed_count"] == 0
    assert value["assessment_status"] == "Требует проверки"


def test_catalogs_union_percentiles_and_article_not_approved():
    data = catalog()
    assert len(data["journals"]) == 229 and len(data["passports"]) == 13 and not data["issues"]
    assert [citescore_quartile(n) for n in [49, 50, 74, 75]] == ["Q3", "Q2", "Q2", "Q1"]
    records, issues = publication_catalog(["0006-3835"], ["BIT Numerical Mathematics"], "1.2.2")
    assert records and records[0]["match_level"] == "direct" and not issues
    assert {m["quartile"] for m in records[0]["metrics"]} >= {"Q2", "Q3"}
    assert all(m["year"] == 2025 for m in records[0]["metrics"])
    assert publication_catalog(["9999-9999"], ["Not evaluated"], "1.2.2") == ([], [])
    assert publication_catalog(["0006-3835"], ["Wrong name"])[1]


def test_http_validation_context_and_export_parity(db_session, proofs, monkeypatch):
    class Clock:
        @staticmethod
        def now(zone):
            return datetime(2026, 10, 8, tzinfo=zone)
    monkeypatch.setattr(council, "datetime", Clock)
    degree = "Доктор технических наук"
    employee = person(db_session, [degree])
    proofs([{**SOURCE, "degree": degree}], [work(db_session, employee, n) for n in range(9)])
    db_session.commit()
    def database():
        yield db_session
    app.dependency_overrides[get_db] = database
    try:
        client = TestClient(app)
        for query in ["specialty=oops", "branch=oops"]:
            assert client.get("/admin/council?" + query).status_code == 422
            assert client.get("/admin/council/export.xlsx?" + query).status_code == 422
        query = "?sort=confirmed_count"
        page = client.get("/admin/council" + query)
        assert page.status_code == 200 and "Соответствует публикационным критериям пункта 5" in page.text
        book = load_workbook(BytesIO(client.get("/admin/council/export.xlsx" + query).content))
        headers = [c.value for c in book["Сотрудники"][1]]
        row = dict(zip(headers, [c.value for c in book["Сотрудники"][2]]))
        assert row["Всего"] == 9 and row["Научная степень"] == degree
        assert dict(book["Параметры"].values)["Отрасль оценки"] == "технические"
        assert not any("ходатайств" in str(c.value).lower() for sheet in book for cells in sheet for c in cells)
        assert "council-date" not in page.text and "Монографии" not in page.text
        assert "Дата ходатайства" not in page.text
        legacy = client.get("/admin/council/export.xlsx?petition_date=2000-01-01&branch=экономические")
        assert legacy.status_code == 422
    finally:
        app.dependency_overrides.clear()


@pytest.mark.parametrize("populated", [False, True])
def test_empty_branch_is_rejected_before_export_and_omitted_branch_keeps_default(db_session, proofs, populated):
    if populated:
        employee = person(db_session, ["Доктор технических наук"])
        work(db_session, employee)
        db_session.commit()
    with pytest.raises(council.CouncilInputError, match="Неизвестная отрасль науки"):
        council.council_candidates(db_session, branch="")
    app.dependency_overrides[get_db] = lambda: db_session
    try:
        client = TestClient(app)
        for query in ("branch=", "branch=&specialty=1.1.2", "branch=&specialty=1.1.2&previous_specialty="):
            for path in ("/admin/council", "/admin/council/export.xlsx", "/admin/council/employees/1/articles"):
                response = client.get(path + "?" + query)
                assert response.status_code == 422
                assert "Неизвестная отрасль науки" in response.text
                assert "content-disposition" not in response.headers
        assert client.get("/admin/council").status_code == 200
        response = client.get("/admin/council/export.xlsx")
        assert response.status_code == 200
        book = load_workbook(BytesIO(response.content))
        assert book["Подбор диссовета"]["C4"].value == "технические"
        assert book["Сотрудники"].max_row == (2 if populated else 1)
    finally:
        app.dependency_overrides.clear()


def test_unknown_kind_conflicts_and_safe_source_url(db_session, proofs):
    employee = person(db_session, ["Доктор технических наук"])
    record = work(db_session, employee)
    record["bases"][0]["source_url"] = "javascript:alert(1)"
    records = [record, {**record, "kind": "monograph"}]
    proofs(publications=records)
    value = result(db_session)
    assert value["confirmed_count"] == 0
    assert value["articles"][0]["confirmed_bases"][0]["source_url"] is None


def test_no_degree_or_only_title_does_not_form_conclusion(db_session, proofs):
    person(db_session, ["Профессор", "Доцент"])
    assert council.council_candidates(db_session)["total"] == 0


def test_days_within_same_year_do_not_conflict_and_bases_count_once(db_session, proofs):
    employee = person(db_session, ["Доктор технических наук"])
    record = work(db_session, employee)
    record["bases"].append({**SOURCE, "basis": "RSCI", "applicable_year": 2024})
    proofs(publications=[record])
    assert result(db_session)["confirmed_count"] == 1
    proofs(publications=[{**record, "publication_date": "2024-01-01", "date_evidence": SOURCE},
                         {**record, "publication_date": "2024-02-01", "date_evidence": SOURCE}])
    assert result(db_session)["confirmed_count"] == 1


def test_exact_date_can_confirm_missing_year_without_losing_preliminary_count(db_session, proofs):
    employee = person(db_session, ["Доктор технических наук"])
    record = work(db_session, employee, year=None, publication_date="2024-03-01", date_evidence=SOURCE)
    record["bases"][0]["applicable_year"] = 2024
    proofs(publications=[record])
    value = result(db_session)
    assert value["confirmed_count"] == value["total_count"] == 1


def test_identity_mismatch_cannot_be_resolved_by_either_identifier(db_session, proofs):
    employee = person(db_session, ["Доктор технических наук"])
    record = work(db_session, employee)
    record["source_hash"] = "hash-of-another-work"
    proofs(publications=[record])
    value = result(db_session)
    assert value["confirmed_count"] == 0
    assert any("разные работы" in reason for reason in value["articles"][0]["confirmation_reasons"])


@pytest.mark.parametrize("passport_id,branch,accepted", [("1.3.8-technical", "технические", True),
    ("1.3.8-physical-mathematical", "технические", False),
    ("1.3.8-physical-mathematical", "физико-математические", True)])
def test_thematic_review_uses_correct_passport_branch(db_session, proofs, passport_id, branch, accepted):
    employee = person(db_session, ["Доктор технических наук"])
    passport = catalog()["passports"][passport_id]
    record = work(db_session, employee)
    record["thematic_review"] = {**SOURCE, "specialty_code": "1.3.8", "branch": branch,
        "passport_id": passport_id, "passport_sha256": passport["sha256"],
        "point_numbers": [passport["directions"][0]["number"]], "checked_conditions": [],
        "materials": ["Цель, методы, научный результат и текст статьи"], "result": "confirmed"}
    proofs(publications=[record])
    value = result(db_session, branch)
    article = value["articles"][0]
    assessment.assess_article(article, [record], date(2021, 10, 8), date(2026, 10, 8), 2, branch, "1.3.8", {})
    assert ("соответствие подтверждено" in article["thematic_status"]) == accepted
    assert article["confirmed"]  # The thematic outcome adds no numeric rule.


@pytest.mark.parametrize("system", sorted(assessment.SYSTEMS))
def test_external_quartile_outside_catalog_requires_category_and_source(db_session, proofs, system):
    employee = person(db_session, ["Доктор технических наук"])
    record = work(db_session, employee, basis="Q1")
    record["bases"][0].update(system=system, category="Verified category", metric="Verified metric", metric_year=2024)
    proofs(publications=[record])
    assert result(db_session)["confirmed_count"] == 1
    record["bases"][0].pop("category")
    proofs(publications=[record])
    assert result(db_session)["confirmed_count"] == 0


def test_rinc_and_hse_list_do_not_supply_rsci(db_session, proofs):
    employee = person(db_session, ["Доктор технических наук"])
    db_session.add(EmployeePublication(employee_id=employee.id, title="RINC", year=2024, publication_type="ARTICLE",
        source_hash="rinc", raw_data={"hse_list": True, "rinc": True}))
    value = result(db_session)
    assert value["rsci_count"] == value["confirmed_count"] == 0


def test_degree_document_can_include_employee_without_profile_degree(db_session, proofs):
    person(db_session, [])
    proofs(degrees=[{**SOURCE, "degree": "PhD"}])
    value = result(db_session)
    assert value["clause"] == 6 and value["degree_confirmed"] and value["count"] == 0


def test_technical_assessment_uses_current_period_without_petition(db_session, proofs):
    degree = "Доктор технических наук"
    employee = person(db_session, [degree])
    proofs([{**SOURCE, "degree": degree}], [work(db_session, employee, n) for n in range(9)])
    db_session.commit()
    value = council.council_candidates(db_session, today=date(2026, 10, 8))["candidates"][0]
    assert value["confirmed_count"] == value["minimum"] == 9
    assert value["assessment_status"].startswith("Соответствует")
    assert value["assessment_reasons"] == []


def test_calendar_years_ignore_legacy_boundary_policy(db_session, proofs):
    employee = person(db_session, ["Доктор технических наук"])
    record = work(db_session, employee, year=2021, publication_date="2021-10-08", date_evidence=SOURCE)
    proofs(publications=[record], policy={"boundary_dates": {"value": "inclusive"}})
    assert result(db_session)["confirmed_count"] == 0
    proofs(publications=[record], policy={"boundary_dates": {**SOURCE, "value": "inclusive"}})
    assert result(db_session)["confirmed_count"] == 0


def test_invalid_evidence_schema_is_rejected(monkeypatch, tmp_path):
    path = tmp_path / "invalid.json"
    monkeypatch.setattr(assessment, "EVIDENCE", path)
    path.write_text(json.dumps({"schema_version": 1, "employees": {"x": {"publications": [{"doi": "10.1234/x", "bases": "К1"}]}}}), encoding="utf8")
    assessment.evidence.cache_clear()
    try:
        with pytest.raises(ValueError):
            assessment.evidence()
    finally:
        assessment.evidence.cache_clear()


@pytest.mark.parametrize("basis,flag", [("К1", "k1"), ("К2", "k2"), ("RSCI", "rsci"), ("Q1", "q1"), ("Q2", "q2")])
@pytest.mark.parametrize("kind", ["BOOK", None])
@pytest.mark.parametrize("admitted", [False, True])
def test_manual_monograph_bases_respect_preliminary_type_policy(db_session, proofs, basis, flag, kind, admitted):
    employee = person(db_session, ["Доктор технических наук"])
    record = work(db_session, employee, kind=kind, basis=basis)
    record["kind"] = "monograph"
    policy = {"publication_types": {**SOURCE, "values": ["monograph"]}} if admitted else {}
    proofs(publications=[record], policy=policy)
    full_result = council.council_candidates(db_session, today=date(2026, 10, 8), branch="технические")
    candidate = full_result["candidates"][0]
    assert candidate[f"{flag}_count"] == candidate["total_count"] == int(admitted)
    # The policy permits a type; it does not itself confirm a quartile.
    assert candidate["confirmed_count"] == (int(admitted) if basis not in {"Q1", "Q2"} else 0)
    book = load_workbook(BytesIO(council_xlsx(full_result, [full_result, council.council_candidates(db_session, full_result["end"], branch="физико-математические", paginate=False)])))
    headers = [cell.value for cell in book["Сотрудники"][1]]
    exported = dict(zip(headers, [cell.value for cell in book["Сотрудники"][2]]))
    assert exported["Всего"] == candidate["total_count"] == int(admitted)
    publication = dict(zip([cell.value for cell in book["Публикации"][1]], [cell.value for cell in book["Публикации"][2]]))
    assert publication[basis] == ("Да" if admitted else "Нет")
    if basis != "RSCI":
        assert exported[basis] == int(admitted)


def test_admitted_type_needs_sourced_policy_and_consistent_material(db_session, proofs):
    employee = person(db_session, ["Доктор технических наук"])
    record = work(db_session, employee, kind="BOOK")
    record["kind"] = "monograph"
    proofs(publications=[record], policy={"publication_types": {"values": ["monograph"]}})
    assert result(db_session)["total_count"] == 0
    record["kind"] = "article"
    proofs(publications=[record])
    assert result(db_session)["total_count"] == 0


@pytest.mark.parametrize("broken_index", [0, 1])
def test_valid_snapshot_keeps_preliminary_metrics_when_other_is_broken(db_session, proofs, monkeypatch, tmp_path, broken_index):
    from app.services import council_catalog as catalogs
    paths = list(catalogs.CATALOGS)
    valid_path = paths[1 - broken_index]
    valid = json.loads(valid_path.read_text(encoding="utf8"))
    journal = valid["journals"][0]
    broken = tmp_path / "broken" / "catalog.json"
    broken.parent.mkdir()
    broken.write_text("{}", encoding="utf8")
    paths[broken_index] = broken
    monkeypatch.setattr(catalogs, "CATALOGS", tuple(paths))
    catalogs.catalog.cache_clear()
    try:
        employee = person(db_session, ["Доктор технических наук"])
        db_session.add(EmployeePublication(employee_id=employee.id, title="Snapshot work", year=2025,
            publication_type="ARTICLE", source_hash="snapshot", raw_data={"journal": journal["name"], "issn": journal["issn"]}))
        db_session.commit()
        full_result = council.council_candidates(db_session, today=date(2026, 10, 8), branch="технические")
        candidate = full_result["candidates"][0]
        article = candidate["articles"][0]
        flag = "q2" if broken_index == 0 else "q1"
        assert candidate[f"{flag}_count"] == candidate["total_count"] == 1
        assert candidate["confirmed_count"] == 1
        assert full_result["catalog_issues"] and not article["catalog_issues"]
        assert {r["catalog"] for r in article["catalog_records"]} == {valid_path.parent.name}
        book = load_workbook(BytesIO(council_xlsx(full_result, [full_result, council.council_candidates(db_session, full_result["end"], branch="физико-математические", paginate=False)])))
        assert book["Сотрудники"].cell(2, 5 if flag == "q2" else 4).value == 1
        assert "целостность каталога" in " ".join(str(cell.value) for row in book["Параметры"] for cell in row)
        app.dependency_overrides[get_db] = lambda: db_session
        try:
            response = TestClient(app).get("/admin/council")
            assert response.status_code == 200 and "целостность каталога" in response.text
            assert journal["name"] in TestClient(app).get(f"/admin/council/employees/{employee.id}/articles").text
        finally:
            app.dependency_overrides.clear()
    finally:
        catalogs.catalog.cache_clear()


@pytest.mark.parametrize('year,kind,expected', [(2021, 'ARTICLE', False), (2022, 'ARTICLE', True),
    (2026, 'ARTICLE', True), (2027, 'ARTICLE', False), (2024, None, False), (2024, 'BOOK', False)])
def test_automatic_source_quartile_calendar_years(db_session, proofs, year, kind, expected):
    employee = person(db_session, ['Доктор технических наук'])
    db_session.add(EmployeePublication(employee_id=employee.id, title='Automatic', year=year,
        publication_type=kind, source_hash='automatic', raw_data={'journal': 'Журнал',
            'csv_enrichment': {'Название журнала': 'Журнал', 'Квартиль scopus': 'Q2', 'Год': str(year)}}))
    candidate = result(db_session)
    assert candidate['confirmed_count'] == int(expected)
    assert not candidate['degree_confirmed']
    assert candidate['assessment_status'] == 'Требует проверки'
    article = candidate['articles'][0]
    assert len([m for m in article['messages'] if 'период' in m.lower()]) == 1
    if expected:
        assert article['confirmed_bases'][0]['basis'] == 'Q2'
        assert not any('Вид материала не подтверждён' in m for m in article['messages'])
    full_result = council.council_candidates(db_session, today=date(2026, 10, 8))
    book = load_workbook(BytesIO(council_xlsx(full_result, [full_result, council.council_candidates(db_session, full_result["end"], branch="физико-математические", paginate=False)])))
    headers = [cell.value for cell in book['Сотрудники'][1]]
    exported = dict(zip(headers, [cell.value for cell in book['Сотрудники'][2]]))
    assert exported['Q2'] == candidate['q2_count']
    assert exported['Всего'] == candidate['q2_count']


def test_automatic_quartile_identity_conflict_and_messages(db_session, proofs):
    employee = person(db_session, ['Доктор технических наук'])
    db_session.add(EmployeePublication(employee_id=employee.id, title='Conflict', year=2024,
        publication_type='ARTICLE', source_hash='conflict', doi_url='https://doi.org/10.1234/conflict',
        raw_data={'journal': 'Текущий журнал', 'csv_enrichment': {'Название журнала': 'Текущий журнал', 'Квартиль scopus': 'Q2'}},
        quartiles={'schema_version': 1, 'entries': [{'system': 'Scopus', 'quartile': 'Q1',
            'doi': '10.1234/conflict', 'journal': 'Другой журнал', 'publication_year': 2024}]}))
    article = result(db_session)['articles'][0]
    assert not article['confirmed']
    assert len(article['messages']) == len(set(article['messages']))
    issue = article['warnings']['journal'][0]
    assert 'Другой журнал' in issue and 'Текущий журнал' in issue
    assert article['messages'].count(issue) == 1
    app.dependency_overrides[get_db] = lambda: db_session
    try:
        response = TestClient(app).get(f'/admin/council/employees/{employee.id}/articles')
        from bs4 import BeautifulSoup
        card = BeautifulSoup(response.text, 'html.parser').select_one('li.council__article')
        assert sum(tip.get_text().count(issue) for tip in card.select('[role=tooltip]')) == 1
        assert 'Зачёт в подтверждённое число' not in card.get_text()
        assert 'Специальность для рассмотрения статьи не выбрана' not in card.get_text()
        assert 'Пункт:' not in response.text and 'Минимум:' not in response.text
    finally:
        app.dependency_overrides.clear()


def test_branch_selection_threshold_and_export(db_session, proofs):
    employee = person(db_session, ['Доктор технических наук'])
    proofs([{**SOURCE, 'degree': 'Доктор технических наук'}], [work(db_session, employee, n) for n in range(10)])
    db_session.commit()
    app.dependency_overrides[get_db] = lambda: db_session
    try:
        client = TestClient(app)
        for branch, expected in [('технические', 'Соответствует'), ('физико-математические', 'Требует проверки')]:
            response = client.get('/admin/council', params={'branch': branch})
            assert response.status_code == 200
            from bs4 import BeautifulSoup
            parsed = BeautifulSoup(response.text, 'html.parser')
            assert parsed.select_one('#council-branch option[selected]').get('value') == branch
            assert expected in parsed.select_one('.council__candidate').get_text()
            book = load_workbook(BytesIO(client.get('/admin/council/export.xlsx', params={'branch': branch}).content))
            headers = [cell.value for cell in book['Сотрудники'][1]]
            row = dict(zip(headers, [cell.value for cell in book['Сотрудники'][2]]))
            assert row['Всего'] == 10
            parameters = dict(book['Параметры'].values)
            assert parameters['Отрасль оценки'] == branch
            threshold = next(value for key, value in parameters.items() if key.startswith('Доктор · ') and branch in key.split(' · ')[1].split(', '))
            assert threshold.startswith('9 ' if branch == 'технические' else '11 ')
            assert parameters['Начало периода'] == datetime(2022, 1, 1)
            assert parameters['Конец периода'] == datetime(2026, 12, 31)
    finally:
        app.dependency_overrides.clear()


@pytest.mark.parametrize('category,metric,expected', [
    ('Applied Mathematics', 'CiteScore', True),
    ('Analysis', 'CiteScore', False),
    ('Analysis', 'SJR', True),
])
def test_automatic_quartile_categories_and_metrics_preserve_conflicts(db_session, proofs, category, metric, expected):
    employee = person(db_session, ['Доктор технических наук'])
    first = {'system': 'Scopus', 'quartile': 'Q1', 'category': 'Analysis', 'metric': 'CiteScore',
             'metric_year': 2024, 'publication_year': 2024, 'doi': '10.1234/categories',
             'journal': 'Journal of Mathematics', 'source': 'Journal metrics 2024',
             'source_url': 'https://example.org/metrics', 'checked_on': '2026-10-08'}
    second = {**first, 'quartile': 'Q2', 'category': category, 'metric': metric}
    db_session.add(EmployeePublication(employee_id=employee.id, title='Categorized article', year=2024,
        publication_type='ARTICLE', source_hash='categories', doi_url='https://doi.org/10.1234/categories',
        raw_data={'journal': first['journal']}, quartiles={'schema_version': 1, 'entries': [first, second]}))
    candidate = result(db_session)
    assert candidate['q1_count'] == candidate['q2_count'] == candidate['total_count'] == 1
    assert candidate['confirmed_count'] == int(expected)
    article = candidate['articles'][0]
    assert any(reason.startswith('Подтверждённые основания источников расходятся') for reason in article['confirmation_reasons']) == (not expected)
    assert [(basis.get('category'), basis.get('metric'), basis.get('metric_year'), basis['source'], basis['source_url'])
            for basis in article['automatic_quartile_bases']] == [
                ('Analysis', 'CiteScore', 2024, first['source'], first['source_url']),
                (category, metric, 2024, second['source'], second['source_url'])]


def test_warning_display_does_not_turn_unknown_type_into_confirmed_article(db_session, proofs):
    employee = person(db_session, ['Доктор технических наук'])
    db_session.add(EmployeePublication(employee_id=employee.id, title='Missing kind', year=2024,
        source_hash='missing', raw_data={'journal': 'Journal', 'csv_enrichment': {'Название журнала': 'Journal', 'Квартиль scopus': 'Q1'}}))
    candidate = result(db_session)
    article = candidate['articles'][0]
    assert article['display_type'] == 'Статья' and article['types'] == []
    assert article['warnings']['type'] and 'не указан тип материала' in article['warnings']['type'][0]
    assert not article['confirmed'] and candidate['confirmed_count'] == 0


@pytest.mark.parametrize("specialty,branches", [
    ("", ["технические", "физико-математические"]),
    ("1.1.2", ["физико-математические"]),
    ("2.3.2", ["технические"]),
    *[(code, ["технические", "физико-математические"]) for code in
      ["1.2.2", "1.3.8", "2.2.2", "2.2.13", "2.2.14", "2.2.15", "2.3.1", "2.3.5", "2.3.6", "2.3.7"]],
])
def test_specialty_limits_branch_options_and_rejects_invalid_requests(db_session, proofs, specialty, branches):
    from bs4 import BeautifulSoup
    employee = person(db_session, ["Доктор технических наук"])
    db_session.commit()
    app.dependency_overrides[get_db] = lambda: db_session
    try:
        client = TestClient(app)
        for branch in assessment.BRANCHES:
            params = {"specialty": specialty, "branch": branch}
            page = client.get("/admin/council", params=params)
            if branch in branches:
                assert page.status_code == 200
                options = BeautifulSoup(page.text, "html.parser").select("#council-branch option")
                assert [option["value"] for option in options] == branches
                assert next(option["value"] for option in options if "selected" in option.attrs) == branch
            else:
                assert page.status_code == 422
                assert client.get("/admin/council/export.xlsx", params=params).status_code == 422
                assert client.get(f"/admin/council/employees/{employee.id}/articles", params=params).status_code == 422
    finally:
        app.dependency_overrides.clear()


@pytest.mark.parametrize("previous,specialty,old_branch,new_branch", [
    ("2.3.2", "1.1.2", "технические", "физико-математические"),
    ("1.1.2", "2.3.2", "физико-математические", "технические"),
    ("", "1.1.2", "технические", "физико-математические"),
])
def test_specialty_change_without_js_uses_compatible_branch(db_session, proofs, previous, specialty, old_branch, new_branch):
    from bs4 import BeautifulSoup
    app.dependency_overrides[get_db] = lambda: db_session
    try:
        client = TestClient(app)
        params = {"previous_specialty": previous, "specialty": specialty, "branch": old_branch,
                  "q": "Тест", "sort": "full_name", "direction": "asc"}
        page = client.get("/admin/council", params=params)
        assert page.status_code == 200
        parsed = BeautifulSoup(page.text, "html.parser")
        assert parsed.select_one('#council-branch option[selected]')["value"] == new_branch
        assert parsed.select_one('[name=previous_specialty]')["value"] == specialty
        assert parsed.select_one('[name=q]')["value"] == "Тест"
        book = load_workbook(BytesIO(client.get("/admin/council/export.xlsx", params=params).content))
        assert dict(book["Параметры"].values)["Отрасль оценки"] == new_branch
        for invalid_previous in [None, specialty, "oops"]:
            invalid = {**params, "previous_specialty": invalid_previous}
            if invalid_previous is None:
                del invalid["previous_specialty"]
            assert client.get("/admin/council", params=invalid).status_code == 422
            assert client.get("/admin/council/export.xlsx", params=invalid).status_code == 422
        invalid = {**params, "branch": "экономические"}
        assert client.get("/admin/council", params=invalid).status_code == 422
    finally:
        app.dependency_overrides.clear()