import os
import tempfile
import pytest
from pathlib import Path
from fastapi.testclient import TestClient
# Use temporary database for tests
temp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
os.environ["DB_PATH"] = temp_db.name
os.environ["ADMIN_USERNAME"] = "testadmin"
os.environ["ADMIN_PASSWORD"] = "testpass12345"
from app.main import app
from app import config
from app.database import init_db, get_db
from app.auth import hash_password, verify_password, create_user, authenticate_user, ensure_admin_user, change_user_password
config.ADMIN_USERNAME = "testadmin"
config.ADMIN_PASSWORD = "testpass12345"
client = TestClient(app)
def setup_module():
init_db()
ensure_admin_user()
def teardown_module():
try:
Path(temp_db.name).unlink(missing_ok=True)
except Exception:
pass
def test_password_hashing():
pwd = "superSecretPassword123"
h, salt = hash_password(pwd)
assert verify_password(pwd, h, salt) is True
assert verify_password("wrongPassword", h, salt) is False
def test_user_creation_and_auth():
uid = create_user("demo_user", "demo_pass_999")
assert uid > 0
auth = authenticate_user("demo_user", "demo_pass_999")
assert auth is not None
assert auth["username"] == "demo_user"
wrong = authenticate_user("demo_user", "wrong_pass")
assert wrong is None
def test_unauthenticated_access_blocked():
# Attempting to access protected API without session
resp = client.get("/api/me")
assert resp.status_code == 401
resp = client.get("/api/cards")
assert resp.status_code == 401
resp = client.post("/api/cards", json={"term": "apple", "translation": "яблоко"})
assert resp.status_code == 401
resp = client.get("/api/collections")
assert resp.status_code == 401
resp = client.post("/api/cards/batch", json={"card_ids": [1], "action": "delete"})
assert resp.status_code == 401
resp = client.get("/api/export/csv")
assert resp.status_code == 401
resp = client.get("/api/srs/queue")
assert resp.status_code == 401
resp = client.get("/api/srs/stats")
assert resp.status_code == 401
resp = client.post("/api/srs/review", json={"card_id": 1, "rating": 3})
assert resp.status_code == 401
def test_login_and_card_crud_workflow():
# 1. Login with testadmin
resp = client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "ok"
assert "user" in data
# 2. Check /api/me
resp = client.get("/api/me")
assert resp.status_code == 200
assert resp.json()["user"]["username"] == "testadmin"
# 3. Create a card
card_payload = {
"term": "reluctant",
"translation": "неохотный, нежелающий",
"context_sentence": "He was reluctant to leave the room.",
"phonetic": "/rɪˈlʌktənt/",
"part_of_speech": "Adjective",
"definition_en": "Not wanting to take some action",
"tags": "general, daily",
"notes": "Remember: reluctant to do sth"
}
resp = client.post("/api/cards", json=card_payload)
assert resp.status_code == 200
created = resp.json()
card_id = created["id"]
assert created["term"] == "reluctant"
# 4. Get cards list
resp = client.get("/api/cards")
assert resp.status_code == 200
res_data = resp.json()
assert res_data["total"] >= 1
assert any(c["id"] == card_id for c in res_data["cards"])
# 5. Filter cards by search
resp = client.get("/api/cards?search=reluctant")
assert resp.status_code == 200
assert resp.json()["total"] >= 1
# 6. Filter cards by tag
resp = client.get("/api/cards?tag=general")
assert resp.status_code == 200
assert resp.json()["total"] >= 1
# 7. Check tags list
resp = client.get("/api/tags")
assert resp.status_code == 200
tags = resp.json()
tag_names = [t["tag"] for t in tags]
assert "general" in tag_names
assert "daily" in tag_names
# 8. Update card
update_payload = {
"translation": "неохотный, сопротивляющийся",
"notes": "Updated note"
}
resp = client.put(f"/api/cards/{card_id}", json=update_payload)
assert resp.status_code == 200
updated = resp.json()
assert updated["translation"] == "неохотный, сопротивляющийся"
assert updated["notes"] == "Updated note"
# 9. Check stats
resp = client.get("/api/stats")
assert resp.status_code == 200
stats = resp.json()
assert stats["total_cards"] >= 1
assert stats["today_added"] >= 1
# 10. Delete card
resp = client.delete(f"/api/cards/{card_id}")
assert resp.status_code == 200
assert resp.json()["status"] == "ok"
# Verify deleted
resp = client.get("/api/cards")
assert not any(c["id"] == card_id for c in resp.json()["cards"])
# 11. Logout
resp = client.post("/api/logout")
assert resp.status_code == 200
# Verify session ended
resp = client.get("/api/me")
assert resp.status_code == 401
def test_login_rate_limiting():
# Attempt multiple failed logins from the same client
for _ in range(5):
resp = client.post("/api/login", json={"username": "testadmin", "password": "wrong_password_999"})
assert resp.status_code == 401
# 6th attempt should be rate limited (429)
resp = client.post("/api/login", json={"username": "testadmin", "password": "wrong_password_999"})
assert resp.status_code == 429
def test_lookup_endpoint_authenticated():
from app.auth import _login_failures
_login_failures.clear()
# Login first
resp = client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
assert resp.status_code == 200
resp = client.get("/api/lookup?q=reluctant")
assert resp.status_code == 200
data = resp.json()
assert data["query"] == "reluctant"
assert "translation" in data
assert "phonetic" in data
assert len(data["translation"]) > 0
def test_collections_and_batch_operations():
# Login
client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
# Create 3 cards in different collections
c1 = client.post("/api/cards", json={
"term": "ubiquitous",
"translation": "вездесущий",
"collection": "Academic",
"tags": "vocab, tech"
}).json()
c2 = client.post("/api/cards", json={
"term": "perseverance",
"translation": "настойчивость",
"collection": "Academic",
"tags": "character"
}).json()
c3 = client.post("/api/cards", json={
"term": "bite the bullet",
"translation": "стиснуть зубы",
"collection": "Idioms",
"tags": "idioms, daily"
}).json()
# Test GET /api/collections
resp = client.get("/api/collections")
assert resp.status_code == 200
cols = resp.json()
col_names = {c["name"]: c["count"] for c in cols}
assert "Academic" in col_names and col_names["Academic"] >= 2
assert "Idioms" in col_names and col_names["Idioms"] >= 1
# Test filter by collection in /api/cards
resp = client.get("/api/cards?collection=Academic")
assert resp.status_code == 200
academic_cards = resp.json()
assert all(c["collection"] == "Academic" for c in academic_cards["cards"])
assert any(c["id"] == c1["id"] for c in academic_cards["cards"])
assert any(c["id"] == c2["id"] for c in academic_cards["cards"])
assert not any(c["id"] == c3["id"] for c in academic_cards["cards"])
# Test Batch set_collection
batch_resp = client.post("/api/cards/batch", json={
"card_ids": [c1["id"], c2["id"]],
"action": "set_collection",
"value": "Advanced English"
})
assert batch_resp.status_code == 200
assert batch_resp.json()["affected"] == 2
# Verify collection updated
card1_check = client.get("/api/cards?search=ubiquitous").json()["cards"][0]
assert card1_check["collection"] == "Advanced English"
# Test Batch add_tags
batch_resp = client.post("/api/cards/batch", json={
"card_ids": [c1["id"], c2["id"]],
"action": "add_tags",
"value": "stage2, vocabulary"
})
assert batch_resp.status_code == 200
card1_tags = client.get("/api/cards?search=ubiquitous").json()["cards"][0]["tags"]
assert "stage2" in card1_tags
assert "vocabulary" in card1_tags
# Test Batch delete
batch_resp = client.post("/api/cards/batch", json={
"card_ids": [c1["id"], c2["id"], c3["id"]],
"action": "delete"
})
assert batch_resp.status_code == 200
assert batch_resp.json()["affected"] == 3
# Verify deleted
all_cards = client.get("/api/cards").json()["cards"]
deleted_ids = {c1["id"], c2["id"], c3["id"]}
assert not any(c["id"] in deleted_ids for c in all_cards)
def test_export_endpoints():
client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
# Create a test card with context sentence for cloze
card_resp = client.post("/api/cards", json={
"term": "serendipity",
"translation": "счастливая случайность",
"context_sentence": "Finding this book was pure serendipity.",
"phonetic": "ˌser.ənˈdɪp.ə.ti",
"part_of_speech": "Noun",
"definition_en": "Good luck in making unexpected discoveries",
"collection": "Vocabulary",
"tags": "literature, rare",
"notes": "Beautiful English word"
})
card = card_resp.json()
# 1. CSV Export
resp = client.get("/api/export/csv")
assert resp.status_code == 200
assert "text/csv" in resp.headers["Content-Type"]
assert resp.content.startswith(b"\xef\xbb\xbf") # UTF-8 BOM
csv_text = resp.content.decode("utf-8-sig")
assert "serendipity" in csv_text
assert "счастливая случайность" in csv_text
assert "term,translation,context,phonetic" in csv_text
# 2. JSON Export
resp = client.get("/api/export/json")
assert resp.status_code == 200
assert "application/json" in resp.headers["Content-Type"]
json_data = resp.json()
assert "cards" in json_data
assert any(c["term"] == "serendipity" for c in json_data["cards"])
# 3. Anki Export
resp = client.get("/api/export/anki")
assert resp.status_code == 200
anki_text = resp.text
assert "#separator:tab" in anki_text
assert "#html:true" in anki_text
assert "{{c1::serendipity}}" in anki_text # Cloze verified
assert "collection_Vocabulary" in anki_text
# 4. Markdown Export
resp = client.get("/api/export/markdown")
assert resp.status_code == 200
md_text = resp.text
assert "| **serendipity** |" in md_text
assert "### serendipity" in md_text
# Filter export by collection
resp_filtered = client.get("/api/export/json?collection=NonExistentCollection")
assert resp_filtered.status_code == 200
assert len(resp_filtered.json()["cards"]) == 0
# Clean up
client.delete(f"/api/cards/{card['id']}")
def test_sm2_algorithm_unit():
from app.srs import calculate_sm2, calculate_next_review
from datetime import datetime, timezone
base = datetime(2026, 9, 22, 12, 0, 0, tzinfo=timezone.utc)
# Rating 1 (Again)
res1 = calculate_sm2(1, reps=2, interval=6, ease=2.5, base_time=base)
assert res1["srs_reps"] == 0
assert res1["srs_interval"] == 1
assert res1["srs_ease"] == 2.3
assert res1["srs_state"] == "learning"
# Rating 2 (Hard)
res2 = calculate_sm2(2, reps=1, interval=1, ease=2.5, base_time=base)
assert res2["srs_reps"] == 2
assert res2["srs_interval"] == 1
assert res2["srs_ease"] == 2.35
assert res2["srs_state"] == "learning"
res2_later = calculate_sm2(2, reps=2, interval=10, ease=2.5, base_time=base)
assert res2_later["srs_reps"] == 3
assert res2_later["srs_interval"] == 12 # round(10 * 1.2) = 12
assert res2_later["srs_state"] == "review" # reps >= 3
# Rating 3 (Good)
res3_first = calculate_sm2(3, reps=0, interval=0, ease=2.5, base_time=base)
assert res3_first["srs_reps"] == 1
assert res3_first["srs_interval"] == 1
assert res3_first["srs_state"] == "review"
res3_second = calculate_sm2(3, reps=1, interval=1, ease=2.5, base_time=base)
assert res3_second["srs_reps"] == 2
assert res3_second["srs_interval"] == 6
assert res3_second["srs_state"] == "review"
res3_third = calculate_sm2(3, reps=2, interval=6, ease=2.5, base_time=base)
assert res3_third["srs_reps"] == 3
assert res3_third["srs_interval"] == 15 # round(6 * 2.5) = 15
res3_mastered = calculate_sm2(3, reps=3, interval=15, ease=2.5, base_time=base)
assert res3_mastered["srs_reps"] == 4
assert res3_mastered["srs_interval"] == 38 # round(15 * 2.5) = 38 >= 21
assert res3_mastered["srs_state"] == "mastered"
# Rating 4 (Easy)
res4_first = calculate_sm2(4, reps=0, interval=0, ease=2.5, base_time=base)
assert res4_first["srs_reps"] == 1
assert res4_first["srs_interval"] == 4
assert res4_first["srs_ease"] == 2.65
# Boundary checks: Ease minimum 1.3
res_min_ease = calculate_sm2(1, reps=0, interval=1, ease=1.35, base_time=base)
assert res_min_ease["srs_ease"] == 1.3
# Boundary checks: Ease maximum 3.0
res_max_ease = calculate_sm2(4, reps=1, interval=4, ease=2.95, base_time=base)
assert res_max_ease["srs_ease"] == 3.0
# Invalid rating
with pytest.raises(ValueError):
calculate_sm2(0)
with pytest.raises(ValueError):
calculate_sm2(5)
def test_srs_endpoints_flow():
client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
# Create a new card
card_resp = client.post("/api/cards", json={
"term": "meticulous",
"translation": "дотошный, педантичный",
"collection": "General"
})
card = card_resp.json()
card_id = card["id"]
assert card["srs_state"] == "new"
# 1. Check /api/srs/queue
resp = client.get("/api/srs/queue")
assert resp.status_code == 200
queue_data = resp.json()
assert any(c["id"] == card_id for c in queue_data["cards"])
# 2. Check /api/srs/stats before review
resp = client.get("/api/srs/stats")
assert resp.status_code == 200
stats_before = resp.json()
assert "new_cards" in stats_before
assert "learning" in stats_before
assert "mastered" in stats_before
assert "streak_days" in stats_before
# 3. Submit Review: Rating 3 (Good)
rev_resp = client.post("/api/srs/review", json={"card_id": card_id, "rating": 3})
assert rev_resp.status_code == 200
rev_data = rev_resp.json()
assert rev_data["status"] == "ok"
updated_card = rev_data["card"]
assert updated_card["srs_reps"] == 1
assert updated_card["srs_interval"] == 1
assert updated_card["srs_state"] == "review"
# 4. Check stats after review: total_reviews incremented, streak >= 1
stats_after = client.get("/api/srs/stats").json()
assert stats_after["total_reviews"] >= stats_before["total_reviews"] + 1
assert stats_after["streak_days"] >= 1
# 5. Review invalid rating
bad_rev = client.post("/api/srs/review", json={"card_id": card_id, "rating": 99})
assert bad_rev.status_code in (400, 422)
# 6. Review non-existent card
missing_rev = client.post("/api/srs/review", json={"card_id": 999999, "rating": 3})
assert missing_rev.status_code == 404
# Clean up
client.delete(f"/api/cards/{card_id}")
def test_pwa_routes_response():
resp_m = client.get("/manifest.json")
assert resp_m.status_code in (200, 404)
if resp_m.status_code == 200:
assert resp_m.headers["Content-Type"].startswith("application/manifest+json")
resp_sw = client.get("/sw.js")
assert resp_sw.status_code in (200, 404)
if resp_sw.status_code == 200:
assert resp_sw.headers["Service-Worker-Allowed"] == "/"
assert "no-cache" in resp_sw.headers["Cache-Control"]
def test_change_password_flow():
# 1. Create a dedicated user for password change test
test_uid = create_user("pwd_user", "old_secret_pass_123")
assert test_uid > 0
# 2. Login
login_resp = client.post("/api/login", json={"username": "pwd_user", "password": "old_secret_pass_123"})
assert login_resp.status_code == 200
# 3. Try to change with wrong old password
fail_resp = client.post("/api/change-password", json={
"old_password": "wrong_old_password",
"new_password": "brand_new_pass_456"
})
assert fail_resp.status_code == 400
# 4. Try to change with short new password (<6 chars)
short_resp = client.post("/api/change-password", json={
"old_password": "old_secret_pass_123",
"new_password": "123"
})
assert short_resp.status_code == 422
# 5. Successfully change password
ok_resp = client.post("/api/change-password", json={
"old_password": "old_secret_pass_123",
"new_password": "brand_new_pass_456"
})
assert ok_resp.status_code == 200
assert ok_resp.json()["status"] == "ok"
# 6. Verify old password no longer authenticates
old_auth = authenticate_user("pwd_user", "old_secret_pass_123")
assert old_auth is None
# 7. Verify new password authenticates
new_auth = authenticate_user("pwd_user", "brand_new_pass_456")
assert new_auth is not None
assert new_auth["username"] == "pwd_user"
# ---------------------------------------------------------------------------
# Stage 5: Bookshelf & Interactive Reader Tests
# ---------------------------------------------------------------------------
def make_mock_epub(title="The Time Machine", author="H. G. Wells", chapters=None):
import io, zipfile
if chapters is None:
chapters = [
("Chapter 1: The Machine", "The Time Traveller was expounding a recondite matter to us.
"),
("Chapter 2: The Journey", "I seemed to see the clock hands spin around.
")
]
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as z:
z.writestr("META-INF/container.xml", """
""")
manifest_items = []
spine_items = []
for i, (ch_title, ch_body) in enumerate(chapters, 1):
manifest_items.append(f'{ch_title}
{ch_body}
""")
opf = f"""
{title}
{author}
{''.join(manifest_items)}
{''.join(spine_items)}
"""
z.writestr("OEBPS/content.opf", opf)
return buf.getvalue()
def test_reader_unit():
from app.reader import count_words, parse_text, parse_epub
# 1. count_words
assert count_words("") == 0
assert count_words(" \n\t ") == 0
assert count_words("Hello world!") == 2
assert count_words("Привет, мир! Это тест.") == 4
assert count_words("Don't stop believing") == 3
# 2. parse_text with Markdown headers
md_content = """# My Book
Author: Arthur Dent
## Chapter 1: Hitchhiker
Far out in the uncharted backwaters.
## Chapter 2: The Towel
A towel is about the most useful thing.
"""
t, a, chs = parse_text(md_content)
assert t == "My Book"
assert a == "Arthur Dent"
assert len(chs) == 2
assert "Chapter 1: Hitchhiker" in chs[0]["title"]
assert "Far out" in chs[0]["content"]
# 3. parse_text with Chapter X headings
txt_content = """Chapter 1. The Boy Who Lived
Mr. and Mrs. Dursley of number four Privet Drive.
Chapter 2 - The Vanishing Glass
Nearly ten years had passed.
"""
t, a, chs = parse_text(txt_content, title="Harry Potter", author="J.K. Rowling")
assert t == "Harry Potter"
assert a == "J.K. Rowling"
assert len(chs) == 2
assert "Chapter 1" in chs[0]["title"]
assert "Chapter 2" in chs[1]["title"]
# 3b. Smart chapter detection: word-spelled (CHAPTER ONE), numbered sections, ALL CAPS, bold
smart_txt = """Some intro notes before starting.
CHAPTER ONE: THE BEGINNING
It was a cold day in April.
1.1 Network Fundamentals
Devices communicate over IP.
**Key Architecture**
The architecture is modular.
CONCLUSION AND NEXT STEPS
Final thoughts on the topic.
"""
st_title, st_auth, st_chs = parse_text(smart_txt, title="Smart Book")
assert len(st_chs) >= 4
titles = [c["title"] for c in st_chs]
assert any("CHAPTER ONE" in t for t in titles)
assert any("1.1 Network Fundamentals" in t for t in titles)
assert any("Key Architecture" in t for t in titles)
assert any("CONCLUSION AND NEXT STEPS" in t for t in titles)
# 4. parse_text chunking on long text
long_words = ["word"] * 7000
long_text = "\n\n".join([" ".join(long_words[i:i + 400]) for i in range(0, len(long_words), 400)])
t, a, chs = parse_text(long_text, title="Long Essay")
assert t == "Long Essay"
assert len(chs) >= 2
# 5. parse_epub
epub_bytes = make_mock_epub("Brave New World", "Aldous Huxley")
t, a, chs = parse_epub(epub_bytes)
assert t == "Brave New World"
assert a == "Aldous Huxley"
assert len(chs) == 2
assert "Chapter 1: The Machine" in chs[0]["title"]
assert "Time Traveller" in chs[0]["content"]
# 6. parse_epub invalid archive error
with pytest.raises(ValueError):
parse_epub(b"not a zip file")
def test_books_unauthenticated_blocked():
unauth = TestClient(app)
resp = unauth.get("/api/books")
assert resp.status_code == 401
resp = unauth.post("/api/books", json={"title": "Test", "content": "Text"})
assert resp.status_code == 401
resp = unauth.get("/api/books/1")
assert resp.status_code == 401
resp = unauth.put("/api/books/1/progress", json={"current_chapter": 1, "current_position": 0.5})
assert resp.status_code == 401
resp = unauth.delete("/api/books/1")
assert resp.status_code == 401
def test_books_crud_workflow():
# Login as admin
client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
# 1. Create book via JSON
payload = {
"title": "Fight Club",
"author": "Chuck Palahniuk",
"content": "Chapter 1\nTyler gets me a job.\n\nChapter 2\nA guy had come to the club.",
"source_type": "text"
}
resp = client.post("/api/books", json=payload)
assert resp.status_code == 200
book = resp.json()
assert book["id"] > 0
book_id = book["id"]
assert book["title"] == "Fight Club"
assert book["author"] == "Chuck Palahniuk"
assert book["source_type"] == "text"
assert book["chapters_count"] == 2
assert book["total_words"] > 0
assert book["current_chapter"] == 0
assert book["current_position"] == 0.0
# 2. List books (ensure chapters_json is excluded from list)
resp = client.get("/api/books")
assert resp.status_code == 200
books = resp.json()
assert isinstance(books, list)
target = next((b for b in books if b["id"] == book_id), None)
assert target is not None
assert target["title"] == "Fight Club"
assert target["chapters_count"] == 2
assert "chapters_json" not in target
# 3. Get single book (includes chapters_json and chapters)
resp = client.get(f"/api/books/{book_id}")
assert resp.status_code == 200
detail = resp.json()
assert detail["id"] == book_id
assert "chapters_json" in detail
assert "chapters" in detail
assert len(detail["chapters"]) == 2
assert "Chapter 1" in detail["chapters"][0]["title"]
# 4. Update reading progress
resp = client.put(f"/api/books/{book_id}/progress", json={
"current_chapter": 1,
"current_position": 0.75
})
assert resp.status_code == 200
updated = resp.json()
assert updated["current_chapter"] == 1
assert updated["current_position"] == 0.75
# 5. Delete book
resp = client.delete(f"/api/books/{book_id}")
assert resp.status_code == 200
assert resp.json()["status"] == "ok"
assert resp.json()["deleted_id"] == book_id
# 6. Verify deleted (404)
resp = client.get(f"/api/books/{book_id}")
assert resp.status_code == 404
def test_books_multipart_file_uploads():
# Login
client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
# 1. Upload TXT file
txt_data = b"Chapter 1: The Start\nOnce upon a time in a distant land.\n\nChapter 2: The End\nThey lived happily ever after."
resp = client.post(
"/api/books",
files={"file": ("fairytale.txt", txt_data, "text/plain")},
data={"title": "Fairy Tale", "author": "Brothers Grimm"}
)
assert resp.status_code == 200
book_txt = resp.json()
assert book_txt["title"] == "Fairy Tale"
assert book_txt["author"] == "Brothers Grimm"
assert book_txt["source_type"] == "txt"
assert book_txt["chapters_count"] == 2
# 2. Upload Markdown file
md_data = b"# The Cyberpunk Guide\nBy: Anonymous\n\n## Section 1\nWelcome to the matrix.\n\n## Section 2\nJack into cyberspace."
resp = client.post(
"/api/books",
files={"file": ("cyberpunk.md", md_data, "text/markdown")},
)
assert resp.status_code == 200
book_md = resp.json()
assert book_md["title"] == "The Cyberpunk Guide"
assert book_md["author"] == "Anonymous"
assert book_md["source_type"] == "md"
assert book_md["chapters_count"] == 2
# 3. Upload EPUB file
epub_bytes = make_mock_epub("The Invisible Man", "H. G. Wells")
resp = client.post(
"/api/books",
files={"file": ("invisible_man.epub", epub_bytes, "application/epub+zip")},
)
assert resp.status_code == 200
book_epub = resp.json()
assert book_epub["title"] == "The Invisible Man"
assert book_epub["author"] == "H. G. Wells"
assert book_epub["source_type"] == "epub"
assert book_epub["chapters_count"] == 2
def test_srs_queue_and_quiz_tag_filtering():
# Login
client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
# Create distinct tagged cards for book-scoped testing
c1 = client.post("/api/cards", json={
"term": "recondite",
"translation": "малопонятный, глубокий",
"tags": "book:time-machine, scifi",
"collection": "Literature"
}).json()
c2 = client.post("/api/cards", json={
"term": "invisible",
"translation": "невидимый",
"tags": "book:invisible-man, scifi",
"collection": "Literature"
}).json()
c3 = client.post("/api/cards", json={
"term": "soap",
"translation": "мыло",
"tags": "book:fight-club",
"collection": "Fiction"
}).json()
# 1. Filter queue by tag
resp = client.get("/api/srs/queue?tag=book:time-machine")
assert resp.status_code == 200
queue_cards = resp.json()["cards"]
card_ids = [c["id"] for c in queue_cards]
assert c1["id"] in card_ids
assert c2["id"] not in card_ids
assert c3["id"] not in card_ids
# 2. Filter quiz by tag
resp = client.get("/api/srs/quiz?tag=book:fight-club")
assert resp.status_code == 200
quiz_items = resp.json()["quiz"]
assert any(q["card"]["id"] == c3["id"] for q in quiz_items)
assert not any(q["card"]["id"] == c1["id"] for q in quiz_items)
assert not any(q["card"]["id"] == c2["id"] for q in quiz_items)
def test_gzip_compression():
# 1. Assets >= 1000 bytes receive GZip encoding when requested
resp_js = client.get("/static/js/app.js", headers={"Accept-Encoding": "gzip"})
assert resp_js.status_code == 200
assert resp_js.headers.get("content-encoding") == "gzip"
resp_css = client.get("/static/css/style.css", headers={"Accept-Encoding": "gzip"})
assert resp_css.status_code == 200
assert resp_css.headers.get("content-encoding") == "gzip"
resp_html = client.get("/", headers={"Accept-Encoding": "gzip"})
assert resp_html.status_code == 200
assert resp_html.headers.get("content-encoding") == "gzip"
# 2. Assets < 1000 bytes are not compressed (respecting minimum_size=1000)
resp_manifest = client.get("/manifest.json", headers={"Accept-Encoding": "gzip"})
assert resp_manifest.status_code == 200
assert resp_manifest.headers.get("content-encoding") is None
# 3. Requests without gzip Accept-Encoding do not receive gzip
resp_raw = client.get("/static/js/app.js", headers={"Accept-Encoding": "identity"})
assert resp_raw.status_code == 200
assert resp_raw.headers.get("content-encoding") is None
def test_duplicate_prevention_and_in_library_flag():
# Login as testadmin
client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
# 1. Create initial card for 'admired'
resp1 = client.post("/api/cards", json={
"term": "admired",
"translation": "восхищались",
"context_sentence": "People admired his courage.",
"tags": "jfk, book:kennedy",
"collection": "Literature",
})
assert resp1.status_code == 200
card1 = resp1.json()
card_id = card1["id"]
# 2. Try adding same term with different case ('Admired') and new context
resp2 = client.post("/api/cards", json={
"term": "Admired",
"translation": "восхищались",
"context_sentence": "He admired his older brother.",
"tags": "history, reader",
"collection": "General",
})
assert resp2.status_code == 200
card2 = resp2.json()
# Must be same card ID (no duplicate created!)
assert card2["id"] == card_id
assert card2.get("already_exists") is True
# Both contexts should be present and tags merged
assert "People admired his courage." in card2["context_sentence"]
assert "He admired his older brother." in card2["context_sentence"]
assert "book:kennedy" in card2["tags"]
assert "history" in card2["tags"]
# Verify database count has only 1 card for 'admired'
cards_list = client.get("/api/cards?search=admired").json()["cards"]
assert len(cards_list) == 1
assert cards_list[0]["id"] == card_id
# 3. Test /api/lookup returns in_library: True for existing word
resp_lookup_existing = client.get("/api/lookup?term=admired")
assert resp_lookup_existing.status_code == 200
data_existing = resp_lookup_existing.json()
assert data_existing["in_library"] is True
assert data_existing["library_card"]["id"] == card_id
# Test /api/lookup returns in_library: False for word not in user library
resp_lookup_new = client.get("/api/lookup?term=xylophone_unseen_xyz")
assert resp_lookup_new.status_code == 200
data_new = resp_lookup_new.json()
assert data_new["in_library"] is False
assert data_new["library_card"] is None
def make_mock_pdf(title="CCNA 200-301 Official Guide", author="Wendell Odom", with_outline=True):
import io
import pypdf
from pypdf.generic import DecodedStreamObject, NameObject, DictionaryObject
writer = pypdf.PdfWriter()
def add_text_page(text_lines):
page = writer.add_blank_page(width=400, height=400)
stream_content = [b"BT /F1 12 Tf 50 350 Td"]
for idx, l in enumerate(text_lines):
if idx > 0:
stream_content.append(b"0 -15 Td")
escaped = l.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)")
stream_content.append(f"({escaped}) Tj".encode("latin1"))
stream_content.append(b"ET")
content = DecodedStreamObject()
content.set_data(b" ".join(stream_content))
page[NameObject("/Contents")] = content
page[NameObject("/Resources")] = DictionaryObject({
NameObject("/Font"): DictionaryObject({
NameObject("/F1"): DictionaryObject({
NameObject("/Type"): NameObject("/Font"),
NameObject("/Subtype"): NameObject("/Type1"),
NameObject("/BaseFont"): NameObject("/Helvetica"),
})
})
})
p0_lines = ["Introduction to Networking", "CCNA Official Guide Header"] + [f"Introductory networking line {i} with several words for threshold." for i in range(12)]
p1_lines = ["Chapter 1: Ethernet Switching", "Router# show ip interface brief", "GigabitEthernet0/0 10.0.0.1 up up", "Router# configure terminal", "Switch(config)# vlan 10"] + [f"Ethernet switching explanation line {i} with concepts." for i in range(12)]
p2_lines = ["Chapter 2: IPv4 Routing", "Router# show ip route", "C 10.0.0.0/24 is directly connected"] + [f"IPv4 routing explanation line {i} with concepts." for i in range(12)]
add_text_page(p0_lines)
add_text_page(p1_lines)
add_text_page(p2_lines)
if with_outline:
writer.add_outline_item("Introduction", 0)
writer.add_outline_item("Chapter 1: Ethernet Switching", 1)
writer.add_outline_item("Chapter 2: IPv4 Routing", 2)
buf = io.BytesIO()
writer.write(buf)
return buf.getvalue()
def test_pdf_reader_unit():
import io
import pypdf
from app.reader import parse_pdf, format_cli_blocks, clean_page_lines
# 1. Test CLI block detection
sample_cli = "Check interfaces:\nRouter# show ip interface brief\nGigabitEthernet0/0 10.0.0.1 up up\nRouter# exit\nAll done."
formatted = format_cli_blocks(sample_cli)
assert "```text" in formatted
assert "Router# show ip interface brief" in formatted
assert "```" in formatted
# 2. Test outline-based PDF parsing
pdf_bytes = make_mock_pdf("CCNA Guide", "Cisco Press", with_outline=True)
title, author, chapters = parse_pdf(pdf_bytes, filename="ccna_guide.pdf")
assert title == "ccna_guide"
assert len(chapters) == 3
assert chapters[0]["title"] == "Introduction"
assert chapters[1]["title"] == "Chapter 1: Ethernet Switching"
assert chapters[2]["title"] == "Chapter 2: IPv4 Routing"
assert "```text" in chapters[1]["content"]
# 3. Test fallback when no outline (heading regex detection)
pdf_no_outline = make_mock_pdf("Manual", "Admin", with_outline=False)
t2, a2, chs2 = parse_pdf(pdf_no_outline, filename="network_manual.pdf")
assert len(chs2) >= 2
assert any("Chapter 1" in ch["title"] for ch in chs2)
# 4. Test Scanned PDF Guard (< 100 words raises ValueError)
empty_writer = pypdf.PdfWriter()
empty_writer.add_blank_page(width=100, height=100)
buf_scanned = io.BytesIO()
empty_writer.write(buf_scanned)
import pytest
with pytest.raises(ValueError, match="текстового слоя"):
parse_pdf(buf_scanned.getvalue(), filename="scan.pdf")
def test_books_pdf_multipart_upload():
# 1. Successful PDF upload
pdf_data = make_mock_pdf()
files = {"file": ("ccna_cert_guide.pdf", pdf_data, "application/pdf")}
resp = client.post("/api/books", files=files)
assert resp.status_code == 200
book = resp.json()
assert book["title"] == "ccna_cert_guide"
assert book["source_type"] == "pdf"
assert len(book["chapters"]) == 3
assert book["total_words"] > 100
book_id = book["id"]
# Verify retrieval
get_resp = client.get(f"/api/books/{book_id}")
assert get_resp.status_code == 200
assert get_resp.json()["source_type"] == "pdf"
# 2. Upload scanned PDF without OCR text returns 400 with informative detail
import io, pypdf
empty_writer = pypdf.PdfWriter()
empty_writer.add_blank_page(width=100, height=100)
buf = io.BytesIO()
empty_writer.write(buf)
scan_files = {"file": ("scanned_cisco.pdf", buf.getvalue(), "application/pdf")}
err_resp = client.post("/api/books", files=scan_files)
assert err_resp.status_code == 400
assert "текстового слоя" in err_resp.json()["detail"]
# Cleanup
client.delete(f"/api/books/{book_id}")
def test_http_connection_pooling():
import pytest
from app.dictionary import get_http_client, close_http_client, lookup_term
import asyncio
async def run_pool_test():
# 1. Singleton returns same instance
c1 = get_http_client()
c2 = get_http_client()
assert c1 is c2
assert not c1.is_closed
# 2. Check connection limits configuration
pool = c1._transport._pool
assert pool._max_keepalive_connections == 10
assert pool._max_connections == 20
# 3. Lookup using pooled client
res = await lookup_term("resilience")
assert res["query"] == "resilience"
assert "translation" in res
# 4. Clean closure on shutdown
await close_http_client()
assert c1.is_closed
# Next call creates a new active client
c3 = get_http_client()
assert c3 is not c1
assert not c3.is_closed
await close_http_client()
asyncio.run(run_pool_test())
def test_sqlite_pragma_tuning_and_composite_indexes():
from app.database import get_db, get_db_connection
# 1. Verify PRAGMAs on a newly opened connection
conn = get_db_connection()
try:
j_mode = conn.execute("PRAGMA journal_mode;").fetchone()[0]
sync = conn.execute("PRAGMA synchronous;").fetchone()[0]
mmap = conn.execute("PRAGMA mmap_size;").fetchone()[0]
cache = conn.execute("PRAGMA cache_size;").fetchone()[0]
temp = conn.execute("PRAGMA temp_store;").fetchone()[0]
fk = conn.execute("PRAGMA foreign_keys;").fetchone()[0]
assert j_mode.lower() == "wal"
assert sync == 1 # NORMAL
assert mmap == 268435456 # 256MB mmap
assert cache == -32000 # 32MB cache
assert temp == 2 # MEMORY
assert fk == 1 # foreign keys ON
# 2. Verify all composite indexes exist in sqlite_master
index_rows = conn.execute(
"SELECT name, tbl_name, sql FROM sqlite_master WHERE type = 'index'"
).fetchall()
index_names = {r["name"] for r in index_rows}
expected_indexes = {
"idx_vocab_user_created",
"idx_vocab_user_srs_state",
"idx_vocab_user_srs_created",
"idx_vocab_user_term_nocase",
"idx_books_user_updated",
"idx_srs_card",
"idx_srs_reviewed",
"idx_sessions_user",
}
for idx in expected_indexes:
assert idx in index_names, f"Expected index {idx} not found in database!"
# 3. Verify EXPLAIN QUERY PLAN uses the composite indexes
plan1 = conn.execute(
"EXPLAIN QUERY PLAN SELECT * FROM vocabulary_cards WHERE user_id = ? ORDER BY created_at DESC LIMIT 20",
(1,),
).fetchall()
plan1_str = " ".join([r[3] for r in plan1])
assert "idx_vocab_user_created" in plan1_str
plan2 = conn.execute(
"EXPLAIN QUERY PLAN SELECT * FROM vocabulary_cards WHERE user_id = ? AND srs_state = 'new' ORDER BY created_at DESC LIMIT 10",
(1,),
).fetchall()
plan2_str = " ".join([r[3] for r in plan2])
assert "idx_vocab_user_srs_created" in plan2_str or "idx_vocab_user_created" in plan2_str
plan3 = conn.execute(
"EXPLAIN QUERY PLAN SELECT * FROM books WHERE user_id = ? ORDER BY updated_at DESC",
(1,),
).fetchall()
plan3_str = " ".join([r[3] for r in plan3])
assert "idx_books_user_updated" in plan3_str
plan4 = conn.execute(
"EXPLAIN QUERY PLAN SELECT * FROM srs_logs WHERE card_id = ?",
(1,),
).fetchall()
plan4_str = " ".join([r[3] for r in plan4])
assert "idx_srs_card" in plan4_str
finally:
conn.close()
def test_debloat_dictionary_and_wiktionary_rest_parser():
import asyncio
import app.dictionary as dictionary
from app.dictionary import fetch_wiktionary_wikitext, fetch_wiktionary_definitions, lookup_term, get_http_client
# 1. Verify Datamuse is removed (YAGNI debloat)
assert not hasattr(dictionary, "fetch_datamuse"), "fetch_datamuse should be completely removed"
# 2. Test modern Wiktionary REST wikitext parser isolates English and extracts IPA & RU
async def run_checks():
client = get_http_client()
res_wiki = await fetch_wiktionary_wikitext("chat", client=client)
assert len(res_wiki["ipa"]) > 0
assert any("t͡ʃæt" in ipa for ipa in res_wiki["ipa"])
# Ensure French word "chat" (cat / кот) didn't leak into English translations
assert "кот" not in res_wiki["ru_translations"]
assert any(t in res_wiki["ru_translations"] for t in ["болтать", "разговаривать", "беседа", "разговор", "чатиться"])
# 3. Test definition cleaning and HTML entity decoding
res_defs = await fetch_wiktionary_definitions("reluctant", client=client)
assert "Adjective" in res_defs["parts_of_speech"]
assert len(res_defs["definitions"]) > 0
# 4. Test end-to-end lookup_term without Datamuse
term_res = await lookup_term("diligent")
assert term_res["query"] == "diligent"
assert len(term_res["translation"]) > 0
asyncio.run(run_checks())
def test_offline_dictionary_lookup_and_coverage():
"""Validates the Zero Daemon offline dictionary: 100k+ entries, word forms, instant mmap lookup."""
import asyncio
import time
from app.database import get_db
from app.dictionary import lookup_term
# 1. Verify offline_dictionary table exists and contains > 100,000 entries
with get_db() as conn:
count = conn.execute("SELECT COUNT(*) FROM offline_dictionary").fetchone()[0]
assert count > 100000, f"Expected > 100k entries, got {count}"
# Verify index exists
indexes = [r[1] for r in conn.execute("PRAGMA index_list('offline_dictionary')").fetchall()]
assert any("idx_offline_dict_term" in idx for idx in indexes)
# Check PRAGMA mmap_size
mmap_size = conn.execute("PRAGMA mmap_size").fetchone()[0]
assert mmap_size >= 268435456
async def run_lookups():
# 2. Test base words, inflected forms, irregulars, and IT terminology
test_words = [
("abandoned", "заброшенный"),
("ABANDONED", "заброшенный"), # Case-insensitivity
("hypotheses", "гипотез"), # Irregular plural
("men", "люди"), # Irregular plural
("went", "от go"), # Irregular past tense
("throughput", "пропускная способность"), # Networking / IT term
("quenched", "удовлетворить"), # Regular verb past tense (-ed)
]
for word, expected_snippet in test_words:
t0 = time.perf_counter()
res = await lookup_term(word)
dt_ms = (time.perf_counter() - t0) * 1000
assert dt_ms < 150.0, f"Lookup for {word} took {dt_ms:.2f}ms (expected < 150ms)"
assert res.get("source") == "offline" or res.get("translation"), f"Failed for {word}: {res}"
assert expected_snippet.lower() in res["translation"].lower(), f"Expected '{expected_snippet}' in '{res['translation']}'"
# 3. Test HTTP endpoint integration (/api/lookup)
login_res = client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
assert login_res.status_code == 200
api_res = client.get("/api/lookup?term=throughput")
assert api_res.status_code == 200
json_data = api_res.json()
assert json_data["query"] == "throughput"
assert "пропускная способность" in json_data["translation"].lower()
assert json_data["source"] == "offline"
assert "in_library" in json_data
asyncio.run(run_lookups())
def test_sentence_translation_ai_and_caching():
"""Validates local AI sentence translation endpoint, SQLite caching, and in-context word meaning."""
import time
from app.database import compute_sentence_hash, get_cached_sentence_translation, save_sentence_translation
# 1. Unauthenticated request must be blocked
client.cookies.clear()
unauth_res = client.post("/api/translate-sentence", json={"sentence": "He had a prodigious memory."})
assert unauth_res.status_code == 401
# 2. Login as admin
login_res = client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
assert login_res.status_code == 200
# 3. Empty sentence validation
empty_res = client.post("/api/translate-sentence", json={"sentence": " "})
assert empty_res.status_code == 400
# 4. Direct SQLite cache unit test
sentence = "He took the matter into his own hands and succeeded."
target_word = "matter"
h = compute_sentence_hash(sentence, target_word)
assert len(h) == 64
# Initially not cached
assert get_cached_sentence_translation(sentence, target_word) is None
# Save to cache
save_sentence_translation(
sentence=sentence,
translated_text="Он взял дело в свои руки и преуспел.",
target_word=target_word,
word_meaning="дело, вопрос",
source="unit_test",
)
cached_entry = get_cached_sentence_translation(sentence, target_word)
assert cached_entry is not None
assert cached_entry["translated_text"] == "Он взял дело в свои руки и преуспел."
assert cached_entry["word_meaning"] == "дело, вопрос"
assert cached_entry["source"] == "unit_test"
# 5. Fast cached retrieval via HTTP endpoint (< 10ms)
t0 = time.perf_counter()
api_res = client.post(
"/api/translate-sentence",
json={"sentence": sentence, "target_word": target_word},
)
dt_ms = (time.perf_counter() - t0) * 1000
assert api_res.status_code == 200
data = api_res.json()
assert data["cached"] is True
assert data["sentence_ru"] == "Он взял дело в свои руки и преуспел."
assert data["word_meaning"] == "дело, вопрос"
assert dt_ms < 50.0
# 6. Test AI status endpoint
status_res = client.get("/api/ai/status")
assert status_res.status_code == 200
status_data = status_res.json()
assert "online" in status_data
assert "models" in status_data
assert "active_model" in status_data
def test_phrasal_verb_detector_unit_and_integration():
"""Validates phrasal verb lemmatization, context detection, separable verbs, and API integration."""
from app.phrasal import lemmatize_word, detect_phrasal_verb
# 1. Lemmatization tests
assert "take" in lemmatize_word("took")
assert "give" in lemmatize_word("giving")
assert "figure" in lemmatize_word("figured")
assert "run" in lemmatize_word("running")
# 2. Direct unit detection tests
# Contiguous 2-word phrasal verb: clicking verb or particle
res_verb = detect_phrasal_verb("The plane took off into the sky.", "took")
assert res_verb is not None
assert res_verb["term"] == "take off"
assert res_verb["raw"] == "took off"
assert res_verb["separable"] is False
res_part = detect_phrasal_verb("The plane took off into the sky.", "off")
assert res_part is not None
assert res_part["term"] == "take off"
# Contiguous 3-word phrasal verb
res_triple = detect_phrasal_verb("She came up with a great solution.", "came")
assert res_triple is not None
assert res_triple["term"] == "come up with"
assert res_triple["raw"] == "came up with"
res_fwd = detect_phrasal_verb("I look forward to our trip.", "forward")
assert res_fwd is not None
assert res_fwd["term"] == "look forward to"
# Separable phrasal verb
res_sep_v = detect_phrasal_verb("Please turn it off before leaving.", "turn")
assert res_sep_v is not None
assert res_sep_v["term"] == "turn off"
assert res_sep_v["separable"] is True
res_sep_p = detect_phrasal_verb("Please turn it off before leaving.", "off")
assert res_sep_p is not None
assert res_sep_p["term"] == "turn off"
assert res_sep_p["separable"] is True
# Negative case (no phrasal verb)
res_none = detect_phrasal_verb("The red apple fell onto the soft grass.", "apple")
assert res_none is None
# 3. HTTP Endpoint Integration
client.cookies.clear()
unauth = client.get("/api/detect-phrasal-verb?sentence=The+plane+took+off&word=took")
assert unauth.status_code == 401
login_res = client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
assert login_res.status_code == 200
detect_res = client.get("/api/detect-phrasal-verb?sentence=The+plane+took+off&word=took")
assert detect_res.status_code == 200
data = detect_res.json()
assert data["detected"] is True
assert data["phrasal_verb"]["term"] == "take off"
# 4. Contextual integration inside /api/lookup
lookup_res = client.get("/api/lookup?term=took&context=The+plane+took+off+into+the+sky")
assert lookup_res.status_code == 200
lookup_data = lookup_res.json()
assert lookup_data["query"] == "took"
assert lookup_data.get("phrasal_verb") is not None
assert lookup_data["phrasal_verb"]["term"] == "take off"
assert "взлетать" in lookup_data["phrasal_verb"]["translation"]
assert lookup_data["phrasal_verb"]["in_library"] is False
# 5. In-library status check for phrasal verb
add_card = client.post("/api/cards", json={
"term": "take off",
"translation": "взлетать, снимать",
"context_sentence": "The plane took off into the sky.",
"collection": "Фразовые глаголы & Идиомы",
"tags": "phrasal-verbs, reader"
})
assert add_card.status_code == 200
lookup_after = client.get("/api/lookup?term=took&context=The+plane+took+off+into+the+sky")
assert lookup_after.status_code == 200
after_data = lookup_after.json()
assert after_data["phrasal_verb"]["in_library"] is True
assert after_data["phrasal_verb"]["library_card"] is not None
def test_cefr_testing_and_gap_ingestion():
# 1. Unauthenticated check
client.cookies.clear()
unauth = client.get("/api/cefr/levels")
assert unauth.status_code == 401
# 2. Authenticated user login
login_res = client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
assert login_res.status_code == 200
# 3. Check level summaries
levels_res = client.get("/api/cefr/levels")
assert levels_res.status_code == 200
levels = levels_res.json()["levels"]
assert len(levels) == 4
level_map = {l["level"]: l for l in levels}
assert "A1" in level_map and "A2" in level_map and "B1" in level_map and "B2" in level_map
assert level_map["A1"]["total_questions"] >= 65
assert level_map["A2"]["total_questions"] >= 65
assert level_map["B1"]["total_questions"] >= 65
assert level_map["B2"]["total_questions"] >= 65
# 4. Fetch quiz batch
quiz_res = client.get("/api/cefr/quiz?level=A2&count=5")
assert quiz_res.status_code == 200
quiz_data = quiz_res.json()
assert quiz_data["level"] == "A2"
assert len(quiz_data["quiz"]) == 5
q1 = quiz_data["quiz"][0]
assert "id" in q1
assert "sentence" in q1
assert "clue" in q1
assert len(q1["options"]) == 4
# Pick a distractor from the question's actual options
with get_db() as conn:
correct_term = conn.execute("SELECT term FROM cefr_questions WHERE id = ?", (q1["id"],)).fetchone()["term"]
wrong_opt = [opt for opt in q1["options"] if opt.lower() != correct_term.lower()][0]
ans_wrong = client.post("/api/cefr/answer", json={
"question_id": q1["id"],
"selected_option": wrong_opt
})
assert ans_wrong.status_code == 200
w_data = ans_wrong.json()
assert w_data["correct"] is False
assert w_data["card_created"] is True or w_data["card_updated"] is True
assert w_data["card_id"] is not None
assert w_data["correct_option"] != ""
assert w_data["explanation"] != ""
assert w_data["selected_card_id"] is not None
assert w_data["selected_card_term"] == wrong_opt
assert len(w_data["options_breakdown"]) == 4
# Verify target word card in SRS
card_res = client.get(f"/api/cards/{w_data['card_id']}")
assert card_res.status_code == 200
ingested_card = card_res.json()
assert ingested_card["srs_state"] == "learning"
assert "level-a2" in ingested_card["tags"]
assert "cefr-test" in ingested_card["tags"]
# Verify selected distractor card in SRS
sel_card_res = client.get(f"/api/cards/{w_data['selected_card_id']}")
assert sel_card_res.status_code == 200
sel_card = sel_card_res.json()
assert sel_card["srs_state"] == "learning"
assert "level-a2" in sel_card["tags"]
assert "cefr-test" in sel_card["tags"]
# Test manual ingestion of remaining option via /api/cefr/add-word
other_opt = [opt for opt in q1["options"] if opt != w_data["correct_option"] and opt != wrong_opt][0]
add_word_res = client.post("/api/cefr/add-word", json={
"level": "A2",
"term": other_opt,
"context_sentence": q1["sentence"]
})
assert add_word_res.status_code == 200
add_word_data = add_word_res.json()
assert add_word_data["status"] == "ok"
assert add_word_data["card_id"] is not None
assert add_word_data["term"] == other_opt
# 6. Check that mode='mistakes' returns this question
mistakes_res = client.get("/api/cefr/quiz?level=A2&mode=mistakes")
assert mistakes_res.status_code == 200
m_data = mistakes_res.json()
assert any(q["id"] == q1["id"] for q in m_data["quiz"])
# 7. Now answer CORRECTLY
ans_correct = client.post("/api/cefr/answer", json={
"question_id": q1["id"],
"selected_option": w_data["correct_option"]
})
assert ans_correct.status_code == 200
c_data = ans_correct.json()
assert c_data["correct"] is True
assert c_data["card_created"] is False
# 8. Check level summary reflects mastered count
levels_res2 = client.get("/api/cefr/levels")
l2_map = {l["level"]: l for l in levels_res2.json()["levels"]}
assert l2_map["A2"]["mastered_count"] >= 1
assert l2_map["A2"]["srs_cards_count"] >= 1
# 9. Test level reset
reset_res = client.post("/api/cefr/reset", json={"level": "A2"})
assert reset_res.status_code == 200
levels_res3 = client.get("/api/cefr/levels")
l3_map = {l["level"]: l for l in levels_res3.json()["levels"]}
assert l3_map["A2"]["mastered_count"] == 0
def test_grammar_engine_and_handwritten_journaling():
"""
Tests Task 16: Murphy Grammar Engine, micro-exercises, automated gap ingestion,
and handwritten paper notebook journaling toggle.
"""
# 1. Unauthenticated check
client.cookies.clear()
unauth_resp = client.get("/api/grammar/units")
assert unauth_resp.status_code == 401
# 2. Authenticated user login
login_res = client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
assert login_res.status_code == 200
# 3. Get all curated units
resp = client.get("/api/grammar/units")
assert resp.status_code == 200
data = resp.json()
assert "units" in data
units = data["units"]
assert len(units) >= 15
assert units[0]["unit_num"] == 1
assert "Present Continuous" in units[0]["title"]
assert units[0]["level"] == "A1-A2"
assert units[0]["is_completed"] is False
assert units[0]["handwritten_logged"] is False
unit_1_id = units[0]["id"]
# 3. Get single unit detail
detail_res = client.get(f"/api/grammar/units/{unit_1_id}")
assert detail_res.status_code == 200
u_data = detail_res.json()
assert u_data["unit_num"] == 1
assert "golden_formula" in u_data
assert "S + am / is / are" in u_data["golden_formula"]
assert len(u_data["contrasts"]) >= 1
assert "journaling_prompt" in u_data
assert "тетрадь" in u_data["journaling_prompt"].lower()
assert len(u_data["exercises"]) == 5
ex0 = u_data["exercises"][0]
assert len(ex0["options"]) == 4
# 4. Submit an INCORRECT answer to exercise 0 -> Automated Gap Ingestion
ans_wrong = client.post("/api/grammar/answer", json={
"unit_id": unit_1_id,
"exercise_idx": 0,
"selected_option": "wrong_choice_xyz"
})
assert ans_wrong.status_code == 200
w_data = ans_wrong.json()
assert w_data["correct"] is False
assert w_data["card_created"] is True
assert w_data["card_id"] is not None
assert w_data["correct_option"] == "are singing"
assert "Listen!" in w_data["explanation"]
# Verify the card was ingested into active vocabulary with collection 'Грамматика Мерфи'
card_res = client.get(f"/api/cards/{w_data['card_id']}")
assert card_res.status_code == 200
c_info = card_res.json()
assert c_info["srs_state"] == "learning"
assert c_info["collection"] == "Грамматика Мерфи"
assert "grammar" in c_info["tags"]
assert "murphy-unit-1" in c_info["tags"]
# 5. Submit a CORRECT answer
ans_correct = client.post("/api/grammar/answer", json={
"unit_id": unit_1_id,
"exercise_idx": 0,
"selected_option": w_data["correct_option"]
})
assert ans_correct.status_code == 200
c_data = ans_correct.json()
assert c_data["correct"] is True
assert c_data["card_created"] is False
# 6. Toggle handwritten paper notebook status
hw_res = client.post("/api/grammar/handwritten", json={
"unit_id": unit_1_id,
"handwritten": True
})
assert hw_res.status_code == 200
assert hw_res.json()["handwritten_logged"] is True
# 7. Record unit completion
comp_res = client.post("/api/grammar/complete", json={
"unit_id": unit_1_id,
"score": 80,
"handwritten_logged": True
})
assert comp_res.status_code == 200
assert comp_res.json()["score"] == 80
# 8. Verify list reflects updated progress
units_after = client.get("/api/grammar/units").json()["units"]
u1_after = next(u for u in units_after if u["id"] == unit_1_id)
assert u1_after["is_completed"] is True
assert u1_after["score"] == 80
assert u1_after["handwritten_logged"] is True
def test_background_tasks_and_book_mining():
# 1. Unauthenticated check
client.cookies.clear()
unauth = client.post("/api/tasks/mine-books", json={"level": "A1", "count": 2})
assert unauth.status_code == 401
# 2. Login
login_res = client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
assert login_res.status_code == 200
# 3. Create a book with known vocabulary
book_resp = client.post("/api/books", json={
"title": "Adventures in London",
"author": "Arthur Conan Doyle",
"content": "Chapter 1: The Arrival\nThey decided to invite their neighbor for breakfast. The journey was long and dangerous.\n\nChapter 2: The Mystery\nHe could not remember where he left his expensive ticket.",
"source_type": "text"
})
assert book_resp.status_code == 200
book_id = book_resp.json()["id"]
# 4. Start mining task
mine_resp = client.post("/api/tasks/mine-books", json={
"book_id": book_id,
"level": "A1",
"count": 2
})
assert mine_resp.status_code == 200
data = mine_resp.json()
assert data["status"] == "ok"
assert "task_id" in data
task_id = data["task_id"]
# 5. Check task status via active and ID endpoint
active_resp = client.get("/api/tasks/active")
assert active_resp.status_code == 200
# Wait briefly or poll task status
import time
t_data = None
for _ in range(30):
t_res = client.get(f"/api/tasks/{task_id}")
assert t_res.status_code == 200
t_data = t_res.json()["task"]
if t_data["status"] in ("completed", "failed", "cancelled"):
break
time.sleep(0.1)
assert t_data is not None
assert t_data["status"] == "completed"
assert t_data["progress_current"] > 0
# 6. Verify quiz retrieval includes source book metadata
quiz_res = client.get("/api/cefr/quiz?level=A1&count=20")
assert quiz_res.status_code == 200
items = quiz_res.json()["quiz"]
book_items = [q for q in items if q.get("source_book_title") == "Adventures in London"]
assert len(book_items) > 0
assert book_items[0]["source_book_title"] == "Adventures in London"
assert "source_chapter_title" in book_items[0]
# 7. Test task cancellation
from app.tasks import create_task
with get_db() as conn:
task_id_pending = create_task(conn, 1, "mine_books", total=10)
cancel_resp = client.post(f"/api/tasks/{task_id_pending}/cancel")
assert cancel_resp.status_code == 200
assert cancel_resp.json()["status"] == "ok"
assert cancel_resp.json()["cancelled"] is True
def test_rfc_parser_and_api(monkeypatch):
from app.reader import extract_rfc_number, parse_rfc_text
# 1. Number extraction
assert extract_rfc_number("9110") == 9110
assert extract_rfc_number("RFC 8446") == 8446
assert extract_rfc_number("rfc-6749") == 6749
assert extract_rfc_number("https://www.rfc-editor.org/rfc/rfc7540.txt") == 7540
assert extract_rfc_number("https://datatracker.ietf.org/doc/rfc9000/") == 9000
with pytest.raises(ValueError):
extract_rfc_number("invalid-rfc-string-without-number")
with pytest.raises(ValueError):
extract_rfc_number("rfc -5")
# 2. Parsing raw text with pagination artifacts
sample_rfc = """Internet Engineering Task Force (IETF)
Request for Comments: 9999
Category: Standards Track
Test Protocol Specification
Abstract
This is a test abstract for RFC 9999 demonstrating cleaning.
Table of Contents
1. Introduction .................................................... 1
2. Protocol Framing ................................................ 2
Authors' Addresses ................................................. 3
Standards Track [Page 1]
\x0c
RFC 9999 Test Protocol Specification June 2026
1. Introduction
The Test Protocol is designed for testing RFC parsers.
It provides ultra-clean chapter segmentation.
2. Protocol Framing
Framing uses standard binary encodings.
Authors' Addresses
Jane Doe
Example Corp
"""
clean_title, author, chapters, total_words = parse_rfc_text(sample_rfc, rfc_number=9999)
assert clean_title == "RFC 9999: Test Protocol Specification"
assert author == "IETF"
assert len(chapters) >= 3
# Check that pagination footer/header was stripped
for ch in chapters:
assert "[Page 1]" not in ch["content"]
assert "\x0c" not in ch["content"]
assert any("Introduction" in ch["title"] for ch in chapters)
assert any("Framing" in ch["title"] for ch in chapters)
assert any("Addresses" in ch["title"] for ch in chapters)
assert total_words > 20
# 3. Test API endpoint unauthenticated
anon_client = TestClient(app)
resp = anon_client.post("/api/books/fetch-rfc", json={"rfc": "9999"})
assert resp.status_code == 401
# 4. Test API endpoint authenticated with mock fetch
import app.main as main_mod
def mock_fetch_and_parse_rfc(query):
return ("RFC 9999: Test Protocol Specification", "IETF (Jane Doe)", chapters, 9999, total_words)
monkeypatch.setattr(main_mod, "fetch_and_parse_rfc", mock_fetch_and_parse_rfc)
# Login demo user
login_resp = client.post("/api/login", json={"username": "demo_user", "password": "demo_pass_999"})
assert login_resp.status_code == 200
# Invalid input
bad_resp = client.post("/api/books/fetch-rfc", json={"rfc": "no_number_here"})
assert bad_resp.status_code == 400
# Fetch valid RFC
fetch_resp = client.post("/api/books/fetch-rfc", json={"rfc": "9999"})
assert fetch_resp.status_code == 200
book_data = fetch_resp.json()
assert book_data["source_type"] == "rfc"
assert book_data["title"] == "RFC 9999: Test Protocol Specification"
assert book_data["already_existed"] is False
assert len(book_data["chapters"]) >= 3
book_id = book_data["id"]
# Fetch again without force -> should return existing book
fetch_again = client.post("/api/books/fetch-rfc", json={"rfc": "9999"})
assert fetch_again.status_code == 200
again_data = fetch_again.json()
assert again_data["id"] == book_id
assert again_data["already_existed"] is True
# Check bookshelf retrieval includes RFC book
shelf_resp = client.get("/api/books")
assert shelf_resp.status_code == 200
shelf_books = shelf_resp.json()
rfc_books = [b for b in shelf_books if b["source_type"] == "rfc"]
assert len(rfc_books) > 0
assert rfc_books[0]["id"] == book_id
def test_cefr_context_resolution_and_repair():
from app.cefr import resolve_context_sentence_for_term, process_cefr_answer, ingest_cefr_word
from app.database import repair_mismatched_cefr_contexts
with get_db() as conn:
# 1. Test resolve_context_sentence_for_term
# Case A: candidate has the term and placeholder '___'
res_a = resolve_context_sentence_for_term(
conn,
"decision",
"Choosing a career path is a major life ___."
)
assert res_a == "Choosing a career path is a major life decision."
assert "___" not in res_a
# Case B: candidate does NOT match the term, term exists in cefr_questions
res_b = resolve_context_sentence_for_term(
conn,
"opportunity",
"Choosing a career path is a major life ___."
)
assert "opportunity" in res_b.lower()
assert "career path" not in res_b
assert "___" not in res_b
# 2. Test process_cefr_answer mistake ingestion
# Seed test user
user_row = conn.execute("SELECT id FROM users WHERE username = 'testadmin'").fetchone()
if not user_row:
create_user("testadmin", "testpass12345")
user_row = conn.execute("SELECT id FROM users WHERE username = 'testadmin'").fetchone()
assert user_row is not None
user_id = user_row["id"]
# Find or create a question for 'decision'
q = conn.execute(
"SELECT id, level, term, sentence, options_json FROM cefr_questions WHERE term = 'decision' LIMIT 1"
).fetchone()
if not q:
conn.execute(
"""
INSERT INTO cefr_questions (level, term, translation, sentence, options_json, explanation)
VALUES ('B1', 'decision', 'решение', 'Choosing a career path is a major life ___.', '["decision", "opportunity", "behavior", "relationship"]', 'Major decision.')
"""
)
q = conn.execute("SELECT id, level, term, sentence, options_json FROM cefr_questions WHERE term = 'decision' LIMIT 1").fetchone()
result = process_cefr_answer(conn, user_id, q["id"], "opportunity")
assert result["correct"] is False
assert result["correct_option"] == "decision"
assert result["selected_option"] == "opportunity"
# Check options_breakdown has distinct authentic context sentences
breakdown = result["options_breakdown"]
assert len(breakdown) == 4
for item in breakdown:
assert "context_sentence" in item
assert "___" not in item["context_sentence"]
assert item["term"].lower() in item["context_sentence"].lower()
# Check vocabulary_cards in DB:
target_card = conn.execute(
"SELECT * FROM vocabulary_cards WHERE user_id = ? AND lower(term) = 'decision'",
(user_id,)
).fetchone()
assert target_card is not None
assert "decision" in target_card["context_sentence"].lower()
assert "___" not in target_card["context_sentence"]
chosen_card = conn.execute(
"SELECT * FROM vocabulary_cards WHERE user_id = ? AND lower(term) = 'opportunity'",
(user_id,)
).fetchone()
assert chosen_card is not None
assert "opportunity" in chosen_card["context_sentence"].lower()
assert "career path" not in chosen_card["context_sentence"]
assert "___" not in chosen_card["context_sentence"]
# 3. Test ingest_cefr_word
added_word = ingest_cefr_word(
conn,
user_id=user_id,
level="B1",
term="behavior",
context_sentence="Choosing a career path is a major life ___."
)
assert added_word["term"] == "behavior"
assert "behavior" in added_word["context_sentence"].lower()
assert "career path" not in added_word["context_sentence"]
assert "___" not in added_word["context_sentence"]
# 4. Test repair_mismatched_cefr_contexts
# Insert a card with corrupted blank
conn.execute("DELETE FROM vocabulary_cards WHERE user_id = ? AND lower(term) = 'negotiate'", (user_id,))
conn.execute(
"""
INSERT INTO vocabulary_cards (user_id, term, translation, context_sentence, tags)
VALUES (?, 'negotiate', 'вести переговоры', 'With persistent practice, you can ___ fluency in English.', 'cefr-test')
""",
(user_id,)
)
corrupted = conn.execute("SELECT context_sentence FROM vocabulary_cards WHERE user_id = ? AND term = 'negotiate'", (user_id,)).fetchone()
assert "___" in corrupted["context_sentence"]
repair_mismatched_cefr_contexts(conn)
repaired = conn.execute("SELECT context_sentence FROM vocabulary_cards WHERE user_id = ? AND term = 'negotiate'", (user_id,)).fetchone()
assert "___" not in repaired["context_sentence"]
assert "negotiate" in repaired["context_sentence"].lower()
def test_robust_sentence_split_and_ai_translation(monkeypatch):
from app.miner import robust_sentence_split
from app.ai import build_translation_prompt
# 1. Test robust sentence segmentation
# Case A: Initials in names (John F. Kennedy)
s_jfk = "Perhaps the most famous is the John F. Kennedy International Airport in New York, but there is also the John F. Kennedy Space Center."
splits_jfk = robust_sentence_split(s_jfk)
assert len(splits_jfk) == 1
assert "John F. Kennedy International Airport" in splits_jfk[0]
# Case B: Honorifics with dots (Mr., Mrs., Dr., etc.)
s_dursley = "When Mr. and Mrs. Dursley woke up on the dull, gray Tuesday our story starts, there was nothing about the cloudy sky outside to suggest that strange things would happen."
splits_dursley = robust_sentence_split(s_dursley)
assert len(splits_dursley) == 1
assert "Mr. and Mrs. Dursley" in splits_dursley[0]
# Case C: Abbreviations (No. 4, U.S. Navy)
s_no4 = "They lived at No. 4 Privet Drive, which was a very quiet street. Next sentence starts here."
splits_no4 = robust_sentence_split(s_no4)
assert len(splits_no4) == 2
assert splits_no4[0] == "They lived at No. 4 Privet Drive, which was a very quiet street."
assert splits_no4[1] == "Next sentence starts here."
s_us = "The men thought that the U.S. Navy would send a plane to look for them. They waited patiently."
splits_us = robust_sentence_split(s_us)
assert len(splits_us) == 2
assert "U.S. Navy" in splits_us[0]
# Case D: Complex paragraph with Mr., Mrs., dialogue
s_hp = (
"At half past eight, Mr. Dursley picked up his briefcase, pecked Mrs. Dursley on the cheek, "
"and tried to kiss Dudley good-bye but missed, because Dudley was now having a tantrum and "
"throwing his cereal at the walls. “Little tyke,” chortled Mr. Dursley as he left the house."
)
splits_hp = robust_sentence_split(s_hp)
assert len(splits_hp) == 2
assert "pecked Mrs. Dursley on the cheek" in splits_hp[0]
assert splits_hp[0].startswith("At half past eight, Mr. Dursley")
assert "Little tyke" in splits_hp[1]
# 2. Test build_translation_prompt
prompt = build_translation_prompt(
sentence="At half past eight, Mr. Dursley picked up his briefcase.",
target_word="briefcase",
context="Full context paragraph here.",
dict_hint="портфель, кейс, дипломат",
)
assert "Контекст абзаца:" in prompt
assert "Full context paragraph here." in prompt
assert "briefcase" in prompt
assert "портфель, кейс, дипломат" in prompt
assert "Mr. -> мистер" in prompt
assert "БЕЗ ДОСЛОВНОГО КАЛЬКИРОВАНИЯ" in prompt
assert "ПОЛНОТА И СВЯЗНОСТЬ" in prompt
# 3. Test API endpoint /api/translate-sentence
# Authenticate client
login_resp = client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
assert login_resp.status_code == 200
# Mock translate_sentence_ai
import app.main as main_mod
async def mock_translate_sentence_ai(sentence, target_word="", context=""):
return {
"original_sentence": sentence,
"sentence_ru": "В половине девятого мистер Дурсли взял свой портфель.",
"target_word": target_word,
"word_meaning": "портфель",
"source": "mock_ollama",
"cached": False,
}
monkeypatch.setattr(main_mod, "translate_sentence_ai", mock_translate_sentence_ai)
tr_resp = client.post("/api/translate-sentence", json={
"sentence": "At half past eight, Mr. Dursley picked up his briefcase.",
"target_word": "briefcase",
"context": "Context paragraph",
})
assert tr_resp.status_code == 200
tr_data = tr_resp.json()
assert tr_data["sentence_ru"] == "В половине девятого мистер Дурсли взял свой портфель."
assert tr_data["word_meaning"] == "портфель"
assert tr_data["target_word"] == "briefcase"
def test_phrasal_verb_detection_no_false_positive_for_noun_phrase():
from app.phrasal import detect_phrasal_verb
# Seed temporary test dictionary with tawny owl
with get_db() as conn:
conn.execute(
"INSERT OR IGNORE INTO offline_dictionary (term, translation, phonetic, pos) VALUES (?, ?, '', ?)",
("tawny owl", "неясыть, неясыть серая", "Noun"),
)
conn.execute(
"INSERT OR IGNORE INTO offline_dictionary (term, translation, phonetic, pos) VALUES (?, ?, '', ?)",
("owl", "сыч, сова, филин", "Noun"),
)
conn.execute(
"INSERT OR IGNORE INTO offline_dictionary (term, translation, phonetic, pos) VALUES (?, ?, '', ?)",
("tawny", "рыжевато-коричневый", "Adjective"),
)
sentence = "None of them noticed a large, tawny owl flutter past the window."
# tawny owl is a noun phrase (an animal species), NOT a phrasal verb!
assert detect_phrasal_verb(sentence, "owl") is None
assert detect_phrasal_verb(sentence, "tawny") is None
assert detect_phrasal_verb(sentence, "large") is None
# Genuine phrasal verbs must still be detected
s2 = "None of them noticed him take off his coat."
pv2 = detect_phrasal_verb(s2, "take")
assert pv2 is not None
assert pv2["term"] == "take off"
s3 = "He turned it off quietly."
pv3 = detect_phrasal_verb(s3, "turned")
assert pv3 is not None
assert pv3["term"] == "turn off"
def test_sentence_dictionary_hints_and_fauna_precision():
from app.ai import get_sentence_dictionary_hints, build_translation_prompt
with get_db() as conn:
conn.execute(
"INSERT OR IGNORE INTO offline_dictionary (term, translation, phonetic, pos) VALUES (?, ?, '', ?)",
("tawny owl", "неясыть, неясыть серая", "Noun"),
)
conn.execute(
"INSERT OR IGNORE INTO offline_dictionary (term, translation, phonetic, pos) VALUES (?, ?, '', ?)",
("owl", "сыч, сова, филин", "Noun"),
)
conn.execute(
"INSERT OR IGNORE INTO offline_dictionary (term, translation, phonetic, pos) VALUES (?, ?, '', ?)",
("large", "большой, крупный", "Adjective"),
)
sentence = "None of them noticed a large, tawny owl flutter past the window."
hints = get_sentence_dictionary_hints(sentence, "tawny")
assert "owl" in hints
assert "НЕ орёл" in hints
assert "large" in hints
prompt = build_translation_prompt(
sentence=sentence,
target_word="tawny",
context="",
dict_hint="рыжевато-коричневый",
sentence_hints=hints,
)
assert "Словарные подсказки по ключевым словам предложения" in prompt
assert "НЕ орёл" in prompt
assert "ПОЛНОТА И СВЯЗНОСТЬ" in prompt
assert "large" in prompt