464 lines
17 KiB
Python
464 lines
17 KiB
Python
import os
|
|
import io
|
|
import uuid
|
|
from fastapi.testclient import TestClient
|
|
import pytest
|
|
|
|
os.environ.setdefault("SECRET_KEY", "x" * 32)
|
|
os.environ.setdefault("DATABASE_URL", "sqlite:////tmp/delphi_test.sqlite")
|
|
|
|
from app.main import app # noqa: E402
|
|
from app.auth.security import get_current_user # noqa: E402
|
|
|
|
|
|
class _User:
|
|
def __init__(self):
|
|
self.id = 1
|
|
self.username = "tester"
|
|
self.is_admin = True
|
|
self.is_active = True
|
|
|
|
|
|
@pytest.fixture()
|
|
def client():
|
|
app.dependency_overrides[get_current_user] = lambda: _User()
|
|
try:
|
|
yield TestClient(app)
|
|
finally:
|
|
app.dependency_overrides.pop(get_current_user, None)
|
|
|
|
|
|
def _dummy_docx_bytes():
|
|
# Minimal docx that docxtpl can open. To avoid binary template creation,
|
|
# we use a pre-generated minimal DOCX header stored as bytes.
|
|
# Fallback: create in-memory empty docx using python-docx if available.
|
|
try:
|
|
from docx import Document
|
|
except Exception:
|
|
return b"PK\x03\x04" # still accepted and stored; preview will not render
|
|
d = Document()
|
|
p = d.add_paragraph()
|
|
p.add_run("Hello ")
|
|
p.add_run("{{CLIENT_NAME}}")
|
|
buf = io.BytesIO()
|
|
d.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
def test_upload_search_get_versions_and_preview(client: TestClient):
|
|
# Upload a DOCX template
|
|
payload = {
|
|
"name": "Engagement Letter",
|
|
"category": "GENERAL",
|
|
"description": "Test template",
|
|
"semantic_version": "1.0.0",
|
|
}
|
|
files = {
|
|
"file": ("letter.docx", _dummy_docx_bytes(), "application/vnd.openxmlformats-officedocument.wordprocessingml.document"),
|
|
}
|
|
resp = client.post("/api/templates/upload", data=payload, files=files)
|
|
assert resp.status_code == 200, resp.text
|
|
tpl = resp.json()
|
|
tpl_id = tpl["id"]
|
|
|
|
# Search
|
|
resp = client.get("/api/templates/search?q=Engagement")
|
|
assert resp.status_code == 200
|
|
assert any(item["id"] == tpl_id for item in resp.json())
|
|
|
|
# Also match by description via q
|
|
resp = client.get("/api/templates/search?q=template")
|
|
assert resp.status_code == 200
|
|
ids = {item["id"] for item in resp.json()}
|
|
assert tpl_id in ids
|
|
|
|
# Get template
|
|
resp = client.get(f"/api/templates/{tpl_id}")
|
|
assert resp.status_code == 200
|
|
|
|
# List versions
|
|
resp = client.get(f"/api/templates/{tpl_id}/versions")
|
|
assert resp.status_code == 200
|
|
versions = resp.json()
|
|
assert len(versions) >= 1
|
|
vid = versions[0]["id"]
|
|
|
|
# Preview with context that resolves CLIENT_NAME
|
|
resp = client.post(
|
|
f"/api/templates/{tpl_id}/preview",
|
|
json={"context": {"CLIENT_NAME": "Alice"}, "version_id": vid},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert "resolved" in body and body["resolved"].get("CLIENT_NAME") == "Alice"
|
|
assert isinstance(body.get("unresolved", []), list)
|
|
assert body.get("output_size", 0) >= 0
|
|
|
|
|
|
def _docx_with_tokens(text: str) -> bytes:
|
|
try:
|
|
from docx import Document
|
|
except Exception:
|
|
return b"PK\x03\x04"
|
|
d = Document()
|
|
d.add_paragraph(text)
|
|
buf = io.BytesIO()
|
|
d.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
def test_add_version_and_form_variable_resolution(client: TestClient):
|
|
# Upload initial template
|
|
files = {"file": ("vars.docx", _docx_with_tokens("{{OFFICE_NAME}}"), "application/vnd.openxmlformats-officedocument.wordprocessingml.document")}
|
|
resp = client.post(
|
|
"/api/templates/upload",
|
|
data={"name": "VarsTpl", "semantic_version": "1.0.0"},
|
|
files=files,
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
tpl_id = resp.json()["id"]
|
|
|
|
# Add a new version
|
|
files2 = {"file": ("vars2.docx", _docx_with_tokens("{{OFFICE_NAME}} {{NEW_FIELD}}"), "application/vnd.openxmlformats-officedocument.wordprocessingml.document")}
|
|
resp = client.post(
|
|
f"/api/templates/{tpl_id}/versions",
|
|
data={"semantic_version": "1.1.0", "approve": True},
|
|
files=files2,
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
# Insert a FormVariable directly
|
|
from app.database.base import SessionLocal
|
|
from app.models.additional import FormVariable
|
|
db = SessionLocal()
|
|
try:
|
|
db.merge(FormVariable(identifier="OFFICE_NAME", query="static", response="Delphi", active=1))
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
# Preview without explicit context should resolve OFFICE_NAME from FormVariable
|
|
resp = client.post(f"/api/templates/{tpl_id}/preview", json={"context": {}})
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["resolved"].get("OFFICE_NAME") == "Delphi"
|
|
|
|
|
|
def _upload_template(client: TestClient, name: str, category: str = "GENERAL") -> int:
|
|
files = {
|
|
"file": ("t.docx", _dummy_docx_bytes(), "application/vnd.openxmlformats-officedocument.wordprocessingml.document"),
|
|
}
|
|
resp = client.post(
|
|
"/api/templates/upload",
|
|
data={"name": name, "category": category, "semantic_version": "1.0.0"},
|
|
files=files,
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
return resp.json()["id"]
|
|
|
|
|
|
def _add_keywords(client: TestClient, template_id: int, keywords):
|
|
resp = client.post(f"/api/templates/{template_id}/keywords", json={"keywords": list(keywords)})
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
|
|
def test_templates_search_keywords_mode_any_and_all(client: TestClient):
|
|
# Create three templates with overlapping keywords
|
|
t1 = _upload_template(client, "Template A")
|
|
t2 = _upload_template(client, "Template B")
|
|
t3 = _upload_template(client, "Template C")
|
|
|
|
_add_keywords(client, t1, ["divorce", "qdro"]) # both
|
|
_add_keywords(client, t2, ["divorce"]) # only divorce
|
|
_add_keywords(client, t3, ["qdro", "pension"]) # qdro + other
|
|
|
|
# ANY (default)
|
|
resp = client.get(
|
|
"/api/templates/search",
|
|
params=[("keywords", "divorce"), ("keywords", "qdro")],
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
ids = {item["id"] for item in resp.json()}
|
|
assert {t1, t2, t3}.issubset(ids) # all three should appear
|
|
|
|
# ANY (explicit)
|
|
resp = client.get(
|
|
"/api/templates/search",
|
|
params=[("keywords", "divorce"), ("keywords", "qdro"), ("keywords_mode", "any")],
|
|
)
|
|
assert resp.status_code == 200
|
|
ids = {item["id"] for item in resp.json()}
|
|
assert {t1, t2, t3}.issubset(ids)
|
|
|
|
# ALL - must contain both divorce AND qdro
|
|
resp = client.get(
|
|
"/api/templates/search",
|
|
params=[("keywords", "divorce"), ("keywords", "qdro"), ("keywords_mode", "all")],
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
ids = {item["id"] for item in resp.json()}
|
|
assert ids == {t1}
|
|
|
|
|
|
def test_templates_search_pagination_and_sorting(client: TestClient):
|
|
# Ensure clean state for this test
|
|
# Upload multiple templates with different names and categories
|
|
ids = []
|
|
ids.append(_upload_template(client, "Alpha", category="CAT2"))
|
|
ids.append(_upload_template(client, "Charlie", category="CAT1"))
|
|
ids.append(_upload_template(client, "Bravo", category="CAT1"))
|
|
ids.append(_upload_template(client, "Echo", category="CAT3"))
|
|
ids.append(_upload_template(client, "Delta", category="CAT2"))
|
|
|
|
# Sort by name asc, limit 2
|
|
resp = client.get(
|
|
"/api/templates/search",
|
|
params={"sort_by": "name", "sort_dir": "asc", "limit": 2},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
names = [item["name"] for item in resp.json()]
|
|
assert names == sorted(names) # asc
|
|
assert len(names) == 2
|
|
|
|
# Sort by name desc, skip 1, limit 3
|
|
resp = client.get(
|
|
"/api/templates/search",
|
|
params={"sort_by": "name", "sort_dir": "desc", "skip": 1, "limit": 3},
|
|
)
|
|
assert resp.status_code == 200
|
|
names_desc = [item["name"] for item in resp.json()]
|
|
assert len(names_desc) == 3
|
|
assert names_desc == sorted(names_desc, reverse=True)
|
|
|
|
# Sort by category asc (ties unresolved by name asc implicitly by DB)
|
|
resp = client.get(
|
|
"/api/templates/search",
|
|
params={"sort_by": "category", "sort_dir": "asc"},
|
|
)
|
|
assert resp.status_code == 200
|
|
categories = [item["category"] for item in resp.json()]
|
|
assert categories == sorted(categories)
|
|
|
|
# Sort by updated desc
|
|
resp = client.get(
|
|
"/api/templates/search",
|
|
params={"sort_by": "updated", "sort_dir": "desc"},
|
|
)
|
|
assert resp.status_code == 200
|
|
# We can't assert exact order of timestamps easily; just ensure we got results
|
|
assert isinstance(resp.json(), list) and len(resp.json()) >= 5
|
|
|
|
# include_total=true returns object with items and total
|
|
resp = client.get(
|
|
"/api/templates/search",
|
|
params={"sort_by": "name", "sort_dir": "asc", "include_total": True},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert isinstance(data, dict) and "items" in data and "total" in data
|
|
assert isinstance(data["items"], list) and isinstance(data["total"], int)
|
|
|
|
|
|
def test_templates_search_active_filtering(client: TestClient):
|
|
# Create two templates, mark one inactive directly
|
|
tid_active = _upload_template(client, "Active T")
|
|
tid_inactive = _upload_template(client, "Inactive T")
|
|
|
|
# Mark second as inactive via direct DB update
|
|
from app.database.base import SessionLocal
|
|
from app.models.templates import DocumentTemplate
|
|
db = SessionLocal()
|
|
try:
|
|
tpl = db.query(DocumentTemplate).filter(DocumentTemplate.id == tid_inactive).first()
|
|
tpl.active = False
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
# Default active_only=true should return only the active one
|
|
resp = client.get("/api/templates/search")
|
|
assert resp.status_code == 200
|
|
ids = {item["id"] for item in resp.json()}
|
|
assert tid_active in ids
|
|
assert tid_inactive not in ids
|
|
|
|
# active_only=false should return both
|
|
resp = client.get("/api/templates/search", params={"active_only": False})
|
|
assert resp.status_code == 200
|
|
ids2 = {item["id"] for item in resp.json()}
|
|
assert tid_active in ids2 and tid_inactive in ids2
|
|
|
|
|
|
def test_templates_search_category_multi_repeat_and_csv(client: TestClient):
|
|
# Upload templates across multiple categories
|
|
t_cat1_a = _upload_template(client, "C1-A", category="CAT1")
|
|
t_cat1_b = _upload_template(client, "C1-B", category="CAT1")
|
|
t_cat2 = _upload_template(client, "C2-A", category="CAT2")
|
|
t_cat3 = _upload_template(client, "C3-A", category="CAT3")
|
|
|
|
# Repeatable category parameters (?category=CAT1&category=CAT3)
|
|
resp = client.get(
|
|
"/api/templates/search",
|
|
params=[("category", "CAT1"), ("category", "CAT3")],
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
ids = {item["id"] for item in resp.json()}
|
|
# Expect only CAT1 and CAT3 templates
|
|
assert t_cat1_a in ids and t_cat1_b in ids and t_cat3 in ids
|
|
assert t_cat2 not in ids
|
|
|
|
# CSV within a single category parameter (?category=CAT2,CAT3)
|
|
resp = client.get(
|
|
"/api/templates/search",
|
|
params={"category": "CAT2,CAT3"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
ids_csv = {item["id"] for item in resp.json()}
|
|
assert t_cat2 in ids_csv and t_cat3 in ids_csv
|
|
assert t_cat1_a not in ids_csv and t_cat1_b not in ids_csv
|
|
|
|
# Unknown category should return empty set when exclusive
|
|
resp = client.get(
|
|
"/api/templates/search",
|
|
params={"category": "NON_EXISTENT"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
|
|
def test_templates_search_has_keywords_filter(client: TestClient):
|
|
# Create two templates
|
|
t1 = _upload_template(client, "HasKW")
|
|
t2 = _upload_template(client, "NoKW")
|
|
|
|
# Add keywords only to t1
|
|
_add_keywords(client, t1, ["alpha", "beta"])
|
|
|
|
# has_keywords=true should include only t1
|
|
resp = client.get("/api/templates/search", params={"has_keywords": True})
|
|
assert resp.status_code == 200, resp.text
|
|
ids_true = {item["id"] for item in resp.json()}
|
|
assert t1 in ids_true and t2 not in ids_true
|
|
|
|
# has_keywords=false should include only t2
|
|
resp = client.get("/api/templates/search", params={"has_keywords": False})
|
|
assert resp.status_code == 200
|
|
ids_false = {item["id"] for item in resp.json()}
|
|
assert t2 in ids_false and t1 not in ids_false
|
|
|
|
|
|
def test_templates_categories_listing(client: TestClient):
|
|
# Empty DB categories
|
|
resp = client.get("/api/templates/categories")
|
|
assert resp.status_code == 200
|
|
empty = resp.json()
|
|
# May contain defaults from previous tests; ensure it's a list
|
|
assert isinstance(empty, list)
|
|
|
|
# Create active/inactive across categories
|
|
t1 = _upload_template(client, "K-A1", category="K1")
|
|
t2 = _upload_template(client, "K-A2", category="K1")
|
|
t3 = _upload_template(client, "K-B1", category="K2")
|
|
|
|
# Inactivate one of K1
|
|
from app.database.base import SessionLocal
|
|
from app.models.templates import DocumentTemplate
|
|
db = SessionLocal()
|
|
try:
|
|
tpl = db.query(DocumentTemplate).filter(DocumentTemplate.id == t2).first()
|
|
tpl.active = False
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
# active_only=true (default) should count only active: K1:1, K2:1
|
|
resp = client.get("/api/templates/categories")
|
|
assert resp.status_code == 200
|
|
rows = resp.json()
|
|
by_cat = {r["category"]: r["count"] for r in rows}
|
|
assert by_cat.get("K1", 0) >= 1
|
|
assert by_cat.get("K2", 0) >= 1
|
|
|
|
# active_only=false should count both entries in K1
|
|
resp = client.get("/api/templates/categories", params={"active_only": False})
|
|
assert resp.status_code == 200
|
|
rows_all = resp.json()
|
|
by_cat_all = {r["category"]: r["count"] for r in rows_all}
|
|
|
|
# include_total=true shape
|
|
resp = client.get("/api/templates/categories", params={"include_total": True, "active_only": False})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert isinstance(data, dict) and "items" in data and "total" in data
|
|
assert isinstance(data["items"], list) and isinstance(data["total"], int)
|
|
assert by_cat_all.get("K1", 0) >= 2
|
|
assert by_cat_all.get("K2", 0) >= 1
|
|
|
|
|
|
|
|
def test_templates_download_current_version(client: TestClient):
|
|
# Upload a DOCX template
|
|
payload = {
|
|
"name": f"Download Letter {uuid.uuid4().hex[:8]}",
|
|
"category": "GENERAL",
|
|
"description": "Download test",
|
|
"semantic_version": "1.0.0",
|
|
}
|
|
filename = "letter.docx"
|
|
content_type = "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
|
|
data_bytes = _dummy_docx_bytes()
|
|
files = {
|
|
"file": (filename, data_bytes, content_type),
|
|
}
|
|
resp = client.post("/api/templates/upload", data=payload, files=files)
|
|
assert resp.status_code == 200, resp.text
|
|
tpl_id = resp.json()["id"]
|
|
|
|
# Download current approved version
|
|
resp = client.get(f"/api/templates/{tpl_id}/download")
|
|
assert resp.status_code == 200, resp.text
|
|
# Verify headers
|
|
assert resp.headers.get("content-type") == content_type
|
|
cd = resp.headers.get("content-disposition", "")
|
|
assert "attachment;" in cd and filename in cd
|
|
# Body should be non-empty and equal to uploaded bytes
|
|
assert resp.content == data_bytes
|
|
|
|
|
|
def test_templates_download_specific_version_by_id(client: TestClient):
|
|
# Upload initial version
|
|
files_v1 = {"file": ("v1.docx", _docx_with_tokens("V1"), "application/vnd.openxmlformats-officedocument.wordprocessingml.document")}
|
|
resp = client.post(
|
|
"/api/templates/upload",
|
|
data={"name": f"MultiVersion {uuid.uuid4().hex[:8]}", "semantic_version": "1.0.0"},
|
|
files=files_v1,
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
tpl_id = resp.json()["id"]
|
|
|
|
# Add a second version (do not approve so current stays V1)
|
|
v2_bytes = _docx_with_tokens("V2 unique")
|
|
files_v2 = {"file": ("v2.docx", v2_bytes, "application/vnd.openxmlformats-officedocument.wordprocessingml.document")}
|
|
resp2 = client.post(
|
|
f"/api/templates/{tpl_id}/versions",
|
|
data={"semantic_version": "1.1.0", "approve": False},
|
|
files=files_v2,
|
|
)
|
|
assert resp2.status_code == 200, resp2.text
|
|
|
|
# Find versions to get v2 id
|
|
resp_list = client.get(f"/api/templates/{tpl_id}/versions")
|
|
assert resp_list.status_code == 200
|
|
versions = resp_list.json()
|
|
# v2 should be in list; grab the one with semantic_version 1.1.0
|
|
v2 = next(v for v in versions if v["semantic_version"] == "1.1.0")
|
|
v2_id = v2["id"]
|
|
|
|
# Download specifically v2
|
|
resp_dl = client.get(f"/api/templates/{tpl_id}/download", params={"version_id": v2_id})
|
|
assert resp_dl.status_code == 200, resp_dl.text
|
|
assert resp_dl.headers.get("content-type") == "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
|
|
cd2 = resp_dl.headers.get("content-disposition", "")
|
|
assert "v2.docx" in cd2
|
|
assert resp_dl.content == v2_bytes
|