devplacepy/tests/api/tools/deepsearch/history.py

# retoor <retoor@molodetz.nl>

import time
import pytest
import requests
from tests.conftest import BASE_URL
from devplacepy.database import (
    create_deepsearch_session,
    get_table,
    refresh_snapshot,
    set_setting,
    update_deepsearch_session,
)
from devplacepy.services.jobs import queue

JSON = {"Accept": "application/json"}
_counter_dsh = [0]


@pytest.fixture(scope="module", autouse=True)
def _settings(app_server):
    set_setting("rate_limit_per_minute", "1000000")
    set_setting("registration_open", "1")
    yield


def _user():
    _counter_dsh[0] += 1
    name = f"dsh{int(time.time() * 1000)}{_counter_dsh[0]}"
    requests.post(
        f"{BASE_URL}/auth/signup",
        data={
            "username": name,
            "email": f"{name}@t.dev",
            "password": "secret123",
            "confirm_password": "secret123",
            "birth_date": "1990-01-01",
            "accept_terms": "1",
        },
        allow_redirects=True,
    )
    refresh_snapshot()
    user = get_table("users").find_one(username=name)
    return user["uid"], user["api_key"]


def _seed_session(owner_id, status="done", keep_job=True, query="a research question"):
    uid = queue.enqueue(
        "deepsearch",
        {"query": query, "depth": 1, "max_pages": 5},
        "user",
        owner_id,
        f"DeepSearch: {query}",
    )
    create_deepsearch_session(
        uid, "user", owner_id, query, 1, 5, f"ds_{uid.replace('-', '')}"
    )
    update_deepsearch_session(
        uid,
        {
            "status": status,
            "score": 80,
            "confidence": 0.7,
            "source_diversity": 0.6,
            "page_count": 5,
            "chunk_count": 40,
            "summary": "A short summary of findings.",
        },
    )
    if not keep_job:
        get_table("jobs").delete(uid=uid)
    refresh_snapshot()
    return uid


def test_history_lists_own_sessions_newest_first(app_server):
    owner_uid, key = _user()
    first = _seed_session(owner_uid, query="first question")
    time.sleep(0.01)
    second = _seed_session(owner_uid, query="second question")
    r = requests.get(
        f"{BASE_URL}/tools/deepsearch/history",
        headers={"X-API-KEY": key, **JSON},
    )
    assert r.status_code == 200, r.text
    body = r.json()
    uids = [s["uid"] for s in body["sessions"]]
    assert uids.index(second) < uids.index(first)


def test_history_item_shape_for_done_session(app_server):
    owner_uid, key = _user()
    uid = _seed_session(owner_uid)
    r = requests.get(
        f"{BASE_URL}/tools/deepsearch/history",
        headers={"X-API-KEY": key, **JSON},
    )
    assert r.status_code == 200, r.text
    items = {s["uid"]: s for s in r.json()["sessions"]}
    item = items[uid]
    assert item["query"] == "a research question"
    assert item["status"] == "done"
    assert item["score"] == 80
    assert item["reopen_url"] == f"/tools/deepsearch/{uid}/session"
    assert item["chat_available"] is True
    assert item["available"] is True


def test_history_marks_expired_session_unavailable(app_server):
    owner_uid, key = _user()
    uid = _seed_session(owner_uid, keep_job=False)
    r = requests.get(
        f"{BASE_URL}/tools/deepsearch/history",
        headers={"X-API-KEY": key, **JSON},
    )
    assert r.status_code == 200, r.text
    items = {s["uid"]: s for s in r.json()["sessions"]}
    item = items[uid]
    assert item["available"] is False
    assert item["chat_available"] is False


def test_history_is_scoped_to_the_owner(app_server):
    owner_uid, key = _user()
    other_uid, other_key = _user()
    mine = _seed_session(owner_uid, query="mine only")
    _seed_session(other_uid, query="not mine")
    r = requests.get(
        f"{BASE_URL}/tools/deepsearch/history",
        headers={"X-API-KEY": key, **JSON},
    )
    assert r.status_code == 200, r.text
    uids = [s["uid"] for s in r.json()["sessions"]]
    assert mine in uids
    r_other = requests.get(
        f"{BASE_URL}/tools/deepsearch/history",
        headers={"X-API-KEY": other_key, **JSON},
    )
    other_uids = [s["uid"] for s in r_other.json()["sessions"]]
    assert mine not in other_uids


def test_history_scopes_guests_by_ip(app_server):
    uid = queue.enqueue(
        "deepsearch",
        {"query": "guest question", "depth": 1, "max_pages": 5},
        "guest",
        "203.0.113.55",
        "DeepSearch: guest question",
    )
    create_deepsearch_session(
        uid, "guest", "203.0.113.55", "guest question", 1, 5, f"ds_{uid.replace('-', '')}"
    )
    refresh_snapshot()
    r = requests.get(
        f"{BASE_URL}/tools/deepsearch/history",
        headers={"X-Real-IP": "203.0.113.55", **JSON},
    )
    assert r.status_code == 200, r.text
    uids = [s["uid"] for s in r.json()["sessions"]]
    assert uid in uids
    r_other_ip = requests.get(
        f"{BASE_URL}/tools/deepsearch/history",
        headers={"X-Real-IP": "203.0.113.99", **JSON},
    )
    other_uids = [s["uid"] for s in r_other_ip.json()["sessions"]]
    assert uid not in other_uids


def test_history_renders_html(app_server):
    r = requests.get(f"{BASE_URL}/tools/deepsearch/history")
    assert r.status_code == 200
    assert "DeepSearch history" in r.text