Ship native iOS app, Wiki/TTS backend, and skeleton loading UX.
Replace the WebView shell with SwiftUI screens, add account-scoped Wiki and TTS APIs with adaptive review and photo scan support, and keep web/iOS pages usable while data loads asynchronously. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
john
@@ -0,0 +1,347 @@
|
||||
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()
|
||||
@@ -0,0 +1,78 @@
|
||||
import json
|
||||
import os
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from services.llm_service import LlmServiceError, llm_service
|
||||
|
||||
|
||||
class _FakeResponse:
|
||||
def __init__(self, payload: dict):
|
||||
self._payload = json.dumps(payload).encode("utf-8")
|
||||
|
||||
def read(self):
|
||||
return self._payload
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
return False
|
||||
|
||||
|
||||
class LlmServiceTest(unittest.TestCase):
|
||||
def tearDown(self):
|
||||
get_settings = __import__("config", fromlist=["get_settings"]).get_settings
|
||||
get_settings.cache_clear()
|
||||
|
||||
def test_not_configured_without_api_key(self):
|
||||
with patch.dict(os.environ, {"DEEPSEEK_API_KEY": ""}, clear=False):
|
||||
__import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
|
||||
self.assertFalse(llm_service.is_configured())
|
||||
|
||||
@patch("services.llm_service.urllib.request.urlopen")
|
||||
def test_enhance_wiki_page_returns_model_content(self, urlopen_mock: MagicMock):
|
||||
with patch.dict(os.environ, {"DEEPSEEK_API_KEY": "test-key"}, clear=False):
|
||||
__import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
|
||||
urlopen_mock.return_value = _FakeResponse(
|
||||
{
|
||||
"choices": [
|
||||
{"message": {"content": "# demo\n\n整理后的知识页。"}},
|
||||
]
|
||||
}
|
||||
)
|
||||
content = llm_service.enhance_wiki_page(title="demo", draft_markdown="# draft")
|
||||
self.assertIn("整理后的知识页", content)
|
||||
|
||||
def test_missing_api_key_raises(self):
|
||||
with patch.dict(os.environ, {"DEEPSEEK_API_KEY": ""}, clear=False):
|
||||
__import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
|
||||
with self.assertRaises(LlmServiceError):
|
||||
llm_service.enhance_wiki_page(title="demo", draft_markdown="# draft")
|
||||
|
||||
@patch("services.llm_service.urllib.request.urlopen")
|
||||
def test_parse_word_pairs_from_text(self, urlopen_mock: MagicMock):
|
||||
with patch.dict(os.environ, {"DEEPSEEK_API_KEY": "test-key"}, clear=False):
|
||||
__import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
|
||||
urlopen_mock.return_value = _FakeResponse(
|
||||
{
|
||||
"choices": [
|
||||
{
|
||||
"message": {
|
||||
"content": (
|
||||
'{"items":[{"en":"apple","zh":"苹果"},'
|
||||
'{"en":"banana","zh":"香蕉"}]}'
|
||||
)
|
||||
}
|
||||
},
|
||||
]
|
||||
}
|
||||
)
|
||||
pairs = llm_service.parse_word_pairs_from_text("apple 苹果\nbanana 香蕉")
|
||||
self.assertEqual(len(pairs), 2)
|
||||
self.assertEqual(pairs[0]["en"], "apple")
|
||||
self.assertEqual(pairs[0]["zh"], "苹果")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,11 @@
|
||||
from services.text_utils import en_answers_match, normalize_en_answer
|
||||
|
||||
|
||||
def test_normalize_en_answer_collapses_whitespace():
|
||||
assert normalize_en_answer(" A Lot Of ") == "a lot of"
|
||||
|
||||
|
||||
def test_en_answers_match_phrase_with_extra_spaces():
|
||||
assert en_answers_match("a lot of", "a lot of")
|
||||
assert en_answers_match("A Lot Of", "a lot of")
|
||||
assert not en_answers_match("a lot", "a lot of")
|
||||
@@ -0,0 +1,115 @@
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from services.tts_service import TtsService, TtsServiceError, tts_service
|
||||
|
||||
|
||||
class _FakeResponse:
|
||||
def __init__(self, body: bytes, content_type: str = "application/json"):
|
||||
self._body = body
|
||||
self.headers = MagicMock()
|
||||
self.headers.get_content_type.return_value = content_type
|
||||
|
||||
def read(self):
|
||||
return self._body
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
return False
|
||||
|
||||
|
||||
class TtsServiceTest(unittest.TestCase):
|
||||
def tearDown(self):
|
||||
get_settings = __import__("config", fromlist=["get_settings"]).get_settings
|
||||
get_settings.cache_clear()
|
||||
|
||||
def test_not_configured_when_disabled(self):
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"TKMIND_TTS_ENABLED": "false", "TKMIND_TTS_BASE_URL": "http://127.0.0.1:19200"},
|
||||
clear=False,
|
||||
):
|
||||
__import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
|
||||
self.assertFalse(tts_service.is_configured())
|
||||
|
||||
@patch("services.tts_service._open_tts_request")
|
||||
def test_request_includes_user_agent(self, urlopen_mock: MagicMock):
|
||||
wav = b"RIFFxxxx"
|
||||
payload = {"code": 0, "data": base64.b64encode(wav).decode("ascii")}
|
||||
urlopen_mock.return_value = _FakeResponse(json.dumps(payload).encode("utf-8"))
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"TKMIND_TTS_BASE_URL": "https://asr.tkmind.cn", "TKMIND_TTS_ENABLED": "true"},
|
||||
clear=False,
|
||||
):
|
||||
__import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
with patch("services.tts_service._CACHE_DIR", __import__("pathlib").Path(tmp)):
|
||||
tts_service.synthesize("apple")
|
||||
request = urlopen_mock.call_args.args[0]
|
||||
self.assertEqual(request.get_header("User-agent"), "WordLoop-TTS/1.0")
|
||||
|
||||
@patch("services.tts_service._open_tts_request")
|
||||
def test_synthesize_parses_json_base64(self, urlopen_mock: MagicMock):
|
||||
wav = b"RIFFxxxx"
|
||||
payload = {"code": 0, "message": "ok", "data": base64.b64encode(wav).decode("ascii")}
|
||||
urlopen_mock.return_value = _FakeResponse(json.dumps(payload).encode("utf-8"))
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"TKMIND_TTS_BASE_URL": "http://127.0.0.1:19200", "TKMIND_TTS_ENABLED": "true"},
|
||||
clear=False,
|
||||
):
|
||||
__import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
with patch("services.tts_service._CACHE_DIR", __import__("pathlib").Path(tmp)):
|
||||
audio, content_type = tts_service.synthesize("apple")
|
||||
self.assertEqual(audio, wav)
|
||||
self.assertEqual(content_type, "audio/wav")
|
||||
|
||||
@patch("services.tts_service._open_tts_request")
|
||||
def test_synthesize_uses_cache_on_second_call(self, urlopen_mock: MagicMock):
|
||||
wav = b"RIFFcached"
|
||||
payload = {"code": 0, "data": base64.b64encode(wav).decode("ascii")}
|
||||
urlopen_mock.return_value = _FakeResponse(json.dumps(payload).encode("utf-8"))
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"TKMIND_TTS_BASE_URL": "http://127.0.0.1:19200", "TKMIND_TTS_ENABLED": "true"},
|
||||
clear=False,
|
||||
):
|
||||
__import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
with patch("services.tts_service._CACHE_DIR", __import__("pathlib").Path(tmp)):
|
||||
first, _ = tts_service.synthesize("apple")
|
||||
second, _ = tts_service.synthesize("apple")
|
||||
self.assertEqual(first, second)
|
||||
self.assertEqual(urlopen_mock.call_count, 1)
|
||||
|
||||
@patch("services.tts_service._open_tts_request")
|
||||
def test_upstream_error_message(self, urlopen_mock: MagicMock):
|
||||
payload = {"code": 500, "message": "TTS 未启用（ENABLE_TTS=false）", "data": None}
|
||||
urlopen_mock.return_value = _FakeResponse(json.dumps(payload).encode("utf-8"))
|
||||
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{"TKMIND_TTS_BASE_URL": "http://127.0.0.1:19200", "TKMIND_TTS_ENABLED": "true"},
|
||||
clear=False,
|
||||
):
|
||||
__import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
with patch("services.tts_service._CACHE_DIR", __import__("pathlib").Path(tmp)):
|
||||
with self.assertRaises(TtsServiceError) as ctx:
|
||||
tts_service.synthesize("apple")
|
||||
self.assertIn("TTS 未启用", str(ctx.exception))
|
||||
|
||||
def test_extract_audio_base64_from_dict(self):
|
||||
value = TtsService._extract_audio_base64({"audio": "abc"})
|
||||
self.assertEqual(value, "abc")
|
||||
@@ -0,0 +1,60 @@
|
||||
import os
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from database import Base
|
||||
from models import User, WikiLlmUsage
|
||||
from services.wiki_llm_quota_service import WikiLlmQuotaExceeded, wiki_llm_quota_service
|
||||
|
||||
|
||||
class WikiLlmQuotaServiceTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
Base.metadata.create_all(engine)
|
||||
self.db = sessionmaker(bind=engine)()
|
||||
self.user = User(
|
||||
username="alice",
|
||||
password_hash="test",
|
||||
created_at="2026-06-06T00:00:00Z",
|
||||
)
|
||||
self.db.add(self.user)
|
||||
self.db.commit()
|
||||
|
||||
def tearDown(self):
|
||||
self.db.close()
|
||||
get_settings = __import__("config", fromlist=["get_settings"]).get_settings
|
||||
get_settings.cache_clear()
|
||||
|
||||
def test_consume_until_limit(self):
|
||||
with patch.dict(os.environ, {"AI_WIKI_DAILY_LLM_LIMIT": "2"}, clear=False):
|
||||
__import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
|
||||
for _ in range(2):
|
||||
wiki_llm_quota_service.consume(self.db, self.user)
|
||||
self.db.commit()
|
||||
with self.assertRaises(WikiLlmQuotaExceeded):
|
||||
wiki_llm_quota_service.consume(self.db, self.user)
|
||||
|
||||
def test_quota_dict_tracks_usage(self):
|
||||
with patch.dict(os.environ, {"AI_WIKI_DAILY_LLM_LIMIT": "20"}, clear=False):
|
||||
__import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
|
||||
wiki_llm_quota_service.consume(self.db, self.user)
|
||||
self.db.commit()
|
||||
quota = wiki_llm_quota_service.quota_dict(self.db, self.user)
|
||||
self.assertEqual(quota["used_today"], 1)
|
||||
self.assertEqual(quota["remaining_today"], 19)
|
||||
self.assertFalse(quota["quota_exceeded"])
|
||||
|
||||
def test_usage_row_is_unique_per_day(self):
|
||||
row = wiki_llm_quota_service.get_usage_row(self.db, self.user, "2026-06-06")
|
||||
row.call_count = 3
|
||||
self.db.commit()
|
||||
rows = self.db.query(WikiLlmUsage).filter(WikiLlmUsage.user_id == self.user.id).all()
|
||||
self.assertEqual(len(rows), 1)
|
||||
self.assertEqual(rows[0].call_count, 3)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,138 @@
|
||||
import unittest
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from database import Base
|
||||
from models import QuizRecord, User, UserSettings, WikiPage, WikiSource, Word
|
||||
from services.wiki_service import wiki_service
|
||||
|
||||
|
||||
class WikiServiceTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
Base.metadata.create_all(engine)
|
||||
self.db = sessionmaker(bind=engine)()
|
||||
|
||||
def tearDown(self):
|
||||
self.db.close()
|
||||
|
||||
def create_user(self, username: str) -> User:
|
||||
user = User(
|
||||
username=username,
|
||||
password_hash="test",
|
||||
created_at="2026-06-06T00:00:00Z",
|
||||
)
|
||||
self.db.add(user)
|
||||
self.db.flush()
|
||||
self.db.add(
|
||||
UserSettings(
|
||||
user_id=user.id,
|
||||
ai_wiki_enabled=1,
|
||||
ai_wiki_auto_organize=1,
|
||||
)
|
||||
)
|
||||
self.db.commit()
|
||||
return user
|
||||
|
||||
def create_word(self, user: User, text: str) -> Word:
|
||||
word = Word(
|
||||
user_id=user.id,
|
||||
book_id=0,
|
||||
source_text=text,
|
||||
target_text="测试",
|
||||
source_lang="en",
|
||||
target_lang="zh",
|
||||
status="learning",
|
||||
correct_count=1,
|
||||
wrong_count=0,
|
||||
consecutive_correct_count=1,
|
||||
mastery_score=100,
|
||||
train_count=1,
|
||||
total_train_seconds=3,
|
||||
difficulty=5.0,
|
||||
stability_hours=24.0,
|
||||
created_at="2026-06-06T00:00:00Z",
|
||||
)
|
||||
self.db.add(word)
|
||||
self.db.commit()
|
||||
return word
|
||||
|
||||
def test_ingest_builds_private_markdown_page(self):
|
||||
user = self.create_user("alice")
|
||||
word = self.create_word(user, "resilient")
|
||||
wiki_service.ingest_word(self.db, user, word)
|
||||
|
||||
page = (
|
||||
self.db.query(WikiPage)
|
||||
.filter(WikiPage.user_id == user.id, WikiPage.category == "word")
|
||||
.one()
|
||||
)
|
||||
self.assertIn("# resilient", page.content)
|
||||
self.assertIn("WordLoop 原始学习事件 1 条", page.content)
|
||||
self.assertEqual(
|
||||
self.db.query(WikiSource).filter(WikiSource.user_id == user.id).count(),
|
||||
1,
|
||||
)
|
||||
|
||||
def test_quiz_ingest_updates_only_own_wiki(self):
|
||||
alice = self.create_user("alice")
|
||||
bob = self.create_user("bob")
|
||||
word = self.create_word(alice, "resilient")
|
||||
wiki_service.ingest_word(self.db, alice, word)
|
||||
record = QuizRecord(
|
||||
user_id=alice.id,
|
||||
word_id=word.id,
|
||||
question_type="spell",
|
||||
user_answer="resilient",
|
||||
correct_answer="resilient",
|
||||
is_correct=1,
|
||||
duration_seconds=2,
|
||||
created_at="2026-06-06T00:01:00Z",
|
||||
)
|
||||
self.db.add(record)
|
||||
self.db.commit()
|
||||
wiki_service.ingest_quiz(self.db, alice, word, record)
|
||||
|
||||
alice_status = wiki_service.status(self.db, alice)
|
||||
bob_status = wiki_service.status(self.db, bob)
|
||||
self.assertGreater(alice_status["page_count"], 0)
|
||||
self.assertEqual(alice_status["latest_page"]["source_count"], 2)
|
||||
self.assertEqual(bob_status["page_count"], 0)
|
||||
self.assertIsNone(bob_status["latest_page"])
|
||||
|
||||
def test_recompile_prunes_orphan_status_collection(self):
|
||||
user = self.create_user("alice")
|
||||
word = self.create_word(user, "resilient")
|
||||
word.status = "new"
|
||||
self.db.commit()
|
||||
wiki_service.ingest_word(self.db, user, word)
|
||||
|
||||
word.status = "learning"
|
||||
self.db.commit()
|
||||
wiki_service.rebuild(self.db, user)
|
||||
|
||||
slugs = {
|
||||
page.slug
|
||||
for page in self.db.query(WikiPage).filter(WikiPage.user_id == user.id).all()
|
||||
}
|
||||
self.assertNotIn("learning/new", slugs)
|
||||
self.assertIn("learning/learning", slugs)
|
||||
|
||||
def test_graph_returns_wiki_pages_and_links(self):
|
||||
user = self.create_user("alice")
|
||||
word = self.create_word(user, "resilient")
|
||||
wiki_service.ingest_word(self.db, user, word)
|
||||
|
||||
graph = wiki_service.graph(self.db, user)
|
||||
|
||||
node_slugs = {node["slug"] for node in graph["nodes"]}
|
||||
relations = {link["relation"] for link in graph["links"]}
|
||||
self.assertIn("index", node_slugs)
|
||||
self.assertIn(f"vocabulary/{word.id}-resilient", node_slugs)
|
||||
self.assertIn("learning_status", relations)
|
||||
self.assertIn("collection", relations)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,37 @@
|
||||
import os
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from fastapi import HTTPException
|
||||
|
||||
from services.word_scan_service import word_scan_service
|
||||
|
||||
|
||||
class WordScanServiceTest(unittest.TestCase):
|
||||
def tearDown(self):
|
||||
get_settings = __import__("config", fromlist=["get_settings"]).get_settings
|
||||
get_settings.cache_clear()
|
||||
|
||||
def test_heuristic_parse_tab_separated_pairs(self):
|
||||
with patch.dict(os.environ, {"DEEPSEEK_API_KEY": ""}, clear=False):
|
||||
__import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
|
||||
items = word_scan_service.parse_scan_text("apple\t苹果\nbanana\t香蕉")
|
||||
self.assertEqual(len(items), 2)
|
||||
self.assertEqual(items[0]["source_text"], "apple")
|
||||
self.assertEqual(items[0]["target_text"], "苹果")
|
||||
|
||||
def test_empty_text_raises(self):
|
||||
with self.assertRaises(HTTPException) as ctx:
|
||||
word_scan_service.parse_scan_text(" ")
|
||||
self.assertEqual(ctx.exception.status_code, 400)
|
||||
|
||||
def test_unparseable_text_raises(self):
|
||||
with patch.dict(os.environ, {"DEEPSEEK_API_KEY": ""}, clear=False):
|
||||
__import__("config", fromlist=["get_settings"]).get_settings.cache_clear()
|
||||
with self.assertRaises(HTTPException) as ctx:
|
||||
word_scan_service.parse_scan_text("hello world only english")
|
||||
self.assertEqual(ctx.exception.status_code, 422)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user