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
assert "<" not in res_defs["definitions"][0] and ">" not in 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
assert len(term_res["definitions"]) > 0
asyncio.run(run_checks())