203 lines
12 KiB
Python
203 lines
12 KiB
Python
from datetime import date
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import select
|
||
|
||
from app.config import Settings, get_settings
|
||
from app.db import get_db
|
||
from app.main import app
|
||
from app.models import CouncilCorrections, Employee, EmployeePublication
|
||
from app.security import SESSION_COOKIE, sign_session
|
||
from app.services import council
|
||
|
||
|
||
@pytest.fixture()
|
||
def editor(db_session, monkeypatch):
|
||
monkeypatch.setattr(council, '_articles_by_author', lambda: {})
|
||
for number in range(2):
|
||
person = Employee(profile_key=f'editor:{number}', full_name=f'Иванов {number}',
|
||
canonical_url='https://www.hse.ru/staff/test',
|
||
current_data={'sections': [{'title': 'Учёные степени', 'items': ['Доктор технических наук']}]})
|
||
db_session.add(person)
|
||
db_session.flush()
|
||
db_session.add(EmployeePublication(employee_id=person.id, title=f'Статья {number}', year=2024,
|
||
publication_type='ARTICLE', source_hash=f'editor:{number}', raw_data={'journal': 'Общий журнал'}))
|
||
db_session.commit()
|
||
settings = Settings(_env_file=None, admin_username='admin', admin_password='test', session_secret='editor-test')
|
||
app.dependency_overrides[get_db] = lambda: db_session
|
||
app.dependency_overrides[get_settings] = lambda: settings
|
||
client = TestClient(app)
|
||
client.cookies.set(SESSION_COOKIE, sign_session('admin', settings))
|
||
candidate = council.council_candidates(db_session)['candidates'][0]
|
||
article = candidate['articles'][0]
|
||
payload = {'employee_id': candidate['employee_id'], 'edit_key': article['edit_key'], 'revision': 0,
|
||
'title': 'Исправленная статья', 'journal': 'Общий журнал', 'quartile': 'Q1'}
|
||
try:
|
||
yield client, payload
|
||
finally:
|
||
client.close()
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
def save(client, payload):
|
||
return client.post('/admin/council/publications', json=payload, headers={'X-Council-Edit': '1'})
|
||
|
||
|
||
def test_editor_shared_quartile_persistence_filter_and_export(editor, db_session):
|
||
client, payload = editor
|
||
assert save(client, payload).status_code == 200
|
||
db_session.expire_all()
|
||
result = council.council_candidates(db_session, today=date(2026, 10, 9))
|
||
assert result['revision'] == 1
|
||
assert [c['q1_count'] for c in result['candidates']] == [1, 1]
|
||
articles = [a for c in result['candidates'] for a in c['articles']]
|
||
assert all(a['manual_review'] and not a['confirmed'] for a in articles)
|
||
assert next(a for a in articles if a['edit_key'] == payload['edit_key'] and a['title'] == payload['title'])
|
||
assert db_session.scalars(select(EmployeePublication.title)).all() == ['Статья 0', 'Статья 1']
|
||
for filters in ({'title': 'исправленная'}, {'employee': 'иванов 0'}, {'journal': 'общий ЖУРНАЛ'}):
|
||
response = client.get('/admin/council/publications', params=filters)
|
||
assert response.status_code == 200
|
||
assert payload['title'] in response.text
|
||
assert 'Статьи не найдены' in client.get('/admin/council/publications?title=absent').text
|
||
main = client.get('/admin/council')
|
||
assert 'data-council-edit' not in main.text and payload['title'] not in main.text
|
||
fragment = client.get(f"/admin/council/employees/{payload['employee_id']}/articles")
|
||
assert 'data-council-edit' in fragment.text and payload['title'] in fragment.text
|
||
from io import BytesIO
|
||
from openpyxl import load_workbook
|
||
workbook = load_workbook(BytesIO(client.get('/admin/council/export.xlsx').content))
|
||
assert any(payload['title'] in str(row) for row in workbook['Публикации'].values)
|
||
# A stale window must not overwrite another save.
|
||
assert save(client, {**payload, 'title': 'Устаревшая правка'}).status_code == 409
|
||
assert db_session.get(CouncilCorrections, 1).revision == 1
|
||
|
||
|
||
def test_editor_permissions_csrf_validation_and_missing_article(editor, db_session):
|
||
client, payload = editor
|
||
assert client.post('/admin/council/publications', json=payload).status_code == 403
|
||
for values in ({'title': ' '}, {'journal': ' '}, {'journal': '!!!'}, {'quartile': 'Q5'}, {'quartile': '<script>'}, {'revision': -1}):
|
||
assert save(client, {**payload, **values}).status_code == 422
|
||
assert save(client, {**payload, 'edit_key': '0' * 64}).status_code == 404
|
||
assert db_session.get(CouncilCorrections, 1) is None
|
||
client.cookies.clear()
|
||
assert save(client, payload).status_code == 401
|
||
original_title = db_session.scalar(select(EmployeePublication.title).where(
|
||
EmployeePublication.employee_id == payload['employee_id']))
|
||
response = client.get('/admin/council/publications')
|
||
assert response.status_code == 200
|
||
assert original_title in response.text
|
||
assert 'data-council-edit' not in response.text and 'data-council-dialog' not in response.text
|
||
fragment = client.get(f"/admin/council/employees/{payload['employee_id']}/articles")
|
||
assert fragment.status_code == 200 and original_title in fragment.text
|
||
assert 'data-council-article' not in fragment.text and 'data-council-edit' not in fragment.text
|
||
|
||
|
||
def test_editor_journal_reassignment_year_scope_and_unknown_year(editor, db_session):
|
||
client, payload = editor
|
||
old = db_session.scalar(select(EmployeePublication).where(EmployeePublication.employee_id != payload['employee_id']))
|
||
old.year = 2023
|
||
db_session.commit()
|
||
assert save(client, {**payload, 'journal': 'Новый журнал', 'quartile': 'Q2'}).status_code == 200
|
||
db_session.expire_all()
|
||
result = council.council_candidates(db_session)
|
||
corrected = next(c for c in result['candidates'] if c['employee_id'] == payload['employee_id'])
|
||
assert corrected['q2_count'] == 1
|
||
assert corrected['articles'][0]['journals'] == ['Новый журнал']
|
||
untouched = next(c for c in result['candidates'] if c['employee_id'] != payload['employee_id'])
|
||
assert untouched['q2_count'] == 0
|
||
assert untouched['articles'][0]['journals'] == ['Общий журнал']
|
||
old.year = None
|
||
db_session.commit()
|
||
article = next(c for c in council.council_candidates(db_session)['candidates'] if c['employee_id'] == old.employee_id)['articles'][0]
|
||
unknown = {**payload, 'employee_id': old.employee_id, 'edit_key': article['edit_key'], 'revision': 1}
|
||
assert save(client, unknown).status_code == 422
|
||
assert save(client, {**unknown, 'quartile': ''}).status_code == 200
|
||
|
||
|
||
def test_editor_csv_only_and_escaping(editor, db_session, monkeypatch):
|
||
client, payload = editor
|
||
monkeypatch.setattr(council, '_articles_by_author', lambda: {'иванов 0': [{
|
||
'title': 'Выгрузка', 'year': 2024, 'doi': '10.1234/csv', 'types': [], 'journals': [],
|
||
'quartiles': [], 'origins': ['Данные ВШЭ'], 'rows': [2], 'in_csv': True, 'in_db': False, 'url': None}]})
|
||
article = next(a for c in council.council_candidates(db_session)['candidates']
|
||
for a in c['articles'] if a['doi'] == '10.1234/csv')
|
||
payload.update(edit_key=article['edit_key'], title='<script>alert(1)</script>', journal='Журнал', quartile='Q4')
|
||
assert save(client, payload).status_code == 200
|
||
page = client.get('/admin/council/publications')
|
||
assert '<script>alert(1)</script>' not in page.text
|
||
assert '<script>' in page.text
|
||
|
||
|
||
def test_editor_reassignment_clears_old_journal_evidence(editor, db_session):
|
||
client, payload = editor
|
||
source = db_session.scalar(select(EmployeePublication).where(EmployeePublication.employee_id == payload['employee_id']))
|
||
source.raw_data = {'journal': 'Бизнес-информатика', 'issn': '1998-0663'}
|
||
db_session.commit()
|
||
before = next(c for c in council.council_candidates(db_session)['candidates'] if c['employee_id'] == payload['employee_id'])
|
||
assert before['k1_count'] == 1
|
||
assert save(client, {**payload, 'journal': 'Неизвестный журнал', 'quartile': 'Q4'}).status_code == 200
|
||
after = next(c for c in council.council_candidates(db_session)['candidates'] if c['employee_id'] == payload['employee_id'])
|
||
assert (after['k1_count'], after['q1_count'], after['q2_count']) == (0, 0, 0)
|
||
assert after['articles'][0]['catalog_records'] == []
|
||
assert not after['articles'][0]['specialty_codes']
|
||
assert source.raw_data['issn'] == '1998-0663'
|
||
|
||
|
||
def test_editor_pagination_preserves_filters(editor, db_session):
|
||
client, payload = editor
|
||
for number in range(55):
|
||
db_session.add(EmployeePublication(employee_id=payload['employee_id'], title=f'Пагинация {number:02}',
|
||
year=2024, source_hash=f'page:{number}', raw_data={'journal': 'Тестовый журнал'}))
|
||
db_session.commit()
|
||
response = client.get('/admin/council/publications?employee=Иванов&title=Пагинация&journal=Тестовый&offset=999')
|
||
assert response.status_code == 200
|
||
assert 'Показаны 51–55' in response.text
|
||
assert response.text.count('data-council-article') == 5
|
||
assert 'offset=0' in response.text
|
||
assert 'name="employee" value="Иванов"' in response.text
|
||
|
||
|
||
def test_editor_database_failure_preserves_data(editor, db_session, monkeypatch):
|
||
from sqlalchemy.exc import OperationalError
|
||
client, payload = editor
|
||
def unavailable():
|
||
raise OperationalError('UPDATE', {}, Exception('private database details'))
|
||
monkeypatch.setattr(db_session, 'commit', unavailable)
|
||
response = save(client, payload)
|
||
assert response.status_code == 503
|
||
assert 'private database details' not in response.text
|
||
assert db_session.get(CouncilCorrections, 1) is None
|
||
|
||
|
||
def test_manual_specialty_shared_filter_remove_and_reset(editor, db_session):
|
||
client, payload = editor
|
||
assert save(client, {**payload, 'specialty_codes': ['<script>']}).status_code == 422
|
||
assert save(client, {**payload, 'specialty_codes': ['1.2.2', '1.2.2']}).status_code == 200
|
||
result = council.council_candidates(db_session, specialty='1.2.2')
|
||
assert result['matched'] == 2
|
||
assert all(c['specialty_count'] == 1 and c['articles'][0]['manual_specialties'] for c in result['candidates'])
|
||
assert result['candidates'][0]['articles'][0]['edit_specialty_codes'] == ['1.2.2']
|
||
# Empty selection removes the journal's thematic association.
|
||
assert save(client, {**payload, 'revision': 1, 'specialty_codes': []}).status_code == 200
|
||
assert council.council_candidates(db_session, specialty='1.2.2')['matched'] == 0
|
||
# The automatic option restores the source catalog association.
|
||
assert save(client, {**payload, 'revision': 2, 'specialty_codes': None}).status_code == 200
|
||
assert 'общий журнал' not in db_session.get(CouncilCorrections, 1).data['specialties']
|
||
|
||
|
||
def test_specialty_only_edit_preserves_automatic_confirmation(editor, db_session):
|
||
client, payload = editor
|
||
source = db_session.scalar(select(EmployeePublication).where(EmployeePublication.employee_id == payload['employee_id']))
|
||
source.raw_data = {'journal': 'Общий журнал', 'csv_enrichment': {'Название журнала': 'Общий журнал', 'Квартиль scopus': 'Q1'}}
|
||
db_session.commit()
|
||
candidate = next(c for c in council.council_candidates(db_session)['candidates'] if c['employee_id'] == payload['employee_id'])
|
||
assert candidate['confirmed_count'] == 1
|
||
payload.update(title=source.title, specialty_codes=['1.2.2'])
|
||
assert save(client, payload).status_code == 200
|
||
updated = next(c for c in council.council_candidates(db_session)['candidates'] if c['employee_id'] == payload['employee_id'])
|
||
assert updated['confirmed_count'] == 1
|
||
assert updated['articles'][0]['manual_specialties']
|
||
assert not updated['articles'][0]['manual_review']
|
||
assert db_session.get(CouncilCorrections, 1).data['journals'] == {}
|