flow2 / tests /unit /test_dictionary_manager.py
AndrianBalanescu
security+perf: enforce studio auth and cache singletons
1a4837b
Raw History Blame Contribute Delete
4.56 kB
"""Unit tests for the Unified Moldovan Master Dictionary Manager & Studio Endpoints."""
import pytest
import os
import sys
from fastapi.testclient import TestClient
# Add project roots to sys.path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "moldovan-qwen"))
from engine.dictionary_manager import DictionaryManager
from studio_server import app
@pytest.fixture
def dict_manager():
return DictionaryManager()
@pytest.fixture
def client():
import os
headers = {}
if os.environ.get("FLOW_API_KEY"):
headers["Authorization"] = f"Bearer {os.environ['FLOW_API_KEY']}"
return TestClient(app, headers=headers)
def test_dictionary_manager_initialization(dict_manager):
entries = dict_manager.load_all_entries()
assert len(entries) >= 700, f"Expected at least 700 entries, got {len(entries)}"
# Check entry schema
sample = entries[0]
assert "word" in sample
assert "variants" in sample
assert "phonetic" in sample
assert "ro_meaning" in sample
assert "dialect_level" in sample
assert sample["phonetic"].startswith("[") and sample["phonetic"].endswith("]")
def test_dictionary_manager_filtering(dict_manager):
# Test search
res = dict_manager.get_entries(search="harbuz")
assert res["total"] >= 1
assert any(e["word"] == "harbuz" or "harbuz" in e["variants"] for e in res["terms"])
# Test dialect level filter
l4_res = dict_manager.get_entries(dialect_level=4, limit=200)
assert l4_res["total"] > 0
assert all(e["dialect_level"] == 4 for e in l4_res["terms"])
# Test letter filter
h_res = dict_manager.get_entries(letter="H", limit=200)
assert h_res["total"] > 0
assert all(e["word"].upper().startswith("H") for e in h_res["terms"])
def test_dictionary_stats(dict_manager):
stats = dict_manager.get_stats()
assert stats["total_terms"] >= 700
assert stats["phonetic_coverage_pct"] == 100.0
assert stats["with_russian"] > 0
assert stats["with_english"] > 0
assert "1" in stats["levels"]
assert "4" in stats["levels"]
def test_dictionary_crud_cycle(dict_manager):
test_entry = {
"id": "test_cuvant_unic_123",
"word": "ciotkocuvant",
"variants": ["ciotkocuvant", "cetkocuvant"],
"phonetic": "[ciot-ko-cu-VANT]",
"ro_meaning": "cuvant de test foarte bun",
"ru_meaning": "чёткое тестовое слово",
"en_meaning": "awesome test word",
"category": "Argou Urban Chișinău",
"dialect_level": 4
}
# Add
saved = dict_manager.add_entry(test_entry)
assert saved["word"] == "ciotkocuvant"
assert dict_manager.get_entry("test_cuvant_unic_123") is not None
# Update
test_entry["ro_meaning"] = "definitie actualizata"
updated = dict_manager.update_entry("test_cuvant_unic_123", test_entry)
assert updated["ro_meaning"] == "definitie actualizata"
# Delete
deleted = dict_manager.delete_entry("test_cuvant_unic_123")
assert deleted is True
assert dict_manager.get_entry("test_cuvant_unic_123") is None
def test_api_master_dictionary_endpoints(client):
# GET Master list
res = client.get("/api/dictionary/master?limit=10")
assert res.status_code == 200
data = res.json()
assert "terms" in data
assert "total" in data
assert len(data["terms"]) == 10
# GET Stats
stats_res = client.get("/api/dictionary/stats")
assert stats_res.status_code == 200
stats = stats_res.json()
assert stats["total_terms"] >= 700
assert stats["phonetic_coverage_pct"] == 100.0
# POST new entry via API
new_word = {
"word": "zavaliscoterm",
"variants": ["zavaliscoterm"],
"phonetic": "[za-va-lis-co-TERM]",
"ro_meaning": "termen de proba",
"ru_meaning": "пробный термин",
"en_meaning": "sample term",
"category": "Argou Urban Chișinău",
"dialect_level": 4
}
create_res = client.post("/api/dictionary/entry", json=new_word)
assert create_res.status_code == 200
created_data = create_res.json()
entry_id = created_data.get("id") or created_data.get("entry", {}).get("id") or "zavaliscoterm"
# GET created entry
get_res = client.get(f"/api/dictionary/entry/{entry_id}")
assert get_res.status_code == 200
# DELETE created entry
del_res = client.delete(f"/api/dictionary/entry/{entry_id}")
assert del_res.status_code == 200