507 lines
17 KiB
Python
507 lines
17 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"
|
||
|
||
|
||
|