70 lines
3.7 KiB
Python
70 lines
3.7 KiB
Python
from scripts.extract_k1_specialties import active_codes, build_reference, parse_rows
|
||
import pytest
|
||
|
||
|
||
def test_vak_continuation_and_historical_dates():
|
||
rows = [
|
||
{"page": 1, "cells": ["1.", "Journal", "1234-5678", "1.2.2. First", "с 01.02.2022\nпо 01.02.2023"]},
|
||
{"page": 2, "cells": ["", "continued", "", "2.3.5. Second", ""]},
|
||
{"page": 2, "cells": ["", "", "", "2.3.6. Current", "с 02.02.2023"]},
|
||
{"page": 3, "cells": ["", "", "", "2.3.1. Future", "с 01.02.2027"]},
|
||
]
|
||
journal = parse_rows(rows)[0]
|
||
assert journal["name"] == "Journal continued" and journal["pages"] == [1, 2, 3]
|
||
assert journal["assignments"][0]["codes"] == ["1.2.2", "2.3.5"]
|
||
assert active_codes(journal, "2026-09-15") == ["2.3.6"]
|
||
assert active_codes(journal, "2023-02-01") == ["1.2.2", "2.3.5"]
|
||
with pytest.raises(ValueError, match="numbering"):
|
||
parse_rows([{"page": 1, "cells": ["2.", "Journal", "1234-5678", "1.2.2. Test", "с 01.02.2022"]}])
|
||
|
||
|
||
def test_ambiguous_titles_do_not_union_specialties_and_reviewed_matches_are_checked():
|
||
rows = [{"page": n, "cells": [f"{n}.", "Same title", str(n), code + ". Test", "с 01.02.2022"]}
|
||
for n, code in [(1, "1.2.2"), (2, "2.3.1")]]
|
||
journals = parse_rows(rows)
|
||
k1 = {"journals": [{"name": "Same title"}, {"name": "Similar title"}]}
|
||
result = build_reference(journals, k1, [], "2026-09-15")
|
||
assert result[0]["status"] == "ambiguous" and result[0]["specialty_codes"] == []
|
||
assert result[1]["status"] == "not_found" and result[1]["specialty_codes"] == []
|
||
with pytest.raises(ValueError, match="title changed"):
|
||
build_reference(journals, k1, [{"k1_name": "Similar title", "vak_row": 1, "vak_name": "Wrong title"}], "2026-09-15")
|
||
|
||
|
||
def test_malformed_source_start_date_is_preserved_without_inventing_date():
|
||
journal = parse_rows([{"page": 1, "cells": ["1.", "Journal", "1234-5678", "1.3.8. Test", "с 01.02.200"]}])[0]
|
||
assignment = journal["assignments"][0]
|
||
assert assignment["source_dates"] == "с 01.02.200"
|
||
assert assignment["valid_from"] is None and assignment["date_note"]
|
||
assert active_codes(journal, "2026-09-15") == ["1.3.8"]
|
||
|
||
|
||
def test_k1_extraction_is_reproducible_for_lf_and_crlf(tmp_path):
|
||
import hashlib
|
||
import json
|
||
import subprocess
|
||
import sys
|
||
source = tmp_path / 'list.pdf'
|
||
source.write_bytes(b'PDF source bytes for cached-row extraction')
|
||
rows = tmp_path / 'rows.json'
|
||
rows.write_text(json.dumps([{'page': 1, 'cells': ['1.', 'Journal', '1234-5678', '1.2.2. Test', 'с 01.02.2022']}]), encoding='utf8')
|
||
matches = tmp_path / 'matches.json'
|
||
matches.write_text('[]', encoding='utf8')
|
||
canonical = (json.dumps({'journals': [{'name': 'Journal'}]}, indent=2) + '\n').encode('utf8')
|
||
outputs = []
|
||
for name, content in [('lf', canonical), ('crlf', canonical.replace(b'\n', b'\r\n'))]:
|
||
directory = tmp_path / name
|
||
directory.mkdir()
|
||
k1 = directory / 'k1.json'
|
||
k1.write_bytes(content)
|
||
output = directory / 'specialties.json'
|
||
subprocess.run([sys.executable, 'scripts/extract_k1_specialties.py', str(source), '--k1', str(k1),
|
||
'--matches', str(matches), '--rows-cache', str(rows), '--output', str(output)], check=True, capture_output=True)
|
||
data = json.loads(output.read_text(encoding='utf8'))
|
||
assert data['k1_source_sha256'] == hashlib.sha256(canonical).hexdigest()
|
||
assert data['source_sha256'] == hashlib.sha256(source.read_bytes()).hexdigest()
|
||
assert data['journals'][0]['specialty_codes'] == ['1.2.2']
|
||
assert k1.read_bytes() == content
|
||
assert b'\r\n' not in output.read_bytes()
|
||
outputs.append(output.read_bytes())
|
||
assert outputs[0] == outputs[1]
|