f79384e2b1
Функция принимает уже прочитанные json-объекты и сама ничего не открывает — транспорт (локальная ФС, SMB-шара, S3, поток) остаётся за вызывающим. Локальное чтение переехало в CLI как одного из них. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JvCSGmwgL8juhZdXr32qXM
93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
import json
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from src.core.merge_results import merge_documents
|
||
|
||
DOC_A = {
|
||
'file_uid': 'uid-a',
|
||
'error': {'code': '', 'message': ''},
|
||
'data_ocr': {'label': {'form_name': 'Бухгалтерский баланс'}, 'table_structure': []},
|
||
}
|
||
DOC_B = {
|
||
'file_uid': 'uid-b',
|
||
'error': {'code': '', 'message': ''},
|
||
'data_ocr': {'label': {'form_name': 'Отчёт о финансовых результатах'}, 'table_structure': []},
|
||
}
|
||
|
||
|
||
def test_merges_two_documents_into_one_result():
|
||
result = merge_documents([DOC_A, DOC_B], file_id='file-1', legal_entity_id='le-1')
|
||
|
||
assert result == {
|
||
'file_id': 'file-1',
|
||
'legal_entity_id': 'le-1',
|
||
'files': [DOC_A, DOC_B],
|
||
}
|
||
|
||
|
||
def test_does_no_io_at_all(monkeypatch):
|
||
# функция не должна ничего открывать — чтение целиком на вызывающем
|
||
def explode(*args, **kwargs):
|
||
raise AssertionError('merge_documents не должна трогать файловую систему')
|
||
|
||
monkeypatch.setattr('builtins.open', explode)
|
||
|
||
result = merge_documents([DOC_A], file_id='f', legal_entity_id='l')
|
||
|
||
assert result['files'] == [DOC_A]
|
||
|
||
|
||
def test_keeps_document_content_untouched():
|
||
# содержимое кладётся как есть: незнакомые поля и вложенность не теряются
|
||
weird = {'file_uid': 'uid-c', 'unexpected': [1, {'deep': None}], 'кириллица': 'значение'}
|
||
|
||
result = merge_documents([weird], file_id='file-1', legal_entity_id='le-1')
|
||
|
||
assert result['files'] == [weird]
|
||
|
||
|
||
def test_preserves_given_order():
|
||
result = merge_documents([DOC_B, DOC_A], file_id='f', legal_entity_id='l')
|
||
|
||
assert [doc['file_uid'] for doc in result['files']] == ['uid-b', 'uid-a']
|
||
|
||
|
||
def test_accepts_any_iterable():
|
||
# вызывающий может отдавать генератор, например потоковое чтение с шары
|
||
result = merge_documents((d for d in (DOC_A, DOC_B)), file_id='f', legal_entity_id='l')
|
||
|
||
assert len(result['files']) == 2
|
||
|
||
|
||
def test_rejects_empty_input():
|
||
with pytest.raises(ValueError):
|
||
merge_documents([], file_id='f', legal_entity_id='l')
|
||
|
||
|
||
def test_cli_reads_local_files_and_writes_merged(tmp_path):
|
||
# CLI — один из вызывающих: локальное чтение живёт здесь, а не в merge_documents
|
||
a = tmp_path / 'doc_1.json'
|
||
a.write_text(json.dumps(DOC_A, ensure_ascii=False), encoding='utf-8')
|
||
b = tmp_path / 'doc_2.json'
|
||
b.write_text(json.dumps(DOC_B, ensure_ascii=False), encoding='utf-8')
|
||
out = tmp_path / 'merged.json'
|
||
|
||
completed = subprocess.run(
|
||
[sys.executable, '-m', 'src.core.merge_results', str(a), str(b),
|
||
'--file-id', 'file-1', '--legal-entity-id', 'le-1', '-o', str(out)],
|
||
cwd=Path(__file__).resolve().parent.parent,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
|
||
assert completed.returncode == 0, completed.stderr
|
||
assert json.loads(out.read_text(encoding='utf-8')) == {
|
||
'file_id': 'file-1',
|
||
'legal_entity_id': 'le-1',
|
||
'files': [DOC_A, DOC_B],
|
||
}
|