전 저장소 리팩터링과 SSOT 정비
This commit is contained in:
parent
14ecbd4e7d
commit
3dfddcac6f
173 changed files with 19679 additions and 6952 deletions
99
apps/api/app/test_tabular_ingest.py
Normal file
99
apps/api/app/test_tabular_ingest.py
Normal file
|
|
@ -0,0 +1,99 @@
|
|||
"""P4 자유 양식 엑셀/CSV → 텍스트 변환 테스트."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import unittest
|
||||
|
||||
from .services.tabular_ingest import (
|
||||
MAX_UPLOAD_BYTES,
|
||||
TabularIngestError,
|
||||
extract_tabular_text,
|
||||
)
|
||||
|
||||
|
||||
def _xlsx_bytes(rows_by_sheet: dict[str, list[list[object]]]) -> bytes:
|
||||
import openpyxl
|
||||
|
||||
workbook = openpyxl.Workbook()
|
||||
default = workbook.active
|
||||
first = True
|
||||
for sheet_name, rows in rows_by_sheet.items():
|
||||
if first:
|
||||
sheet = default
|
||||
sheet.title = sheet_name
|
||||
first = False
|
||||
else:
|
||||
sheet = workbook.create_sheet(sheet_name)
|
||||
for row in rows:
|
||||
sheet.append(row)
|
||||
buffer = io.BytesIO()
|
||||
workbook.save(buffer)
|
||||
return buffer.getvalue()
|
||||
|
||||
|
||||
class TabularIngestTest(unittest.TestCase):
|
||||
def test_xlsx_label_value_rows_become_pairs(self) -> None:
|
||||
data = _xlsx_bytes(
|
||||
{
|
||||
"추고록": [
|
||||
["이름", "김하늘"],
|
||||
["주호소", "시험 전 복통과 불안, 성적 하락 후 심화"],
|
||||
[None, None],
|
||||
["상담 목표", "불안 대처와 자기 이해"],
|
||||
]
|
||||
}
|
||||
)
|
||||
text = extract_tabular_text(filename="자유양식.xlsx", data=data)
|
||||
self.assertIn("## 시트: 추고록", text)
|
||||
self.assertIn("이름: 김하늘", text)
|
||||
self.assertIn("주호소: 시험 전 복통과 불안, 성적 하락 후 심화", text)
|
||||
|
||||
def test_xlsx_multi_cell_rows_join_with_pipe(self) -> None:
|
||||
data = _xlsx_bytes(
|
||||
{
|
||||
"Sheet1": [
|
||||
["회차", "날짜", "내용 요약"],
|
||||
[1, "2026-05-02", "첫 면담, 라포 형성 시도했으나 침묵이 길었음"],
|
||||
]
|
||||
}
|
||||
)
|
||||
text = extract_tabular_text(filename="log.xlsx", data=data)
|
||||
self.assertIn("회차 | 날짜 | 내용 요약", text)
|
||||
self.assertIn("1 | 2026-05-02 | 첫 면담, 라포 형성 시도했으나 침묵이 길었음", text)
|
||||
|
||||
def test_csv_cp949_fallback(self) -> None:
|
||||
csv_text = "라벨,값\n주호소,불안과 무기력이 이어지고 있음\n"
|
||||
data = csv_text.encode("cp949")
|
||||
text = extract_tabular_text(filename="notes.csv", data=data)
|
||||
self.assertIn("주호소: 불안과 무기력이 이어지고 있음", text)
|
||||
|
||||
def test_legacy_xls_rejected_with_guidance(self) -> None:
|
||||
with self.assertRaises(TabularIngestError) as caught:
|
||||
extract_tabular_text(filename="old.xls", data=b"anything")
|
||||
self.assertIn(".xlsx", str(caught.exception))
|
||||
|
||||
def test_unknown_extension_rejected(self) -> None:
|
||||
with self.assertRaises(TabularIngestError):
|
||||
extract_tabular_text(filename="notes.hwp", data=b"1234")
|
||||
|
||||
def test_empty_file_rejected(self) -> None:
|
||||
with self.assertRaises(TabularIngestError):
|
||||
extract_tabular_text(filename="a.xlsx", data=b"")
|
||||
|
||||
def test_oversize_rejected(self) -> None:
|
||||
with self.assertRaises(TabularIngestError):
|
||||
extract_tabular_text(filename="a.xlsx", data=b"0" * (MAX_UPLOAD_BYTES + 1))
|
||||
|
||||
def test_nearly_empty_sheet_rejected(self) -> None:
|
||||
data = _xlsx_bytes({"Sheet1": [["a"]]})
|
||||
with self.assertRaises(TabularIngestError):
|
||||
extract_tabular_text(filename="empty.xlsx", data=data)
|
||||
|
||||
def test_corrupt_xlsx_rejected(self) -> None:
|
||||
with self.assertRaises(TabularIngestError):
|
||||
extract_tabular_text(filename="broken.xlsx", data=b"not a zip at all")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue