2660 lines
100 KiB
Python
2660 lines
100 KiB
Python
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", "<p>The Time Traveller was expounding a recondite matter to us.</p>"),
|
||
("Chapter 2: The Journey", "<p>I seemed to see the clock hands spin around.</p>")
|
||
]
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as z:
|
||
z.writestr("META-INF/container.xml", """<?xml version="1.0"?>
|
||
<container version="1.0" xmlns="urn:oasis:names:tc:opendocument:xmlns:container">
|
||
<rootfiles>
|
||
<rootfile full-path="OEBPS/content.opf" media-type="application/oebps-package+xml"/>
|
||
</rootfiles>
|
||
</container>""")
|
||
manifest_items = []
|
||
spine_items = []
|
||
for i, (ch_title, ch_body) in enumerate(chapters, 1):
|
||
manifest_items.append(f'<item id="c{i}" href="ch{i}.xhtml" media-type="application/xhtml+xml"/>')
|
||
spine_items.append(f'<itemref idref="c{i}"/>')
|
||
z.writestr(f"OEBPS/ch{i}.xhtml", f"""<!DOCTYPE html>
|
||
<html>
|
||
<head><title>{ch_title}</title></head>
|
||
<body>
|
||
<h1>{ch_title}</h1>
|
||
{ch_body}
|
||
</body>
|
||
</html>""")
|
||
|
||
opf = f"""<?xml version="1.0" encoding="utf-8"?>
|
||
<package xmlns="http://www.idpf.org/2007/opf" version="3.0" unique-identifier="id">
|
||
<metadata xmlns:dc="http://purl.org/dc/elements/1.1/">
|
||
<dc:title>{title}</dc:title>
|
||
<dc:creator>{author}</dc:creator>
|
||
</metadata>
|
||
<manifest>
|
||
{''.join(manifest_items)}
|
||
</manifest>
|
||
<spine>
|
||
{''.join(spine_items)}
|
||
</spine>
|
||
</package>"""
|
||
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 "cheatsheet_table" in u_data
|
||
assert "|" in u_data["cheatsheet_table"]
|
||
assert "detailed_notes" in u_data
|
||
assert len(u_data["detailed_notes"]) > 100
|
||
assert "###" in u_data["detailed_notes"]
|
||
assert len(u_data["exercises"]) == 5
|
||
|
||
# Check lookup by unit_num directly
|
||
num_res = client.get("/api/grammar/units/1")
|
||
assert num_res.status_code == 200
|
||
assert num_res.json()["id"] == unit_1_id
|
||
|
||
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
|
||
|
||
|
||
def test_clean_russian_translation_removes_romanization():
|
||
from app.dictionary import clean_russian_translation
|
||
|
||
raw_flutter = "трепет, порхание, развеваться át’sjaпорхать át’/, бить крыльями ’ krýl’jamiмахать крыльями át’ krýl’jami/"
|
||
clean_flutter = clean_russian_translation(raw_flutter)
|
||
assert clean_flutter == "трепет, порхание, развеваться, порхать, бить крыльями, махать крыльями"
|
||
assert "át’sja" not in clean_flutter
|
||
assert "krýl’jami" not in clean_flutter
|
||
assert "/" not in clean_flutter
|
||
|
||
raw_tawny = "рыжевато-коричневый, рыжевато-коричневый ževáto-koríčnevyj/"
|
||
clean_tawny = clean_russian_translation(raw_tawny)
|
||
assert clean_tawny == "рыжевато-коричневый"
|
||
|
||
raw_like = "подобный, похожий óžij/, как, типа ípa/"
|
||
clean_like = clean_russian_translation(raw_like)
|
||
assert clean_like == "подобный, похожий, как, типа"
|
||
|
||
|
||
def test_extract_wikipedia_title_and_lang():
|
||
from app.reader import extract_wikipedia_title_and_lang
|
||
|
||
t1, l1 = extract_wikipedia_title_and_lang("https://simple.wikipedia.org/wiki/Solar_System")
|
||
assert t1 == "Solar_System"
|
||
assert l1 == "simple"
|
||
|
||
t2, l2 = extract_wikipedia_title_and_lang("https://en.wikipedia.org/wiki/Quantum_mechanics")
|
||
assert t2 == "Quantum_mechanics"
|
||
assert l2 == "en"
|
||
|
||
t3, l3 = extract_wikipedia_title_and_lang("Artificial Intelligence", default_lang="simple")
|
||
assert t3 == "Artificial_Intelligence"
|
||
assert l3 == "simple"
|
||
|
||
|
||
def test_clean_wiki_html_unit():
|
||
from app.reader import clean_wiki_html
|
||
|
||
sample_parsoid_html = """
|
||
<!DOCTYPE html>
|
||
<html>
|
||
<body>
|
||
<style>.some-css { color: red; }</style>
|
||
<script>console.log("bad");</script>
|
||
<div class="hatnote">This article is about astronomy.</div>
|
||
<table class="infobox">
|
||
<tr><th>Lead</th></tr>
|
||
<tr><td><img src="//thumb.wikimedia.org/wikipedia/commons/thumb/a/a1/Sun.jpg/300px-Sun.jpg" alt="The Sun"></td></tr>
|
||
<tr><td>Orbital info</td></tr>
|
||
</table>
|
||
<p>The <b>Sun</b> is the star at the center of the Solar System<sup class="reference">[1]</sup>.</p>
|
||
<figure typeof="mw:File/Thumb">
|
||
<img src="//thumb.wikimedia.org/wikipedia/commons/thumb/b/b2/Solar_flare.jpg/400px-Solar_flare.jpg" alt="Solar flare">
|
||
<figcaption>A powerful <a href="./Solar_flare">solar flare</a> erupting from the surface</figcaption>
|
||
</figure>
|
||
<p>It is a nearly perfect ball of hot plasma<sup class="reference">[citation needed]</sup>.</p>
|
||
|
||
<h2><span class="mw-headline">Characteristics</span><span class="mw-editsection">[edit]</span></h2>
|
||
<p>The Sun is a G-type main-sequence star.</p>
|
||
<table class="wikitable">
|
||
<thead><tr><th>Parameter</th><th>Value</th></tr></thead>
|
||
<tbody><tr><td>Radius</td><td>696,340 km</td></tr></tbody>
|
||
</table>
|
||
|
||
<h2>References</h2>
|
||
<p>1. Solar Physics Journal 2024.</p>
|
||
|
||
<h2>Other websites</h2>
|
||
<ul><li>NASA Sun info</li></ul>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
chapters, cover_url, total_words = clean_wiki_html(sample_parsoid_html, article_title="Sun")
|
||
|
||
assert cover_url.startswith("https://thumb.wikimedia.org/")
|
||
assert len(chapters) == 2 # Overview and Characteristics (References and Other websites ignored)
|
||
|
||
ch1 = chapters[0]
|
||
assert ch1["title"] == "Overview"
|
||
assert "star at the center of the Solar System" in ch1["content"]
|
||
assert "[1]" not in ch1["content"]
|
||
assert "[citation needed]" not in ch1["content"]
|
||
assert ".some-css" not in ch1["content"]
|
||
assert "bad" not in ch1["content"]
|
||
assert "<figure class=\"reader-figure\">" in ch1["content"]
|
||
assert "https://thumb.wikimedia.org/wikipedia/commons/thumb/b/b2/Solar_flare.jpg/400px-Solar_flare.jpg" in ch1["content"]
|
||
assert "A powerful solar flare erupting from the surface" in ch1["content"]
|
||
|
||
ch2 = chapters[1]
|
||
assert ch2["title"] == "Characteristics"
|
||
assert "[edit]" not in ch2["title"]
|
||
assert "G-type main-sequence star" in ch2["content"]
|
||
assert "reader-table" in ch2["content"]
|
||
assert "696,340 km" in ch2["content"]
|
||
assert total_words > 20
|
||
|
||
|
||
def test_wikipedia_api_endpoints_workflow(monkeypatch):
|
||
import urllib.request
|
||
import io
|
||
from app.reader import clean_wiki_html
|
||
|
||
sample_html = """
|
||
<html><body>
|
||
<table class="infobox">
|
||
<tr><td><img src="//thumb.wikimedia.org/test/cover.jpg" alt="Planet"></td></tr>
|
||
</table>
|
||
<p>Earth is the third planet from the Sun and the only astronomical object known to harbor life.</p>
|
||
<h2>Atmosphere</h2>
|
||
<p>The atmosphere of Earth is composed mostly of nitrogen and oxygen.</p>
|
||
</body></html>
|
||
"""
|
||
|
||
sample_summary = {
|
||
"title": "Earth",
|
||
"thumbnail": {"source": "https://thumb.wikimedia.org/test/cover.jpg"}
|
||
}
|
||
|
||
class MockResponse:
|
||
def __init__(self, data_bytes, status=200):
|
||
self.data_bytes = data_bytes
|
||
self.status = status
|
||
|
||
def read(self, *args, **kwargs):
|
||
return self.data_bytes
|
||
|
||
def decode(self, *args, **kwargs):
|
||
return self.data_bytes.decode(*args, **kwargs)
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *args):
|
||
pass
|
||
|
||
def mock_urlopen(req, timeout=None):
|
||
url = req.full_url if hasattr(req, "full_url") else str(req)
|
||
if "summary" in url:
|
||
return MockResponse(json.dumps(sample_summary).encode("utf-8"))
|
||
elif "html" in url:
|
||
return MockResponse(sample_html.encode("utf-8"))
|
||
return MockResponse(b"{}")
|
||
|
||
monkeypatch.setattr(urllib.request, "urlopen", mock_urlopen)
|
||
|
||
# 1. Login user
|
||
login_resp = client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
|
||
assert login_resp.status_code == 200
|
||
token = login_resp.cookies.get(config.SESSION_COOKIE_NAME)
|
||
auth_headers = {"Cookie": f"{config.SESSION_COOKIE_NAME}={token}"}
|
||
|
||
# 2. Fetch Wikipedia article
|
||
resp = client.post(
|
||
"/api/books/fetch-wikipedia",
|
||
json={"query": "Earth", "lang": "simple", "topic": "science"},
|
||
headers=auth_headers
|
||
)
|
||
assert resp.status_code == 200
|
||
book_data = resp.json()
|
||
assert book_data["title"] == "Earth"
|
||
assert book_data["source_type"] == "wikipedia"
|
||
assert "cover.jpg" in book_data["cover_url"]
|
||
assert len(book_data["chapters"]) >= 1
|
||
assert book_data["already_existed"] is False
|
||
book_id = book_data["id"]
|
||
|
||
# 3. Fetch again without force - should return already_existed=True
|
||
resp2 = client.post(
|
||
"/api/books/fetch-wikipedia",
|
||
json={"query": "Earth", "lang": "simple", "force": False},
|
||
headers=auth_headers
|
||
)
|
||
assert resp2.status_code == 200
|
||
assert resp2.json()["already_existed"] is True
|
||
assert resp2.json()["id"] == book_id
|
||
|
||
# 4. Check book list includes cover_url and source_type
|
||
list_resp = client.get("/api/books", headers=auth_headers)
|
||
assert list_resp.status_code == 200
|
||
books = list_resp.json()
|
||
wiki_book = next((b for b in books if b["id"] == book_id), None)
|
||
assert wiki_book is not None
|
||
assert wiki_book["source_type"] == "wikipedia"
|
||
assert "cover.jpg" in wiki_book["cover_url"]
|
||
|
||
# 5. Update reading progress
|
||
prog_resp = client.put(
|
||
f"/api/books/{book_id}/progress",
|
||
json={"current_chapter": 1, "current_position": 0.5},
|
||
headers=auth_headers
|
||
)
|
||
assert prog_resp.status_code == 200
|
||
assert prog_resp.json()["current_chapter"] == 1
|
||
assert prog_resp.json()["current_position"] == 0.5
|
||
assert prog_resp.json()["cover_url"] != ""
|
||
|
||
# 6. Delete book
|
||
del_resp = client.delete(f"/api/books/{book_id}", headers=auth_headers)
|
||
assert del_resp.status_code == 200
|
||
assert del_resp.json()["deleted_id"] == book_id
|
||
|
||
|
||
def test_piper_tts_endpoints_and_caching():
|
||
login_resp = client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
|
||
assert login_resp.status_code == 200
|
||
token = login_resp.cookies.get(config.SESSION_COOKIE_NAME)
|
||
auth_headers = {"Cookie": f"{config.SESSION_COOKIE_NAME}={token}"}
|
||
|
||
# 1. Test TTS Status endpoint
|
||
resp = client.get("/api/tts/status", headers=auth_headers)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "available" in data
|
||
assert data["engine"] == "piper-vits"
|
||
voice_ids = [v["id"] for v in data["voices"]]
|
||
assert "en_US-lessac" in voice_ids
|
||
assert "ru_RU-irina" in voice_ids
|
||
assert "en_US-ryan" in voice_ids
|
||
assert "en_GB-alan" in voice_ids
|
||
|
||
# 2. Test TTS synthesis of English word
|
||
tts_resp = client.get("/api/tts?text=Astronomy&lang=en&speed=1.0", headers=auth_headers)
|
||
assert tts_resp.status_code == 200
|
||
assert "audio/wav" in tts_resp.headers.get("content-type", "")
|
||
assert len(tts_resp.content) > 1000
|
||
|
||
# 2b. Test TTS synthesis of Russian phrase with dash and Irina voice
|
||
ru_resp = client.get("/api/tts?text=Важный%20проект%20—%20он%20работает&lang=ru&voice=ru_RU-irina", headers=auth_headers)
|
||
assert ru_resp.status_code == 200
|
||
assert "audio/wav" in ru_resp.headers.get("content-type", "")
|
||
assert len(ru_resp.content) > 1000
|
||
|
||
# 3. Test repeat request returns cached audio
|
||
cached_resp = client.get("/api/tts?text=Astronomy&lang=en&speed=1.0", headers=auth_headers)
|
||
assert cached_resp.status_code == 200
|
||
assert len(cached_resp.content) == len(tts_resp.content)
|
||
|
||
# 4. Test cached audio streaming via /api/audio/{hash}.wav
|
||
from app.tts import get_audio_hash
|
||
h = get_audio_hash("Astronomy", "en", 1.0)
|
||
audio_file_resp = client.get(f"/api/audio/{h}.wav")
|
||
assert audio_file_resp.status_code == 200
|
||
assert "audio/wav" in audio_file_resp.headers.get("content-type", "")
|
||
assert "immutable" in audio_file_resp.headers.get("cache-control", "")
|
||
|
||
# 5. Test card creation attaches audio_url and context_audio_url
|
||
card_resp = client.post(
|
||
"/api/cards",
|
||
json={
|
||
"term": "Galileo",
|
||
"translation": "Галилей",
|
||
"context_sentence": "Galileo pointed his telescope at Jupiter.",
|
||
},
|
||
headers=auth_headers,
|
||
)
|
||
assert card_resp.status_code == 200
|
||
card_data = card_resp.json()
|
||
assert "/api/audio/" in card_data["audio_url"]
|
||
assert "context_audio_url" in card_data
|
||
assert "/api/audio/" in card_data["context_audio_url"]
|
||
|
||
|
||
def test_listening_exercise_and_check():
|
||
login_resp = client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
|
||
assert login_resp.status_code == 200
|
||
token = login_resp.cookies.get(config.SESSION_COOKIE_NAME)
|
||
auth_headers = {"Cookie": f"{config.SESSION_COOKIE_NAME}={token}"}
|
||
|
||
# 1. Fetch listening exercise for tech topic
|
||
resp = client.get("/api/listening/exercise?topic=tech&level=B1", headers=auth_headers)
|
||
assert resp.status_code == 200
|
||
ex = resp.json()
|
||
assert "id" in ex
|
||
assert ex["topic"] == "tech"
|
||
assert "story" in ex
|
||
assert "question" in ex
|
||
assert "/api/audio/" in ex["story_audio_url"]
|
||
assert "/api/audio/" in ex["question_audio_url"]
|
||
# Check that client cannot inspect ideal_answer or keywords before submitting
|
||
assert "ideal_answer" not in ex
|
||
assert "keywords" not in ex
|
||
|
||
# 2. Check evaluation: correct answer for tech-01
|
||
check_resp = client.post(
|
||
"/api/listening/check",
|
||
json={
|
||
"scenario_id": "tech-01",
|
||
"user_answer": "At midnight to minimize downtime",
|
||
},
|
||
headers=auth_headers,
|
||
)
|
||
assert check_resp.status_code == 200
|
||
res = check_resp.json()
|
||
assert res["is_correct"] is True
|
||
assert res["score"] == 100
|
||
assert "midnight" in res["ideal_answer"].lower()
|
||
assert "story" in res
|
||
assert "story_ru" in res
|
||
|
||
# 3. Check evaluation: wrong answer for tech-01
|
||
check_wrong = client.post(
|
||
"/api/listening/check",
|
||
json={
|
||
"scenario_id": "tech-01",
|
||
"user_answer": "Tomorrow morning at sunrise",
|
||
},
|
||
headers=auth_headers,
|
||
)
|
||
assert check_wrong.status_code == 200
|
||
res_wrong = check_wrong.json()
|
||
assert res_wrong["is_correct"] is False
|
||
assert res_wrong["score"] < 50
|
||
|
||
# 4. Check exercise from user's vocabulary cards
|
||
# First ensure a card with context sentence exists
|
||
client.post(
|
||
"/api/cards",
|
||
json={
|
||
"term": "refactoring",
|
||
"translation": "рефакторинг",
|
||
"context_sentence": "Clean architecture requires continuous refactoring of legacy components.",
|
||
},
|
||
headers=auth_headers,
|
||
)
|
||
card_ex_resp = client.get("/api/listening/exercise?topic=cards", headers=auth_headers)
|
||
assert card_ex_resp.status_code == 200
|
||
card_ex = card_ex_resp.json()
|
||
assert card_ex["topic"] == "cards"
|
||
assert "/api/audio/" in card_ex["story_audio_url"]
|
||
|
||
|
||
def test_lessons_presets_crud():
|
||
# 1. Login user
|
||
login_resp = client.post(
|
||
"/api/login",
|
||
json={"username": config.ADMIN_USERNAME, "password": config.ADMIN_PASSWORD},
|
||
)
|
||
assert login_resp.status_code == 200
|
||
token = login_resp.json()["token"]
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
# 2. Get presets
|
||
resp = client.get("/api/lessons/presets", headers=headers)
|
||
assert resp.status_code == 200
|
||
presets = resp.json()["presets"]
|
||
assert len(presets) >= 4
|
||
preset_ids = [p["id"] for p in presets]
|
||
assert "express" in preset_ids
|
||
assert "balanced" in preset_ids
|
||
|
||
# 3. Create custom preset
|
||
new_preset_payload = {
|
||
"name": "Тестовый спринт",
|
||
"icon": "⚡",
|
||
"description": "Быстрый тест",
|
||
"estimated_min": 12,
|
||
"steps": [
|
||
{"id": "step_vocab", "type": "vocab", "enabled": True, "count": 5},
|
||
{"id": "step_ai_debrief", "type": "ai_debrief", "enabled": True},
|
||
],
|
||
}
|
||
save_resp = client.post("/api/lessons/presets", json=new_preset_payload, headers=headers)
|
||
assert save_resp.status_code == 200
|
||
created = save_resp.json()["preset"]
|
||
assert created["name"] == "Тестовый спринт"
|
||
assert created["is_custom"] is True
|
||
custom_id = created["custom_id"]
|
||
|
||
# Verify custom preset appears in list
|
||
resp2 = client.get("/api/lessons/presets", headers=headers)
|
||
assert resp2.status_code == 200
|
||
presets2 = resp2.json()["presets"]
|
||
assert any(p.get("custom_id") == custom_id for p in presets2)
|
||
|
||
# 4. Delete custom preset
|
||
del_resp = client.delete(f"/api/lessons/presets/{custom_id}", headers=headers)
|
||
assert del_resp.status_code == 200
|
||
assert del_resp.json()["deleted"] is True
|
||
|
||
|
||
def test_lessons_grammar_recommendations_and_articles():
|
||
login_resp = client.post(
|
||
"/api/login",
|
||
json={"username": config.ADMIN_USERNAME, "password": config.ADMIN_PASSWORD},
|
||
)
|
||
token = login_resp.json()["token"]
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
# 1. Grammar recommendations
|
||
rec_resp = client.get("/api/lessons/grammar-recommendations", headers=headers)
|
||
assert rec_resp.status_code == 200
|
||
recs = rec_resp.json()
|
||
assert "all_units" in recs
|
||
assert len(recs["all_units"]) >= 15
|
||
assert "weakest_unit" in recs
|
||
assert "previous_unit" in recs
|
||
assert "next_unit" in recs
|
||
|
||
# 2. Articles
|
||
art_resp = client.get("/api/lessons/articles", headers=headers)
|
||
assert art_resp.status_code == 200
|
||
articles = art_resp.json()["articles"]
|
||
assert len(articles) >= 8
|
||
first_art = articles[0]
|
||
assert "title" in first_art
|
||
assert "paragraphs" in first_art
|
||
assert "paragraphs_ru" in first_art
|
||
assert "vocabulary_highlights" in first_art
|
||
|
||
|
||
def test_lessons_debrief_and_history():
|
||
login_resp = client.post(
|
||
"/api/login",
|
||
json={"username": config.ADMIN_USERNAME, "password": config.ADMIN_PASSWORD},
|
||
)
|
||
token = login_resp.json()["token"]
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
# Post session debrief
|
||
debrief_payload = {
|
||
"preset_name": "⚡ Утренний экспресс (10 мин)",
|
||
"duration_seconds": 620,
|
||
"total_score": 85,
|
||
"steps": [
|
||
{"type": "vocab", "title": "Вспоминаем слова", "score": 90, "correct": 9, "total": 10},
|
||
{"type": "grammar_review", "title": "Закрепление грамматики", "score": 80, "correct": 4, "total": 5},
|
||
{"type": "listening", "title": "Аудирование", "score": 100, "correct": 1, "total": 1},
|
||
],
|
||
"errors": [
|
||
{
|
||
"type": "grammar",
|
||
"unit_num": 1,
|
||
"unit_title": "Present Continuous",
|
||
"formula": "S + am/is/are + V-ing",
|
||
"user_answer": "sing",
|
||
"correct_answer": "are singing",
|
||
"explanation": "Слово-маркер Listen указывает на момент речи",
|
||
},
|
||
{
|
||
"type": "vocab",
|
||
"term": "reluctant",
|
||
"translation": "неохотный",
|
||
},
|
||
],
|
||
}
|
||
|
||
debrief_resp = client.post("/api/lessons/debrief", json=debrief_payload, headers=headers)
|
||
assert debrief_resp.status_code == 200
|
||
data = debrief_resp.json()
|
||
assert "session_id" in data
|
||
assert "ai_report" in data
|
||
report = data["ai_report"]
|
||
assert report["grade"] in ["A+", "A", "B", "C"]
|
||
assert "headline" in report
|
||
assert "strengths" in report
|
||
assert "error_patterns" in report
|
||
assert "tomorrow_focus" in report
|
||
|
||
# Check history
|
||
hist_resp = client.get("/api/lessons/history", headers=headers)
|
||
assert hist_resp.status_code == 200
|
||
hist = hist_resp.json()
|
||
assert hist["total_lessons"] >= 1
|
||
assert len(hist["history"]) >= 1
|
||
last_session = hist["history"][0]
|
||
assert last_session["preset_name"] == "⚡ Утренний экспресс (10 мин)"
|
||
assert last_session["errors_count"] == 2
|
||
|
||
|
||
def test_reverse_dictionary_lookup():
|
||
"""
|
||
BUG-06 regression test: Searching Russian words should perform reverse lookup,
|
||
returning the corresponding English term, phonetics, and Russian meaning.
|
||
"""
|
||
from app.auth import _login_failures
|
||
_login_failures.clear()
|
||
login_resp = client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
|
||
assert login_resp.status_code == 200
|
||
token = login_resp.json()["token"]
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
resp = client.get("/api/lookup?q=собака", headers=headers)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["is_reverse"] is True
|
||
assert data["term"].lower() == "dog"
|
||
assert "собака" in data["translation"].lower()
|
||
assert data["phonetic"] != ""
|
||
|
||
|
||
def test_wikipedia_parsoid_html_cleaning():
|
||
"""
|
||
BUG-05 regression test: Wikipedia Parsoid HTML containing data-mw attributes with
|
||
embedded <ref> and literal '>' chars should be cleanly stripped without leaking into text.
|
||
"""
|
||
from app.reader import clean_wiki_html
|
||
|
||
sample_html = (
|
||
'<body>'
|
||
'<span class="mw-empty-elt" about="#mwt1" typeof="mw:Transclusion" '
|
||
'data-mw=\'{"parts":[{"template":{"target":{"wt":"Infobox"},"params":{"title":{"wt":"Test<ref>RefText</ref>"}}}]}\'>'
|
||
'</span>'
|
||
'<p>Linux is a Unix-like operating system.</p>'
|
||
'<table style="font-family:\'Times New Roman\', serif;"><tr><td>Clean Cell</td></tr></table>'
|
||
'</body>'
|
||
)
|
||
chapters, cover, words = clean_wiki_html(sample_html, "Test Article")
|
||
assert len(chapters) >= 1
|
||
content = chapters[0]["content"]
|
||
assert "data-mw" not in content
|
||
assert "{" not in content
|
||
assert "Infobox" not in content
|
||
assert "RefText" not in content
|
||
assert "Linux is a Unix-like operating system." in content
|
||
assert "Times New Roman" not in content # stripped cleanly from table
|
||
assert "Clean Cell" in content
|
||
|
||
|
||
def test_feedback_endpoints_workflow():
|
||
from fastapi.testclient import TestClient
|
||
from pathlib import Path
|
||
|
||
# 1. Unauthenticated access blocked
|
||
unauth_client = TestClient(app)
|
||
resp = unauth_client.get("/api/feedback")
|
||
assert resp.status_code == 401
|
||
|
||
resp = unauth_client.post("/api/feedback", json={"message": "Test bug"})
|
||
assert resp.status_code == 401
|
||
|
||
# 2. Authenticated login
|
||
login_resp = client.post("/api/login", json={"username": "testadmin", "password": "testpass12345"})
|
||
assert login_resp.status_code == 200
|
||
|
||
try:
|
||
# 3. Create a bug report
|
||
bug_payload = {
|
||
"category": "bug",
|
||
"message": "Кнопка в читалке не нажимается на мобилке",
|
||
"context_url": "#reader/book/1/ch/2",
|
||
"context_data": {"screen": "390x844", "device": "Mobile"}
|
||
}
|
||
create_resp = client.post("/api/feedback", json=bug_payload)
|
||
assert create_resp.status_code == 201
|
||
c_data = create_resp.json()
|
||
assert c_data["ok"] is True
|
||
report_id = c_data["id"]
|
||
assert c_data["category"] == "bug"
|
||
assert c_data["status"] == "new"
|
||
|
||
# 4. Create a feature request
|
||
feat_payload = {
|
||
"category": "feature",
|
||
"message": "Добавить темную тему OLED",
|
||
"context_url": "#reader",
|
||
"context_data": {"screen": "1920x1080", "device": "Desktop"}
|
||
}
|
||
create_resp2 = client.post("/api/feedback", json=feat_payload)
|
||
assert create_resp2.status_code == 201
|
||
report_id2 = create_resp2.json()["id"]
|
||
|
||
# 5. List reports
|
||
list_resp = client.get("/api/feedback")
|
||
assert list_resp.status_code == 200
|
||
reports = list_resp.json()
|
||
assert len(reports) >= 2
|
||
r_ids = [r["id"] for r in reports]
|
||
assert report_id in r_ids
|
||
assert report_id2 in r_ids
|
||
|
||
# 6. Filter by category
|
||
bugs_resp = client.get("/api/feedback?category=bug")
|
||
assert bugs_resp.status_code == 200
|
||
assert all(r["category"] == "bug" for r in bugs_resp.json())
|
||
|
||
# 7. Update status to resolved
|
||
patch_resp = client.patch(f"/api/feedback/{report_id}", json={"status": "resolved"})
|
||
assert patch_resp.status_code == 200
|
||
assert patch_resp.json()["status"] == "resolved"
|
||
|
||
# Verify updated
|
||
get_updated = client.get("/api/feedback?status=resolved")
|
||
assert get_updated.status_code == 200
|
||
assert any(r["id"] == report_id and r["status"] == "resolved" for r in get_updated.json())
|
||
|
||
# 8. Delete report
|
||
del_resp = client.delete(f"/api/feedback/{report_id2}")
|
||
assert del_resp.status_code == 200
|
||
assert del_resp.json()["ok"] is True
|
||
|
||
# 9. Verify markdown sync file exists and has content
|
||
reports_md = Path("/home/ai-projects/memory/USER_REPORTS.md")
|
||
assert reports_md.exists()
|
||
content = reports_md.read_text(encoding="utf-8")
|
||
assert "Пользовательские репорты" in content
|
||
assert "Кнопка в читалке не нажимается" in content
|
||
finally:
|
||
try:
|
||
from scripts.sync_reports import sync
|
||
sync()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|
||
|