chatkit/backend/tests/test_csv_strict.py

"""Tests for the StrictCsvParser."""

from __future__ import annotations

import pytest

from app.parsers import get_parser, pick_auto
from app.parsers.base import ParserFile


def _file(text: str, name: str = "answers.csv", encoding: str = "utf-8") -> ParserFile:
    return ParserFile(filename=name, content=text.encode(encoding))


@pytest.mark.unit
def test_can_handle_csv_only_for_answer_zones():
    p = get_parser("csv_strict")
    assert p.can_handle(b"", "x.csv", "student_answers") is True
    assert p.can_handle(b"", "x.csv", "teacher_answers") is True
    assert p.can_handle(b"", "x.csv", "questions") is False
    assert p.can_handle(b"", "x.pdf", "student_answers") is False


@pytest.mark.unit
async def test_student_answers_basic():
    csv_text = ",Q1,Q2,Q3,Q4,Q5\n正確解答,B,A,A,C,D\n梁祐邦,A,=,C,C,D\n田瑜婕,=,A,C,B,D\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "student_answers")
    students = result.data["students"]
    assert [s["name"] for s in students] == ["梁祐邦", "田瑜婕"]
    assert [a["answer"] for a in students[0]["answers"]] == ["A", "=", "C", "C", "D"]
    assert [a["answer"] for a in students[1]["answers"]] == ["=", "A", "C", "B", "D"]
    assert all(a["question_number"] == i + 1 for i, a in enumerate(students[0]["answers"]))


@pytest.mark.unit
async def test_student_answers_blank_cell_becomes_none():
    csv_text = ",Q1,Q2,Q3\n正確解答,B,A,A\n梁祐邦,A,,C\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "student_answers")
    answers = [a["answer"] for a in result.data["students"][0]["answers"]]
    assert answers == ["A", None, "C"]


@pytest.mark.unit
async def test_student_answers_legacy_dash_becomes_equals():
    csv_text = ",Q1,Q2,Q3\n正確解答,B,A,A\n梁祐邦,-,A,A\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "student_answers")
    assert result.data["students"][0]["answers"][0]["answer"] == "="


@pytest.mark.unit
async def test_student_answers_skips_empty_name_rows():
    csv_text = ",Q1,Q2\n正確解答,B,A\n梁祐邦,A,A\n,B,B\n田瑜婕,B,A\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "student_answers")
    assert [s["name"] for s in result.data["students"]] == ["梁祐邦", "田瑜婕"]
    assert any("Skipped" in n for n in result.notes)


@pytest.mark.unit
async def test_student_answers_warns_when_key_row_isnt_marker():
    csv_text = ",Q1,Q2\n梁祐邦,A,B\n田瑜婕,B,A\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "student_answers")
    # Row 1 (after header) treated as the key even when name isn't a marker
    assert len(result.data["students"]) == 1
    assert result.data["students"][0]["name"] == "田瑜婕"
    assert any("treating it as the key" in n for n in result.notes)


@pytest.mark.unit
async def test_student_answers_rejects_too_few_rows():
    csv_text = ",Q1,Q2,Q3\n正確解答,A,B,C\n"
    p = get_parser("csv_strict")
    with pytest.raises(ValueError, match="at least one student"):
        await p.parse([_file(csv_text)], "student_answers")


@pytest.mark.unit
async def test_student_answers_rejects_single_column():
    csv_text = "header\n正確解答\n梁祐邦\n"
    p = get_parser("csv_strict")
    with pytest.raises(ValueError, match="at least 2 columns"):
        await p.parse([_file(csv_text)], "student_answers")


@pytest.mark.unit
async def test_student_answers_pads_short_rows():
    csv_text = ",Q1,Q2,Q3,Q4\n正確解答,B,A,A,C\n梁祐邦,A,=\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "student_answers")
    answers = [a["answer"] for a in result.data["students"][0]["answers"]]
    assert answers == ["A", "=", None, None]


@pytest.mark.unit
async def test_student_answers_arbitrary_header_text_ignored():
    """Row 0 header text is not validated — only column count matters."""
    csv_text = "name,foo,bar,baz\n正確解答,B,A,A\n梁祐邦,A,=,C\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "student_answers")
    assert len(result.data["students"]) == 1


@pytest.mark.unit
async def test_teacher_answers_basic():
    csv_text = ",Q1,Q2,Q3,Q4,Q5\n正確解答,B,A,A,C,D\n梁祐邦,A,=,C,C,D\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "teacher_answers")
    answers = result.data["answers"]
    assert [a["correct_answer"] for a in answers] == ["B", "A", "A", "C", "D"]
    assert [a["question_number"] for a in answers] == [1, 2, 3, 4, 5]


