wordloop/backend/tests/test_adaptive_review_service.py

import unittest
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace

from services.adaptive_review_service import (
    adaptive_review_service,
    response_quality,
    retrievability,
    utc_iso,
)


def make_word(
    word_id: int,
    *,
    stability_hours: float = 24.0,
    difficulty: float = 5.0,
    elapsed_hours: float = 24.0,
    next_review_offset_hours: float = 0.0,
    status: str = "learning",
    wrong_count: int = 0,
    correct_count: int = 0,
    mastery_score: int = 50,
    train_count: int = 0,
    error_bank_member: int = 0,
    error_reinforce_streak: int = 0,
):
    now = datetime(2026, 6, 6, 12, tzinfo=timezone.utc)
    return SimpleNamespace(
        id=word_id,
        stability_hours=stability_hours,
        difficulty=difficulty,
        last_reviewed_at=utc_iso(now - timedelta(hours=elapsed_hours)),
        created_at=utc_iso(now - timedelta(days=10)),
        next_review_at=utc_iso(now + timedelta(hours=next_review_offset_hours)),
        review_due_date=now.strftime("%Y-%m-%d"),
        status=status,
        wrong_count=wrong_count,
        correct_count=correct_count,
        mastery_score=mastery_score,
        train_count=train_count,
        error_bank_member=error_bank_member,
        error_reinforce_streak=error_reinforce_streak,
    )


