import pytest
|
from fastapi import HTTPException
|
|
from app.import_service import (
|
_collect_exam_ids,
|
_parse_exam_to_paper_sql_map,
|
_resolve_exam_paper_map,
|
)
|
|
|
def test_collect_exam_ids_ok():
|
rows = [{"exam_id": 102}, {"exam_id": 101}, {"exam_id": 101}]
|
assert _collect_exam_ids(rows) == [101, 102]
|
|
|
def test_collect_exam_ids_empty():
|
with pytest.raises(HTTPException) as exc:
|
_collect_exam_ids([{"exam_id": None}, {}])
|
assert exc.value.status_code == 400
|
|
|
def test_parse_exam_to_paper_sql_map():
|
sql = """
|
INSERT INTO examination_exam (id, paper_source_id, name) VALUES (101, 9001, 'a');
|
INSERT INTO examination_exam (id, paper_id, name) VALUES (102, 9002, 'b');
|
"""
|
m = _parse_exam_to_paper_sql_map(sql)
|
assert m == {101: 9001, 102: 9002}
|
|
|
def test_resolve_single_paper_multi_exam():
|
paper_rows = [
|
{"id": 9001, "external_paper_id": "P1", "revision": "1"},
|
]
|
exam_rows_exam_ids = [101, 102]
|
files: dict[str, bytes] = {}
|
m = _resolve_exam_paper_map(exam_rows_exam_ids, paper_rows, files)
|
assert m == {101: 9001, 102: 9001}
|
|
|
def test_resolve_multi_paper_requires_exam_sql():
|
paper_rows = [
|
{"id": 9001, "external_paper_id": "P1", "revision": "1"},
|
{"id": 9002, "external_paper_id": "P2", "revision": "1"},
|
]
|
with pytest.raises(HTTPException) as exc:
|
_resolve_exam_paper_map([101, 102], paper_rows, {})
|
assert "examination_exam" in exc.value.detail
|
|
|
def test_resolve_multi_paper_with_exam_sql():
|
paper_rows = [
|
{"id": 9001, "external_paper_id": "P1", "revision": "1"},
|
{"id": 9002, "external_paper_id": "P2", "revision": "1"},
|
]
|
exam_sql = """
|
INSERT INTO examination_exam (id, paper_source_id) VALUES (101, 9001);
|
INSERT INTO examination_exam (id, paper_source_id) VALUES (102, 9002);
|
"""
|
files = {"examination_exam.sql": exam_sql.encode()}
|
m = _resolve_exam_paper_map([101, 102], paper_rows, files)
|
assert m == {101: 9001, 102: 9002}
|