@pytest.mark.unit
async def test_teacher_answers_rejects_equals_in_key():
    csv_text = ",Q1,Q2,Q3\n正確解答,B,=,A\n梁祐邦,A,A,A\n"
    p = get_parser("csv_strict")
    with pytest.raises(ValueError, match="must be a concrete letter"):
        await p.parse([_file(csv_text)], "teacher_answers")


@pytest.mark.unit
async def test_teacher_answers_rejects_blank_cell_in_key():
    csv_text = ",Q1,Q2,Q3\n正確解答,B,,A\n梁祐邦,A,A,A\n"
    p = get_parser("csv_strict")
    with pytest.raises(ValueError, match="empty or not a letter"):
        await p.parse([_file(csv_text)], "teacher_answers")


@pytest.mark.unit
async def test_utf8_bom_accepted():
    csv_text = ",Q1,Q2\n正確解答,B,A\n梁祐邦,A,A\n"
    file = ParserFile(filename="answers.csv", content=("" + csv_text).encode("utf-8"))
    p = get_parser("csv_strict")
    result = await p.parse([file], "student_answers")
    assert result.data["students"][0]["name"] == "梁祐邦"


@pytest.mark.unit
async def test_non_utf8_raises_clear_error():
    csv_text = ",Q1,Q2\n正確解答,B,A\n梁祐邦,A,A\n"
    file = ParserFile(filename="answers.csv", content=csv_text.encode("big5"))
    p = get_parser("csv_strict")
    with pytest.raises(ValueError, match="UTF-8"):
        await p.parse([file], "student_answers")


@pytest.mark.unit
def test_pick_auto_routes_csv_to_strict():
    chosen = pick_auto(b"", "answers.csv", "student_answers")
    assert chosen.name == "csv_strict"
    chosen = pick_auto(b"", "answers.csv", "teacher_answers")
    assert chosen.name == "csv_strict"


@pytest.mark.unit
def test_pick_auto_questions_never_uses_csv():
    chosen = pick_auto(b"", "answers.csv", "questions")
    assert chosen.name == "llm_vision"


@pytest.mark.unit
async def test_lowercase_marker_accepted():
    csv_text = ",Q1,Q2\nkey,B,A\n梁祐邦,A,A\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "student_answers")
    # Should NOT warn — "key" is in the marker set
    assert not any("treating it as the key" in n for n in result.notes)
    assert len(result.data["students"]) == 1


@pytest.mark.unit
async def test_excel_trailing_empty_columns_ignored():
    csv_text = ",Q1,Q2,Q3,,\n正確解答,B,A,A,,\n梁祐邦,A,=,C,,\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "student_answers")
    answers = [a["answer"] for a in result.data["students"][0]["answers"]]
    assert answers == ["A", "=", "C"]


# ---------------------------------------------------------------------------
# Auto-fill correct marker when CSV has no "=" anywhere
# ---------------------------------------------------------------------------


@pytest.mark.unit
async def test_auto_marks_correct_when_no_equals_found():
    """If no '=' exists, cells matching the key get replaced with '='."""
    csv_text = ",Q1,Q2,Q3\n正確解答,B,A,C\n梁祐邦,B,X,C\n田瑜婕,A,A,Y\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "student_answers")
    students = result.data["students"]
    # 梁祐邦: Q1 matches key B → "=", Q2 wrong, Q3 matches C → "="
    assert [a["answer"] for a in students[0]["answers"]] == ["=", "X", "="]
    # 田瑜婕: Q1 wrong, Q2 matches A → "=", Q3 wrong
    assert [a["answer"] for a in students[1]["answers"]] == ["A", "=", "Y"]


@pytest.mark.unit
async def test_does_not_auto_mark_when_equals_already_present():
    """If even one '=' exists, leave everything else alone."""
    csv_text = ",Q1,Q2,Q3\n正確解答,B,A,C\n梁祐邦,B,X,=\n田瑜婕,A,A,Y\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "student_answers")
    students = result.data["students"]
    # 梁祐邦's "B" stays "B" — no auto-conversion since "=" was found in his row
    assert [a["answer"] for a in students[0]["answers"]] == ["B", "X", "="]
    # 田瑜婕's "A" stays "A" — same file, same skip
    assert [a["answer"] for a in students[1]["answers"]] == ["A", "A", "Y"]


