devplacepy/session.py at master

 # retoor <retoor@molodetz.nl>
 import json
 import requests
 from tests.conftest import BASE_URL
 from devplacepy.database import (
     create_deepsearch_session,
     get_table,
     refresh_snapshot,
 )
 from devplacepy.services.jobs import queue
 def _json_headers():
     return {"Accept": "application/json"}
 def _seed_done_session(owner_id="ds-session-owner"):
     uid = queue.enqueue(
         "deepsearch",
         {"query": "the question", "depth": 2, "max_pages": 10},
         "user",
         owner_id,
         "DeepSearch: the question",
     )
     create_deepsearch_session(
         uid, "user", owner_id, "the question", 2, 10, f"ds_{uid.replace('-', '')}"
     )
     report = {
         "query": "the question",
         "summary": "A grounded summary.",
         "findings": [
             {"title": "Finding one", "detail": "Detail.", "confidence": 0.8, "citations": [1]}
         ],
         "sources": [{"url": "https://example.com", "title": "Example", "source": "httpx"}],
         "score": 70,
         "confidence": 0.7,
         "source_diversity": 0.5,
         "page_count": 3,
         "chunk_count": 12,
     }
     result = {
         "query": "the question",
         "score": 70,
         "confidence": 0.7,
         "source_diversity": 0.5,
         "page_count": 3,
         "chunk_count": 12,
         "report": report,
     }
     get_table("jobs").update(
         {"uid": uid, "status": queue.DONE, "result": json.dumps(result)}, ["uid"]
     )
     get_table("deepsearch_sessions").update(
         {
             "uid": uid,
             "status": "done",
             "score": 70,
             "confidence": 0.7,
             "source_diversity": 0.5,
             "page_count": 3,
             "chunk_count": 12,
             "summary": "A grounded summary.",
         },
         ["uid"],
     )
     refresh_snapshot()
     return uid
 def _clear():
     refresh_snapshot()
     jobs = get_table("jobs")
     for row in list(jobs.find(kind="deepsearch")):
         jobs.delete(uid=row["uid"])
     sessions = get_table("deepsearch_sessions")
     for row in list(sessions.find()):
         sessions.delete(uid=row["uid"])
 def test_session_json_shape(app_server):
     try:
         uid = _seed_done_session()
         r = requests.get(
             f"{BASE_URL}/tools/deepsearch/{uid}/session", headers=_json_headers()
         )
         assert r.status_code == 200, r.text
         body = r.json()
         assert body["uid"] == uid
         assert body["status"] == "done"
         assert body["query"] == "the question"
         assert body["score"] == 70
         assert body["chat_ws_url"] == f"/tools/deepsearch/{uid}/chat"
         assert body["export_md_url"] == f"/tools/deepsearch/{uid}/export.md"
         assert body["findings"]
         assert body["sources"]
         assert "viewer_is_admin" in body
         assert "viewer_owns" in body
     finally:
         _clear()
 def test_session_html_renders_without_jinja_global_collision(app_server):
     try:
         uid = _seed_done_session()
         r = requests.get(f"{BASE_URL}/tools/deepsearch/{uid}/session")
         assert r.status_code == 200, r.text
         assert "the question" in r.text
         assert "dp-deepsearch-chat" in r.text
     finally:
         _clear()
 def test_session_reads_disk_report_before_result_commit(app_server):
     from pathlib import Path
     import shutil
     from devplacepy.config import DEEPSEARCH_DIR
     owner_id = "ds-race-owner"
     uid = queue.enqueue(
         "deepsearch",
         {"query": "race question", "depth": 2, "max_pages": 10},
         "user",
         owner_id,
         "DeepSearch: race question",
     )
     create_deepsearch_session(
         uid, "user", owner_id, "race question", 2, 10, f"ds_{uid.replace('-', '')}"
     )
     report = {
         "query": "race question",
         "summary": "A grounded summary from disk.",
         "findings": [
             {"title": "Disk finding", "detail": "D.", "confidence": 0.8, "citations": [1]}
         ],
         "sources": [{"url": "https://example.com", "title": "Example", "source": "httpx"}],
         "score": 84,
         "confidence": 0.85,
         "source_diversity": 0.75,
         "synthesis": "agents",
         "page_count": 12,
         "chunk_count": 61,
     }
     session_dir = DEEPSEARCH_DIR / uid
     session_dir.mkdir(parents=True, exist_ok=True)
     (session_dir / "report.json").write_text(json.dumps(report), encoding="utf-8")
     get_table("deepsearch_sessions").update(
         {"uid": uid, "status": "done"}, ["uid"]
     )
     refresh_snapshot()
     try:
         r = requests.get(
             f"{BASE_URL}/tools/deepsearch/{uid}/session", headers=_json_headers()
         )
         assert r.status_code == 200, r.text
         body = r.json()
         assert body["status"] == "done"
         assert body["score"] == 84
         assert body["chunk_count"] == 61
         assert body["findings"]
         assert body["sources"]
         assert body["chat_ws_url"] == f"/tools/deepsearch/{uid}/chat"
         md = requests.get(f"{BASE_URL}/tools/deepsearch/{uid}/export.md")
         assert md.status_code == 200, md.text
         assert "Disk finding" in md.text
     finally:
         shutil.rmtree(session_dir, ignore_errors=True)
         _clear()
 def test_session_unknown_uid_404(app_server):
     r = requests.get(
         f"{BASE_URL}/tools/deepsearch/nope/session", headers=_json_headers()
     )
     assert r.status_code == 404
 def test_export_markdown(app_server):
     try:
         uid = _seed_done_session()
         r = requests.get(f"{BASE_URL}/tools/deepsearch/{uid}/export.md")
         assert r.status_code == 200, r.text
         assert "DeepSearch report" in r.text
         assert "Finding one" in r.text
     finally:
         _clear()
 def test_export_json(app_server):
     try:
         uid = _seed_done_session()
         r = requests.get(f"{BASE_URL}/tools/deepsearch/{uid}/export.json")
         assert r.status_code == 200, r.text
         body = r.json()
         assert body["query"] == "the question"
         assert body["findings"]
     finally:
         _clear()
 def test_export_unknown_uid_404(app_server):
     r = requests.get(f"{BASE_URL}/tools/deepsearch/nope/export.md")
     assert r.status_code == 404