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()