@pytest.mark.unit
async def test_auto_mark_handles_missing_key_letter():
    """Cells where the key is None stay as the student typed."""
    # Key Q2 will be blank → triggers teacher-side validation if used as
    # teacher_answers, so we test student path with a key letter present
    # alongside one None. Here we craft via a key cell with letter+None mix.
    csv_text = ",Q1,Q2,Q3\n正確解答,B,A,C\n梁祐邦,B,A,X\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "student_answers")
    students = result.data["students"]
    # All match except Q3 → first two become "=", Q3 stays "X"
    assert [a["answer"] for a in students[0]["answers"]] == ["=", "=", "X"]


@pytest.mark.unit
async def test_auto_mark_handles_blank_student_cell():
    """Blank student cells stay None — never auto-converted to '='."""
    csv_text = ",Q1,Q2,Q3\n正確解答,B,A,C\n梁祐邦,B,,C\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "student_answers")
    answers = [a["answer"] for a in result.data["students"][0]["answers"]]
    assert answers == ["=", None, "="]


@pytest.mark.unit
async def test_auto_mark_note_emitted_with_count():
    csv_text = ",Q1,Q2,Q3\n正確解答,B,A,C\n梁祐邦,B,X,C\n"
    p = get_parser("csv_strict")
    result = await p.parse([_file(csv_text)], "student_answers")
    matched = [n for n in result.notes if "auto-converted" in n]
    assert len(matched) == 1
    assert "2 correct" in matched[0]


@pytest.mark.unit
async def test_auto_mark_is_per_file_not_global():
    """File A has '=' → no preprocess; File B has no '=' → preprocessed."""
    csv_a = ",Q1,Q2\n正確解答,B,A\n甲生,=,X\n"
    csv_b = ",Q1,Q2\n正確解答,B,A\n乙生,B,A\n"
    p = get_parser("csv_strict")
    result = await p.parse(
        [_file(csv_a, name="a.csv"), _file(csv_b, name="b.csv")],
        "student_answers",
    )
    students = result.data["students"]
    by_name = {s["name"]: s for s in students}
    # 甲生 from file A — no preprocessing, "=" preserved, "X" untouched
    assert [a["answer"] for a in by_name["甲生"]["answers"]] == ["=", "X"]
    # 乙生 from file B — preprocessing replaces both with "="
    assert [a["answer"] for a in by_name["乙生"]["answers"]] == ["=", "="]
    # Note mentions only file B
    matched = [n for n in result.notes if "auto-converted" in n]
    assert len(matched) == 1
    assert "b.csv" in matched[0]


# ---------------------------------------------------------------------------
# Combined "answers" data_type via process_uploaded_files
# ---------------------------------------------------------------------------


class _FakeUpload:
    """Minimal stand-in for FastAPI's UploadFile."""

    def __init__(self, filename: str, content: bytes) -> None:
        self.filename = filename
        self._content = content

    async def read(self) -> bytes:
        return self._content


@pytest.mark.unit
async def test_combined_answers_persists_both_subtypes(tmp_path, monkeypatch):
    """`data_type='answers'` runs the parser twice and stores both rows."""
    from app import database
    from app.file_processor import process_uploaded_files

    # Point the DB at a fresh local SQLite file and rebuild the engine.
    from app import config as config_module
    from app import db as db_module

    db_file = tmp_path / "test.db"
    cached = config_module.get_settings()
    monkeypatch.setattr(cached, "database_provider", "sqlite", raising=False)
    monkeypatch.setattr(cached, "supabase_db_url", "", raising=False)
    monkeypatch.setattr(cached, "database_path", str(db_file), raising=False)
    db_module.get_engine.cache_clear()
    db_module.get_sessionmaker.cache_clear()

    await database.init_database()
    user_id = await database.create_user("t@example.com", "h", "T")
    session_id = await database.create_session(user_id, "S")

    csv_text = ",Q1,Q2,Q3,Q4,Q5\n正確解答,B,A,A,C,D\n梁祐邦,A,=,C,C,D\n田瑜婕,=,A,C,B,D\n"
    upload = _FakeUpload("answers.csv", csv_text.encode("utf-8"))

    response = await process_uploaded_files(
        [upload],
        "answers",
        session_id,
        parser="csv_strict",
    )

    assert "student_answers" in response
    assert "teacher_answers" in response
    assert len(response["student_answers"]["students"]) == 2
    assert len(response["teacher_answers"]["answers"]) == 5

    rows = await database.get_parsed_data(session_id)
    by_type = {r["data_type"]: r for r in rows}
    assert "student_answers" in by_type
    assert "teacher_answers" in by_type
    assert len(by_type["student_answers"]["structured_data"]["students"]) == 2
    assert len(by_type["teacher_answers"]["structured_data"]["answers"]) == 5