a94609c07b
- conftest: default SHONAR_TEST_DATABASE_URL is a temp SQLite file, matching the bundled-lite engine; set the env var to a PG URL to exercise that path - models.py: add sqlite_where to the two partial unique indexes — without it SQLite built a FULL unique index on recording_id (WHERE not carried over), wrongly blocking a second export asset per recording - test_m9: pass UUID objects (not str) to direct ORM inserts/gets; SQLite's GUID bind processor rejects strings (asyncpg tolerated them) Verified: pytest -q = 78 passed, 1 skipped; ruff check clean
319 lines
12 KiB
Python
319 lines
12 KiB
Python
"""M9 tests: search, exports, retention sweep.
|
|
|
|
Search runs against the real backend dialect (Postgres in CI/dev, SQLite
|
|
via SHONAR_TEST_DATABASE_URL) — both paths share the endpoint contract.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import uuid
|
|
import zipfile
|
|
|
|
from tests.test_recordings import auth, user_tokens, wav_bytes
|
|
|
|
|
|
async def make_recording(client, token, title, notes=None, tags=None, recorded_at=None):
|
|
h = await auth(token)
|
|
audio = wav_bytes()
|
|
body = {"declared_mime_type": "audio/wav", "declared_size_bytes": len(audio), "title": title}
|
|
r = await client.post("/api/v1/uploads", json=body, headers=h)
|
|
sid = r.json()["id"]
|
|
await client.put(
|
|
f"/api/v1/uploads/{sid}/chunks/0", content=audio,
|
|
headers={**h, "content-type": "application/octet-stream"},
|
|
)
|
|
fin = {"duration_seconds": 5.0}
|
|
if recorded_at:
|
|
fin["recorded_at"] = recorded_at
|
|
if notes:
|
|
fin["notes"] = notes
|
|
r = await client.post(f"/api/v1/uploads/{sid}/finalize", json=fin, headers=h)
|
|
assert r.status_code == 201, r.text
|
|
rec = r.json()
|
|
if tags:
|
|
r = await client.patch(
|
|
f"/api/v1/recordings/{rec['id']}", json={"tags": tags}, headers=h
|
|
)
|
|
assert r.status_code == 200
|
|
return rec["id"]
|
|
|
|
|
|
async def add_transcript(rec_id: str, text: str, segments=None):
|
|
"""Insert a transcript row directly (no AI provider under test)."""
|
|
from shonar.db.models import Transcript
|
|
from shonar.db.session import session_factory
|
|
|
|
async with session_factory()() as s:
|
|
# UUID object (not str): SQLite's GUID bind processor requires it.
|
|
s.add(
|
|
Transcript(
|
|
recording_id=uuid.UUID(rec_id) if isinstance(rec_id, str) else rec_id,
|
|
text=text,
|
|
segments=segments,
|
|
provider="test",
|
|
)
|
|
)
|
|
await s.commit()
|
|
|
|
|
|
async def add_summary(rec_id: str, content: dict):
|
|
from shonar.db.models import Summary
|
|
from shonar.db.session import session_factory
|
|
|
|
async with session_factory()() as s:
|
|
s.add(
|
|
Summary(
|
|
recording_id=uuid.UUID(rec_id) if isinstance(rec_id, str) else rec_id,
|
|
content=content,
|
|
provider="test",
|
|
)
|
|
)
|
|
await s.commit()
|
|
|
|
|
|
# --- search -------------------------------------------------------------------
|
|
|
|
|
|
async def test_search_title_and_transcript(client):
|
|
token = await user_tokens(client, email="m9s1@example.com")
|
|
rid_t = await make_recording(client, token, "Quarterly budget review")
|
|
rid_x = await make_recording(client, token, "Grocery list")
|
|
await add_transcript(rid_x, "remember to buy kale chips and quinoa tonight")
|
|
|
|
r = await client.get("/api/v1/search", params={"q": "budget"}, headers=await auth(token))
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["total"] == 1
|
|
assert body["items"][0]["id"] == rid_t
|
|
assert body["items"][0]["field"] == "title"
|
|
|
|
r = await client.get("/api/v1/search", params={"q": "quinoa"}, headers=await auth(token))
|
|
body = r.json()
|
|
assert body["total"] == 1
|
|
assert body["items"][0]["id"] == rid_x
|
|
assert body["items"][0]["field"] == "transcript"
|
|
assert "quinoa" in body["items"][0]["snippet"].lower()
|
|
|
|
|
|
async def test_search_scope_and_tag(client):
|
|
token = await user_tokens(client, email="m9s2@example.com")
|
|
rid = await make_recording(client, token, "Standup", tags=["daily"])
|
|
await add_transcript(rid, "we discussed the daily standup format")
|
|
|
|
# tag scope finds by tag name
|
|
r = await client.get(
|
|
"/api/v1/search", params={"q": "daily", "scope": "tag"}, headers=await auth(token)
|
|
)
|
|
assert r.json()["total"] == 1
|
|
assert r.json()["items"][0]["field"] == "tag"
|
|
|
|
# a transcript-only word does NOT match under scope=title
|
|
r = await client.get(
|
|
"/api/v1/search", params={"q": "discussed", "scope": "title"}, headers=await auth(token)
|
|
)
|
|
assert r.json()["total"] == 0
|
|
r = await client.get(
|
|
"/api/v1/search", params={"q": "discussed", "scope": "transcript"},
|
|
headers=await auth(token),
|
|
)
|
|
assert r.json()["total"] == 1
|
|
|
|
|
|
async def test_search_isolation_and_deleted(client):
|
|
token_a = await user_tokens(client, email="m9s3a@example.com")
|
|
token_b = await user_tokens(client, email="m9s3b@example.com")
|
|
rid = await make_recording(client, token_a, "secret sauce recipe")
|
|
|
|
h_b = await auth(token_b)
|
|
r = await client.get("/api/v1/search", params={"q": "secret"}, headers=h_b)
|
|
assert r.json()["total"] == 0 # other users' data invisible
|
|
|
|
# soft-deleted rows drop out of search
|
|
await client.delete(f"/api/v1/recordings/{rid}", headers=await auth(token_a))
|
|
r = await client.get(
|
|
"/api/v1/search", params={"q": "secret"}, headers=await auth(token_a)
|
|
)
|
|
assert r.json()["total"] == 0
|
|
|
|
|
|
async def test_search_summary_and_notes(client):
|
|
token = await user_tokens(client, email="m9s4@example.com")
|
|
rid = await make_recording(client, token, "Meeting", notes="bring the projector cable")
|
|
await add_summary(rid, {"short": "sprint retro", "action_items": ["fix flaky test"]})
|
|
|
|
r = await client.get("/api/v1/search", params={"q": "projector"}, headers=await auth(token))
|
|
assert r.json()["total"] == 1
|
|
r = await client.get("/api/v1/search", params={"q": "flaky"}, headers=await auth(token))
|
|
assert r.json()["total"] == 1
|
|
|
|
|
|
# --- list filters ---------------------------------------------------------------
|
|
|
|
|
|
async def test_list_filters_tag_status_date(client):
|
|
token = await user_tokens(client, email="m9f1@example.com")
|
|
await make_recording(client, token, "Old one", tags=["keep"],
|
|
recorded_at="2020-01-01T10:00:00Z")
|
|
rid_new = await make_recording(client, token, "New one", tags=["keep"],
|
|
recorded_at="2026-01-01T10:00:00Z")
|
|
h = await auth(token)
|
|
|
|
r = await client.get("/api/v1/recordings", params={"tag": "keep"}, headers=h)
|
|
assert r.json()["total"] == 2
|
|
r = await client.get(
|
|
"/api/v1/recordings", params={"tag": "keep", "from_date": "2025-06-01T00:00:00Z"},
|
|
headers=h,
|
|
)
|
|
body = r.json()
|
|
assert body["total"] == 1 and body["items"][0]["id"] == rid_new
|
|
|
|
r = await client.get("/api/v1/recordings", params={"status": "ai_disabled"}, headers=h)
|
|
assert r.json()["total"] == 2
|
|
r = await client.get("/api/v1/recordings", params={"status": "bogus"}, headers=h)
|
|
assert r.status_code == 422
|
|
r = await client.get("/api/v1/recordings", params={"tag": "nope"}, headers=h)
|
|
assert r.json()["total"] == 0
|
|
|
|
|
|
# --- exports --------------------------------------------------------------------
|
|
|
|
|
|
async def test_export_formats(client):
|
|
token = await user_tokens(client, email="m9e1@example.com")
|
|
rid = await make_recording(client, token, "Retro & Planning", notes="retro notes here",
|
|
tags=["team"])
|
|
await add_transcript(
|
|
rid, "first segment second segment",
|
|
segments=[{"start": 0.0, "end": 2.5, "text": "first segment", "speaker": None},
|
|
{"start": 2.5, "end": 5.0, "text": "second segment", "speaker": "S1"}],
|
|
)
|
|
await add_summary(rid, {"short": "one line", "action_items": ["do the thing"]})
|
|
h = await auth(token)
|
|
|
|
r = await client.get(f"/api/v1/recordings/{rid}/export", params={"fmt": "txt"}, headers=h)
|
|
assert r.status_code == 200
|
|
assert r.text == "first segment second segment"
|
|
assert "attachment" in r.headers["content-disposition"]
|
|
assert ".txt" in r.headers["content-disposition"]
|
|
|
|
r = await client.get(f"/api/v1/recordings/{rid}/export", params={"fmt": "md"}, headers=h)
|
|
assert r.status_code == 200
|
|
assert "# Retro & Planning" in r.text
|
|
assert "do the thing" in r.text
|
|
assert "`[00:02.5]` **S1**: second segment" in r.text
|
|
assert "`team`" in r.text
|
|
|
|
r = await client.get(f"/api/v1/recordings/{rid}/export", params={"fmt": "zip"}, headers=h)
|
|
assert r.status_code == 200
|
|
assert r.headers["content-type"] == "application/zip"
|
|
with zipfile.ZipFile(io.BytesIO(r.content)) as z:
|
|
names = z.namelist()
|
|
assert "transcript.txt" in names and "notes.md" in names
|
|
assert any(n.endswith(".wav") for n in names)
|
|
|
|
r = await client.get(f"/api/v1/recordings/{rid}/export", params={"fmt": "audio"}, headers=h)
|
|
assert r.status_code == 200
|
|
assert r.content.startswith(b"RIFF")
|
|
|
|
|
|
async def test_export_missing_and_foreign(client):
|
|
token = await user_tokens(client, email="m9e2@example.com")
|
|
rid = await make_recording(client, token, "Bare") # no transcript/summary/notes
|
|
token2 = await user_tokens(client, email="m9e2b@example.com")
|
|
|
|
r = await client.get(f"/api/v1/recordings/{rid}/export", params={"fmt": "txt"},
|
|
headers=await auth(token))
|
|
assert r.status_code == 404 # no transcript yet
|
|
r = await client.get(f"/api/v1/recordings/{rid}/export", params={"fmt": "md"},
|
|
headers=await auth(token))
|
|
assert r.status_code == 404
|
|
# zip still works with just the audio
|
|
r = await client.get(f"/api/v1/recordings/{rid}/export", params={"fmt": "zip"},
|
|
headers=await auth(token))
|
|
assert r.status_code == 200
|
|
|
|
r = await client.get(f"/api/v1/recordings/{rid}/export", params={"fmt": "audio"},
|
|
headers=await auth(token2))
|
|
assert r.status_code == 404 # not yours
|
|
|
|
|
|
# --- retention sweep --------------------------------------------------------------
|
|
|
|
|
|
async def test_retention_sweep_purges_expired(client):
|
|
from datetime import timedelta
|
|
|
|
from shonar.db.models import Asset, Recording, utcnow
|
|
from shonar.db.session import session_factory
|
|
from shonar.services import retention
|
|
|
|
token = await user_tokens(client, email="m9r1@example.com")
|
|
rid = await make_recording(client, token, "Doomed")
|
|
h = await auth(token)
|
|
|
|
# Soft-delete, then push deleted_at past the grace window directly.
|
|
r = await client.delete(f"/api/v1/recordings/{rid}", headers=h)
|
|
assert r.status_code == 204
|
|
async with session_factory()() as s:
|
|
rec = await s.get(Recording, uuid.UUID(rid))
|
|
rec.deleted_at = utcnow() - timedelta(days=31)
|
|
# remember the storage key before the row vanishes
|
|
from shonar.db.models import Asset
|
|
|
|
asset = (
|
|
await s.execute(
|
|
Asset.__table__.select().where(Asset.recording_id == rec.id) # noqa: SLF001
|
|
)
|
|
).first()
|
|
storage_key = asset.storage_key if asset else None
|
|
await s.commit()
|
|
assert storage_key is not None
|
|
|
|
purged = await retention.sweep_deleted()
|
|
assert purged["recordings"] == 1
|
|
assert purged["files"] >= 1
|
|
|
|
async with session_factory()() as s:
|
|
assert await s.get(Recording, uuid.UUID(rid)) is None
|
|
from shonar.storage import get_storage
|
|
|
|
assert not await get_storage().exists(storage_key) # file gone too
|
|
|
|
# Within the window, nothing is purged (cancellable).
|
|
rid2 = await make_recording(client, token, "Fresh delete")
|
|
await client.delete(f"/api/v1/recordings/{rid2}", headers=h)
|
|
purged = await retention.sweep_deleted()
|
|
assert purged["recordings"] == 0
|
|
async with session_factory()() as s:
|
|
assert await s.get(Recording, uuid.UUID(rid2)) is not None
|
|
|
|
|
|
async def test_retention_sweep_account(client):
|
|
from datetime import timedelta
|
|
|
|
from shonar.db.models import Recording, utcnow
|
|
from shonar.db.session import session_factory
|
|
from shonar.services import retention
|
|
|
|
token = await user_tokens(client, email="m9r2@example.com")
|
|
rid = await make_recording(client, token, "Gone with user")
|
|
|
|
async with session_factory()() as s:
|
|
user = await s.scalar(select_user("m9r2@example.com"))
|
|
user.deleted_at = utcnow() - timedelta(days=31)
|
|
await s.commit()
|
|
|
|
purged = await retention.sweep_deleted()
|
|
assert purged["users"] == 1
|
|
async with session_factory()() as s:
|
|
assert await s.get(Recording, uuid.UUID(rid)) is None # cascade
|
|
assert await s.scalar(select_user("m9r2@example.com")) is None
|
|
|
|
|
|
def select_user(email: str):
|
|
from sqlalchemy import select
|
|
|
|
from shonar.db.models import User
|
|
|
|
return select(User).where(User.email == email)
|