devplacepy/tests/unit/services/quiz/store.py

# retoor <retoor@molodetz.nl>

import json
from datetime import datetime, timedelta, timezone

import pytest

from devplacepy.database import get_table
from devplacepy.services.quiz import store
from devplacepy.services.quiz.store import QuizError
from devplacepy.utils import generate_uid, make_combined_slug
from tests.conftest import run_async

_counter = [0]


def _user(prefix):
    _counter[0] += 1
    uid = generate_uid()
    get_table("users").insert(
        {
            "uid": uid,
            "username": f"{prefix}{_counter[0]}",
            "terms_version": "1",
            "email": f"{prefix}{_counter[0]}@t.dev",
            "role": "Member",
            "api_key": generate_uid(),
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    return get_table("users").find_one(uid=uid)


def _quiz(owner, **overrides):
    uid = generate_uid()
    fields = {
        "uid": uid,
        "user_uid": owner["uid"],
        "slug": make_combined_slug("Quiz", uid),
        "title": "Quiz",
        "description": "d",
        "status": "draft",
        "published_at": "",
        "shuffle_questions": 0,
        "shuffle_options": 0,
        "reveal_answers": 1,
        "allow_review": 1,
        "time_limit_seconds": 0,
        "pass_percent": 0,
        "question_count": 0,
        "total_points": 0,
        "attempt_count": 0,
        "stars": 0,
    }
    fields.update(overrides)
    get_table("quizzes").insert(store.born_live(fields))
    return store.get_quiz(uid)


def _numeric(quiz_uid, value=1.0, points=1):
    return store.add_question(
        quiz_uid,
        {
            "kind": "numeric",
            "prompt": f"Value {value}",
            "points": points,
            "numeric_value": value,
            "numeric_tolerance": 0.0,
            "options": [],
        },
    )


def _choice(quiz_uid, points=2):
    return store.add_question(
        quiz_uid,
        {
            "kind": "single_choice",
            "prompt": "Pick one",
            "points": points,
            "options": [
                {"label": "wrong", "match_value": "", "is_correct": False},
                {"label": "right", "match_value": "", "is_correct": True},
            ],
        },
    )


def test_recompute_quiz_totals_reflects_the_live_questions(local_db):
    owner = _user("qs")
    quiz = _quiz(owner)
    _numeric(quiz["uid"], points=3)
    _choice(quiz["uid"], points=2)
    refreshed = store.get_quiz(quiz["uid"])
    assert refreshed["question_count"] == 2
    assert refreshed["total_points"] == 5


def test_deleting_a_question_recomputes_and_renumbers(local_db):
    owner = _user("qs")
    quiz = _quiz(owner)
    first = _numeric(quiz["uid"], points=3)
    _numeric(quiz["uid"], value=2.0, points=4)
    store.delete_question(quiz["uid"], first["uid"], owner["uid"])
    refreshed = store.get_quiz(quiz["uid"])
    assert refreshed["question_count"] == 1
    assert refreshed["total_points"] == 4
    assert store.list_questions(quiz["uid"])[0]["position"] == 0


def test_reorder_requires_every_question_exactly_once(local_db):
    owner = _user("qs")
    quiz = _quiz(owner)
    first = _numeric(quiz["uid"])
    _numeric(quiz["uid"], value=2.0)
    with pytest.raises(QuizError):
        store.reorder_questions(quiz["uid"], [first["uid"]])


def test_reorder_applies_the_new_order(local_db):
    owner = _user("qs")
    quiz = _quiz(owner)
    first = _numeric(quiz["uid"])
    second = _numeric(quiz["uid"], value=2.0)
    store.reorder_questions(quiz["uid"], [second["uid"], first["uid"]])
    assert [q["uid"] for q in store.list_questions(quiz["uid"])] == [second["uid"], first["uid"]]


def test_validation_errors_flags_an_empty_quiz(local_db):
    owner = _user("qs")
    quiz = _quiz(owner)
    assert store.validation_errors(quiz["uid"])


def test_validation_errors_is_empty_for_a_complete_quiz(local_db):
    owner = _user("qs")
    quiz = _quiz(owner)
    _choice(quiz["uid"])
    assert store.validation_errors(quiz["uid"]) == []


def test_validation_errors_flags_a_free_text_without_criteria(local_db):
    owner = _user("qs")
    quiz = _quiz(owner)
    store.add_question(
        quiz["uid"],
        {"kind": "free_text", "prompt": "Explain", "points": 1, "options": []},
    )
    assert any("reference answer" in problem for problem in store.validation_errors(quiz["uid"]))


def test_validation_errors_flags_a_matching_with_one_pair(local_db):
    owner = _user("qs")
    quiz = _quiz(owner)
    store.add_question(
        quiz["uid"],
        {
            "kind": "matching",
            "prompt": "Match",
            "points": 1,
            "options": [{"label": "a", "match_value": "b", "is_correct": False}],
        },
    )
    assert any("two pairs" in problem for problem in store.validation_errors(quiz["uid"]))


def test_publish_refuses_an_incomplete_quiz(local_db):
    owner = _user("qs")
    quiz = _quiz(owner)
    with pytest.raises(QuizError):
        store.publish_quiz(quiz["uid"])
    assert store.get_quiz(quiz["uid"])["status"] == "draft"


def test_publish_sets_the_status_once(local_db):
    owner = _user("qs")
    quiz = _quiz(owner)
    _choice(quiz["uid"])
    published = store.publish_quiz(quiz["uid"])
    assert published["status"] == "published"
    assert published["published_at"]
    assert published["publish_won"] is True


def test_publishing_twice_does_not_win_twice(local_db):
    owner = _user("qs")
    quiz = _quiz(owner)
    _choice(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    again = store.publish_quiz(quiz["uid"])
    assert again.get("publish_won") is not True


def test_guard_editable_refuses_a_published_quiz(local_db):
    owner = _user("qs")
    quiz = _quiz(owner)
    _choice(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    with pytest.raises(QuizError):
        store.guard_editable(quiz["uid"])


def test_every_mutating_entrypoint_is_frozen_after_publish(local_db):
    owner = _user("qs")
    quiz = _quiz(owner)
    question = _choice(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    with pytest.raises(QuizError):
        store.add_question(quiz["uid"], {"kind": "numeric", "prompt": "x", "options": []})
    with pytest.raises(QuizError):
        store.edit_question(quiz["uid"], question["uid"], {"kind": "numeric", "prompt": "x", "options": []})
    with pytest.raises(QuizError):
        store.delete_question(quiz["uid"], question["uid"], owner["uid"])
    with pytest.raises(QuizError):
        store.reorder_questions(quiz["uid"], [question["uid"]])
    with pytest.raises(QuizError):
        store.edit_quiz(quiz["uid"], {"title": "new"})


def test_a_published_quiz_keeps_its_question_bytes(local_db):
    owner = _user("qs")
    quiz = _quiz(owner)
    question = _choice(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    with pytest.raises(QuizError):
        store.edit_question(
            quiz["uid"], question["uid"], {"kind": "numeric", "prompt": "changed", "options": []}
        )
    assert store.list_questions(quiz["uid"])[0]["prompt"] == "Pick one"


def test_can_view_quiz_hides_a_draft_from_everyone_else(local_db):
    owner = _user("qs")
    other = _user("qs")
    quiz = _quiz(owner)
    assert store.can_view_quiz(quiz, owner) is True
    assert store.can_view_quiz(quiz, other) is False
    assert store.can_view_quiz(quiz, None) is False


def test_can_view_quiz_allows_everyone_once_published(local_db):
    owner = _user("qs")
    other = _user("qs")
    quiz = _quiz(owner)
    _choice(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    assert store.can_view_quiz(store.get_quiz(quiz["uid"]), other) is True
    assert store.can_view_quiz(store.get_quiz(quiz["uid"]), None) is True


def test_starting_twice_returns_the_same_attempt(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner)
    _choice(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    quiz = store.get_quiz(quiz["uid"])
    first = store.start_attempt(player, quiz)
    second = store.start_attempt(player, quiz)
    assert first["uid"] == second["uid"]
    live = get_table("quiz_attempts").count(
        user_uid=player["uid"], quiz_uid=quiz["uid"], status="in_progress", deleted_at=None
    )
    assert live == 1


def test_starting_materializes_a_blank_answer_per_question(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner)
    _choice(quiz["uid"])
    _numeric(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    attempt = store.start_attempt(player, store.get_quiz(quiz["uid"]))
    answers = store.answers_for(attempt["uid"])
    assert len(answers) == 2
    assert all(not answer["answered_at"] for answer in answers)


def test_starting_snapshots_the_max_points(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner)
    _choice(quiz["uid"], points=5)
    store.publish_quiz(quiz["uid"])
    attempt = store.start_attempt(player, store.get_quiz(quiz["uid"]))
    assert attempt["max_points"] == 5


def test_answering_credits_the_attempt(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner)
    question = _numeric(quiz["uid"], value=7.0, points=4)
    store.publish_quiz(quiz["uid"])
    quiz = store.get_quiz(quiz["uid"])
    attempt = store.start_attempt(player, quiz)
    answer, updated, result = run_async(
        store.answer(
            player, quiz, attempt, question["uid"], {"answer_text": "7", "option_uids": []}
        )
    )
    assert answer["is_correct"] == 1
    assert float(answer["awarded_points"]) == 4.0
    assert float(updated["score_points"]) == 4.0
    assert int(updated["answered_count"]) == 1
    assert result.graded_by == "auto"


def test_a_second_answer_is_refused_and_credits_nothing(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner)
    question = _numeric(quiz["uid"], value=7.0, points=4)
    store.publish_quiz(quiz["uid"])
    quiz = store.get_quiz(quiz["uid"])
    attempt = store.start_attempt(player, quiz)
    run_async(
        store.answer(
            player, quiz, attempt, question["uid"], {"answer_text": "7", "option_uids": []}
        )
    )
    with pytest.raises(QuizError):
        run_async(
            store.answer(
                player,
                quiz,
                store.get_attempt(attempt["uid"]),
                question["uid"],
                {"answer_text": "7", "option_uids": []},
            )
        )
    refreshed = store.get_attempt(attempt["uid"])
    assert float(refreshed["score_points"]) == 4.0
    assert int(refreshed["answered_count"]) == 1


def test_finishing_recomputes_the_score_from_the_answers(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner, pass_percent=50)
    first = _numeric(quiz["uid"], value=1.0, points=2)
    _numeric(quiz["uid"], value=2.0, points=2)
    store.publish_quiz(quiz["uid"])
    quiz = store.get_quiz(quiz["uid"])
    attempt = store.start_attempt(player, quiz)
    run_async(
        store.answer(
            player, quiz, attempt, first["uid"], {"answer_text": "1", "option_uids": []}
        )
    )
    finished, won = store.finish(quiz, store.get_attempt(attempt["uid"]))
    assert won is True
    assert finished["status"] == "completed"
    assert float(finished["score_points"]) == 2.0
    assert float(finished["score_percent"]) == 50.0
    assert int(finished["passed"]) == 1


def test_finishing_twice_wins_only_once(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner)
    _choice(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    quiz = store.get_quiz(quiz["uid"])
    attempt = store.start_attempt(player, quiz)
    store.finish(quiz, attempt)
    _, won = store.finish(quiz, store.get_attempt(attempt["uid"]))
    assert won is False
    assert int(store.get_quiz(quiz["uid"])["attempt_count"]) == 1


def test_an_expired_attempt_refuses_further_answers(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner, time_limit_seconds=60)
    question = _numeric(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    quiz = store.get_quiz(quiz["uid"])
    attempt = store.start_attempt(player, quiz)
    past = (datetime.now(timezone.utc) - timedelta(seconds=10)).isoformat()
    get_table("quiz_attempts").update({"uid": attempt["uid"], "expires_at": past}, ["uid"])
    with pytest.raises(QuizError):
        run_async(
            store.answer(
                player,
                quiz,
                store.get_attempt(attempt["uid"]),
                question["uid"],
                {"answer_text": "1", "option_uids": []},
            )
        )
    assert store.get_attempt(attempt["uid"])["status"] == "expired"


def test_expiry_happens_lazily_on_a_read(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner, time_limit_seconds=60)
    _numeric(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    quiz = store.get_quiz(quiz["uid"])
    attempt = store.start_attempt(player, quiz)
    past = (datetime.now(timezone.utc) - timedelta(seconds=10)).isoformat()
    get_table("quiz_attempts").update({"uid": attempt["uid"], "expires_at": past}, ["uid"])
    serialized = store.serialize_attempt(quiz, store.get_attempt(attempt["uid"]), player)
    assert serialized["status"] == "expired"


def test_the_serializer_withholds_the_answer_key_from_a_player(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner, reveal_answers=0)
    _choice(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    quiz = store.get_quiz(quiz["uid"])
    attempt = store.start_attempt(player, quiz)
    payload = store.serialize_attempt(quiz, attempt, player)
    question = payload["questions"][0]
    assert "correct_boolean" not in question
    assert "expected_answer" not in question
    assert all("is_correct" not in option for option in question["options"])


def test_the_serializer_gives_the_owner_the_answer_key(local_db):
    owner = _user("qs")
    quiz = _quiz(owner, reveal_answers=0)
    _choice(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    quiz = store.get_quiz(quiz["uid"])
    attempt = store.start_attempt(owner, quiz)
    payload = store.serialize_attempt(quiz, attempt, owner)
    question = payload["questions"][0]
    assert any("is_correct" in option for option in question["options"])


def test_the_serializer_reveals_an_answered_question_when_configured(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner, reveal_answers=1)
    question = _numeric(quiz["uid"], value=3.0)
    store.publish_quiz(quiz["uid"])
    quiz = store.get_quiz(quiz["uid"])
    attempt = store.start_attempt(player, quiz)
    run_async(
        store.answer(
            player, quiz, attempt, question["uid"], {"answer_text": "3", "option_uids": []}
        )
    )
    payload = store.serialize_attempt(quiz, store.get_attempt(attempt["uid"]), player)
    assert "numeric_value" in payload["questions"][0]


def test_matching_choices_are_exposed_without_the_pairing(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner, reveal_answers=0)
    store.add_question(
        quiz["uid"],
        {
            "kind": "matching",
            "prompt": "Match",
            "points": 1,
            "options": [
                {"label": "a", "match_value": "A", "is_correct": False},
                {"label": "b", "match_value": "B", "is_correct": False},
            ],
        },
    )
    store.publish_quiz(quiz["uid"])
    quiz = store.get_quiz(quiz["uid"])
    attempt = store.start_attempt(player, quiz)
    question = store.serialize_attempt(quiz, attempt, player)["questions"][0]
    assert sorted(question["match_choices"]) == ["A", "B"]
    assert all("match_value" not in option for option in question["options"])


def test_the_cascade_removes_every_child_under_one_stamp(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner)
    _choice(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    store.start_attempt(player, store.get_quiz(quiz["uid"]))
    stamp = datetime.now(timezone.utc).isoformat()
    store.cascade_questions(quiz["uid"], owner["uid"], stamp)
    for table in ("quiz_questions", "quiz_options", "quiz_attempts", "quiz_answers"):
        live = get_table(table).count(quiz_uid=quiz["uid"], deleted_at=None)
        stamped = get_table(table).count(quiz_uid=quiz["uid"], deleted_at=stamp)
        assert live == 0
        assert stamped > 0


def test_prune_attempts_keeps_completed_records(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner)
    _choice(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    quiz = store.get_quiz(quiz["uid"])
    kept = store.start_attempt(player, quiz)
    store.finish(quiz, kept)
    future = (datetime.now(timezone.utc) + timedelta(days=1)).isoformat()
    store.prune_attempts(future)
    assert store.get_attempt(kept["uid"]) is not None


def test_prune_attempts_removes_an_abandoned_one(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner)
    _choice(quiz["uid"])
    store.publish_quiz(quiz["uid"])
    attempt = store.start_attempt(player, store.get_quiz(quiz["uid"]))
    future = (datetime.now(timezone.utc) + timedelta(days=1)).isoformat()
    store.prune_attempts(future)
    assert store.get_attempt(attempt["uid"]) is None


def test_the_attempt_order_is_persisted_as_json(local_db):
    owner = _user("qs")
    player = _user("qs")
    quiz = _quiz(owner, shuffle_questions=1)
    _numeric(quiz["uid"], value=1.0)
    _numeric(quiz["uid"], value=2.0)
    _numeric(quiz["uid"], value=3.0)
    store.publish_quiz(quiz["uid"])
    attempt = store.start_attempt(player, store.get_quiz(quiz["uid"]))
    order = json.loads(attempt["question_order"])
    assert sorted(order) == sorted(store.question_uids(quiz["uid"]))
    assert store.attempt_order(store.get_attempt(attempt["uid"])) == order