class AdaptiveReviewServiceTest(unittest.TestCase):
    def setUp(self):
        self.now = datetime(2026, 6, 6, 12, tzinfo=timezone.utc)

    def test_fast_unassisted_recall_extends_interval(self):
        word = make_word(1)
        update = adaptive_review_service.update_word(
            word,
            is_correct=True,
            attempts=1,
            hints_used=0,
            duration_seconds=6,
            now=self.now,
        )
        self.assertEqual(update.quality, 1.0)
        self.assertGreater(update.stability_hours, 70)
        self.assertGreater(update.interval_hours, 10)
        self.assertLess(update.difficulty, 5.0)

    def test_hints_and_slow_response_reduce_growth(self):
        fast_word = make_word(1)
        helped_word = make_word(2)
        fast = adaptive_review_service.update_word(
            fast_word,
            is_correct=True,
            attempts=1,
            hints_used=0,
            duration_seconds=6,
            now=self.now,
        )
        helped = adaptive_review_service.update_word(
            helped_word,
            is_correct=True,
            attempts=3,
            hints_used=2,
            duration_seconds=75,
            now=self.now,
        )
        self.assertLess(
            response_quality(
                is_correct=True,
                attempts=3,
                hints_used=2,
                duration_seconds=75,
            ),
            fast.quality,
        )
        self.assertLess(helped.stability_hours, fast.stability_hours)
        self.assertLess(helped.interval_hours, fast.interval_hours)

    def test_failed_recall_returns_soon(self):
        word = make_word(1, stability_hours=72)
        update = adaptive_review_service.update_word(
            word,
            is_correct=False,
            attempts=4,
            hints_used=4,
            duration_seconds=30,
            now=self.now,
        )
        self.assertEqual(update.interval_hours, 0.25)
        self.assertGreater(update.difficulty, 5.0)
        self.assertLess(update.stability_hours, 72)

    def test_overdue_weak_word_is_selected_before_future_mastered_word(self):
        weak = make_word(
            1,
            difficulty=8,
            elapsed_hours=72,
            next_review_offset_hours=-24,
            status="weak",
        )
        mastered = make_word(
            2,
            stability_hours=240,
            difficulty=2,
            elapsed_hours=2,
            next_review_offset_hours=72,
            status="mastered",
        )
        selected = adaptive_review_service.select_words(
            [mastered, weak], limit=2, now=self.now
        )
        self.assertEqual([word.id for word in selected], [1, 2])
        self.assertLess(retrievability(weak, self.now), retrievability(mastered, self.now))

    def test_zero_correct_wrong_word_outranks_recent_mastered(self):
        never_correct = make_word(
            1,
            status="weak",
            wrong_count=3,
            correct_count=0,
            mastery_score=0,
            next_review_offset_hours=-1,
        )
        recent_ok = make_word(
            2,
            status="mastered",
            wrong_count=0,
            correct_count=2,
            mastery_score=100,
            elapsed_hours=1,
            stability_hours=240,
            next_review_offset_hours=48,
        )
        selected = adaptive_review_service.select_words(
            [recent_ok, never_correct], limit=2, now=self.now
        )
        self.assertEqual([word.id for word in selected], [1, 2])

    def test_is_error_priority_covers_weak_and_wrong_heavy(self):
        self.assertTrue(
            adaptive_review_service.is_error_priority(
                make_word(1, status="weak", wrong_count=1, correct_count=0)
            )
        )
        self.assertTrue(
            adaptive_review_service.is_error_priority(
                make_word(2, status="mastered", wrong_count=2, correct_count=1)
            )
        )
        self.assertFalse(
            adaptive_review_service.is_error_priority(
                make_word(
                    3,
                    status="learning",
                    wrong_count=0,
                    correct_count=2,
                    mastery_score=100,
                    train_count=2,
                )
            )
        )

    def test_plan_completed_excludes_clean_success_from_practice_plan(self):
        completed = make_word(
            1,
            status="learning",
            wrong_count=0,
            correct_count=2,
            mastery_score=100,
            train_count=2,
        )
        pending = make_word(
            2, status="new", wrong_count=0, correct_count=0, mastery_score=0, train_count=0
        )
        weak = make_word(
            3,
            status="weak",
            wrong_count=2,
            correct_count=0,
            mastery_score=0,
            train_count=2,
            error_bank_member=1,
        )

        self.assertTrue(adaptive_review_service.is_plan_completed(completed))
        self.assertFalse(adaptive_review_service.is_plan_eligible(completed))
        self.assertTrue(adaptive_review_service.is_plan_eligible(pending))
        self.assertTrue(adaptive_review_service.is_plan_eligible(weak))

        selected = adaptive_review_service.select_practice_plan_words(
            [completed, pending, weak], limit=3, now=self.now
        )
        self.assertEqual([word.id for word in selected], [2, 3])

    def test_untrained_excludes_error_bank_members_without_train_count(self):
        """错题库成员不应出现在未练词池，即使 train_count 仍为 0。"""
        pending = make_word(
            1, status="new", wrong_count=0, correct_count=0, mastery_score=0, train_count=0
        )
        migrated_error = make_word(
            2,
            status="weak",
            wrong_count=1,
            correct_count=0,
            mastery_score=0,
            train_count=0,
            error_bank_member=1,
            error_reinforce_streak=0,
        )

        untrained, reinforce = adaptive_review_service.split_plan_pools(
            [pending, migrated_error]
        )
        self.assertEqual([w.id for w in untrained], [1])
        self.assertEqual([w.id for w in reinforce], [2])

        selected = adaptive_review_service.select_untrained_words(
            [pending, migrated_error], limit=5, now=self.now
        )
        self.assertEqual([word.id for word in selected], [1])

    def test_select_untrained_words_only_returns_new_words(self):
        completed = make_word(
            1,
            status="learning",
            wrong_count=0,
            correct_count=2,
            mastery_score=100,
            train_count=2,
        )
        pending = make_word(
            2, status="new", wrong_count=0, correct_count=0, mastery_score=0, train_count=0
        )
        weak = make_word(
            3,
            status="weak",
            wrong_count=2,
            correct_count=0,
            mastery_score=0,
            train_count=2,
            error_bank_member=1,
        )

        selected = adaptive_review_service.select_untrained_words(
            [completed, pending, weak], limit=5, now=self.now
        )
        self.assertEqual([word.id for word in selected], [2])

    def test_select_error_words_only_returns_trained_plan_words(self):
        completed = make_word(
            1,
            status="learning",
            wrong_count=0,
            correct_count=2,
            mastery_score=100,
            train_count=2,
        )
        pending = make_word(
            2, status="new", wrong_count=0, correct_count=0, mastery_score=0, train_count=0
        )
        weak = make_word(
            3,
            status="weak",
            wrong_count=2,
            correct_count=0,
            mastery_score=0,
            train_count=2,
            error_bank_member=1,
            error_reinforce_streak=0,
        )

        selected = adaptive_review_service.select_error_words(
            [completed, pending, weak], limit=5, now=self.now
        )
        self.assertEqual([word.id for word in selected], [3])

    def test_count_plan_pools(self):
        completed = make_word(
            1,
            status="learning",
            wrong_count=0,
            correct_count=2,
            mastery_score=100,
            train_count=2,
        )
        pending = make_word(
            2, status="new", wrong_count=0, correct_count=0, mastery_score=0, train_count=0
        )
        weak = make_word(
            3,
            status="weak",
            wrong_count=2,
            correct_count=0,
            mastery_score=0,
            train_count=2,
            error_bank_member=1,
            error_reinforce_streak=0,
        )
        untrained, errors, bank_due, bank_total = adaptive_review_service.count_plan_pools(
            [completed, pending, weak]
        )
        self.assertEqual(untrained, 1)
        self.assertEqual(errors, 1)
        self.assertEqual(bank_total, 1)


    def test_is_review_due_false_without_schedule(self):
        word = make_word(1, error_bank_member=1, error_reinforce_streak=2)
        word.next_review_at = None
        word.review_due_date = None
        self.assertFalse(adaptive_review_service.is_review_due(word, self.now))

    def test_cooling_error_bank_word_not_selectable_for_review(self):
        cooling = make_word(
            1,
            wrong_count=2,
            train_count=3,
            error_bank_member=1,
            error_reinforce_streak=2,
            next_review_offset_hours=48,
        )
        selected = adaptive_review_service.select_error_bank_words(
            [cooling], limit=5, now=self.now, clear_count=2
        )
        self.assertEqual(selected, [])


if __name__ == "__main__":
    unittest.main()