125 lines
6.2 KiB
Python
125 lines
6.2 KiB
Python
import csv
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.services import council
|
|
from app.services.council_source import load_reference, prepare_reference
|
|
|
|
|
|
def write_csv(path, rows):
|
|
fields = ['Название публикации', 'Год', 'DOI', 'Авторы НИУ ВШЭ', 'Квартиль scopus', 'Квартиль wos', 'Название журнала', 'EID (Scopus)']
|
|
with path.open('w', encoding='utf-8-sig', newline='') as stream:
|
|
writer = csv.DictWriter(stream, fieldnames=fields, delimiter=';')
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
|
|
|
|
def test_prepared_reference_preserves_quoted_fields_metadata_and_matching(tmp_path):
|
|
source = tmp_path / 'source.csv'
|
|
write_csv(source, [
|
|
{'Название публикации': 'Quoted; title\ncontinuation', 'Год': '2024', 'DOI': 'https://doi.org/10.1234/TEST',
|
|
'Авторы НИУ ВШЭ': ' Ёлкин Пётр; ЕЛКИН ПЕТР; Другой автор ', 'Квартиль scopus': ' q1 ',
|
|
'Квартиль wos': 'Q2', 'Название журнала': 'Бизнес-информатика', 'EID (Scopus)': 'source-id'},
|
|
{'Название публикации': '', 'Авторы НИУ ВШЭ': 'Другой автор'},
|
|
{'Название публикации': 'Unknown', 'Год': '', 'DOI': 'invalid', 'Авторы НИУ ВШЭ': ''},
|
|
])
|
|
before = source.read_bytes()
|
|
data = prepare_reference(source)
|
|
assert source.read_bytes() == before
|
|
assert data['source_rows'] == 3
|
|
assert data['source_sha256'] == hashlib.sha256(before.replace(b'\r\n', b'\n')).hexdigest()
|
|
assert data['by_author'] == {'другой автор': [0], 'елкин петр': [0]}
|
|
record = data['publications'][0]
|
|
assert record['title'] == 'Quoted; title\ncontinuation'
|
|
assert record['doi'] == '10.1234/test' and record['year'] == 2024
|
|
assert record['quartiles'] == ['Scopus Q1', 'Web of Science Q2']
|
|
assert record['journals'] == ['Бизнес-информатика']
|
|
assert record['metadata']['EID (Scopus)'] == 'source-id'
|
|
output = tmp_path / 'data.json'
|
|
output.write_text(json.dumps(data, ensure_ascii=False), encoding='utf-8')
|
|
articles = load_reference(output)
|
|
assert articles['елкин петр'][0]['rows'] == [2]
|
|
assert articles['другой автор'][0]['title'] == record['title']
|
|
assert len(articles['елкин петр']) == 1
|
|
assert articles['елкин петр'][0]['origins'] == ['Данные ВШЭ']
|
|
|
|
|
|
@pytest.mark.parametrize('change', ['author', 'index', 'bool_index', 'year', 'schema', 'source_row', 'list'])
|
|
def test_corrupt_prepared_reference_is_rejected(tmp_path, change):
|
|
source = tmp_path / 'source.csv'
|
|
write_csv(source, [{'Название публикации': 'First', 'Авторы НИУ ВШЭ': 'Author'},
|
|
{'Название публикации': 'Second', 'Авторы НИУ ВШЭ': 'Other'}])
|
|
data = prepare_reference(source)
|
|
if change == 'author': data['by_author'] = {'stranger': [0], 'other': [1]}
|
|
if change == 'index': data['by_author']['author'] = [3]
|
|
if change == 'bool_index': data['by_author']['other'] = [True]
|
|
if change == 'year': data['publications'][0]['year'] = True
|
|
if change == 'schema': data['schema_version'] = True
|
|
if change == 'source_row': data['publications'][0]['source_row'] = 77
|
|
if change == 'list': data['publications'][0]['journals'] = 'Journal'
|
|
path = tmp_path / 'data.json'
|
|
path.write_text(json.dumps(data), encoding='utf8')
|
|
with pytest.raises(ValueError): load_reference(path)
|
|
|
|
|
|
def test_committed_reference_is_reproducible_and_loaded_once(monkeypatch):
|
|
source = council.EXPORT.with_suffix('.csv')
|
|
data = json.loads(council.EXPORT.read_text(encoding='utf-8'))
|
|
assert data == prepare_reference(source)
|
|
assert data['source_rows'] == 1000 and len(data['by_author']) == 776
|
|
calls = []
|
|
original = council.load_reference
|
|
def counted(path):
|
|
calls.append(path)
|
|
return original(path)
|
|
monkeypatch.setattr(council, 'load_reference', counted)
|
|
council._articles_by_author.cache_clear()
|
|
try:
|
|
first = council._articles_by_author()
|
|
assert council._articles_by_author() is first
|
|
assert calls == [council.EXPORT]
|
|
finally:
|
|
council._articles_by_author.cache_clear()
|
|
|
|
|
|
def test_converter_does_not_replace_output_when_input_is_invalid(tmp_path):
|
|
import subprocess
|
|
import sys
|
|
source = tmp_path / 'invalid.csv'
|
|
source.write_text('Wrong header\nvalue\n', encoding='utf8')
|
|
output = tmp_path / 'prepared.json'
|
|
output.write_text('previous reference', encoding='utf8')
|
|
result = subprocess.run([sys.executable, 'scripts/prepare_council_articles.py', str(source), '--output', str(output)],
|
|
capture_output=True, text=True)
|
|
assert result.returncode != 0
|
|
assert output.read_text(encoding='utf8') == 'previous reference'
|
|
assert not list(tmp_path.glob('*.tmp'))
|
|
same = subprocess.run([sys.executable, 'scripts/prepare_council_articles.py', str(source), '--output', str(source)],
|
|
capture_output=True, text=True)
|
|
assert same.returncode != 0 and source.read_text(encoding='utf8') == 'Wrong header\nvalue\n'
|
|
|
|
|
|
def test_preparation_is_identical_for_lf_and_crlf_sources(tmp_path):
|
|
lf_dir = tmp_path / 'lf'
|
|
crlf_dir = tmp_path / 'crlf'
|
|
lf_dir.mkdir()
|
|
crlf_dir.mkdir()
|
|
lf = lf_dir / 'source.csv'
|
|
crlf = crlf_dir / 'source.csv'
|
|
write_csv(crlf, [{'Название публикации': 'Multiline\npublication', 'Год': '2024',
|
|
'Авторы НИУ ВШЭ': 'Author; Coauthor', 'Квартиль scopus': 'Q1'}])
|
|
canonical = crlf.read_bytes().replace(b'\r\n', b'\n')
|
|
lf.write_bytes(canonical)
|
|
crlf.write_bytes(canonical.replace(b'\n', b'\r\n'))
|
|
before = crlf.read_bytes()
|
|
from_lf = prepare_reference(lf)
|
|
from_crlf = prepare_reference(crlf)
|
|
assert from_lf == from_crlf
|
|
assert from_crlf['source_sha256'] == hashlib.sha256(canonical).hexdigest()
|
|
assert from_crlf['source_hash_eol'] == 'LF'
|
|
assert from_crlf['publications'][0]['title'] == 'Multiline\npublication'
|
|
assert crlf.read_bytes() == before
|