|
# 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,
|
|
)
|
|
from devplacepy.services.jobs import queue
|
|
|
|
JSON = {"Accept": "application/json"}
|
|
_counter = [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[0] += 1
|
|
name = f"dsl{int(time.time() * 1000)}{_counter[0]}"
|
|
requests.post(
|
|
f"{BASE_URL}/auth/signup",
|
|
data={
|
|
"username": name,
|
|
"email": f"{name}@t.dev",
|
|
"password": "secret123",
|
|
"confirm_password": "secret123",
|
|
},
|
|
allow_redirects=True,
|
|
)
|
|
refresh_snapshot()
|
|
user = get_table("users").find_one(username=name)
|
|
return user["uid"], user["api_key"]
|
|
|
|
|
|
def _seed_session(owner_id, query="test query"):
|
|
uid = queue.enqueue(
|
|
"deepsearch",
|
|
{"query": query, "depth": 1, "max_pages": 5},
|
|
"user",
|
|
owner_id,
|
|
"DeepSearch",
|
|
)
|
|
create_deepsearch_session(
|
|
uid, "user", owner_id, query, 1, 5, f"ds_{uid.replace('-', '')}"
|
|
)
|
|
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_list_returns_own_sessions(app_server):
|
|
try:
|
|
owner_uid, api_key = _user()
|
|
uid = _seed_session(owner_uid, query="first research")
|
|
headers = {"X-API-KEY": api_key, **JSON}
|
|
r = requests.get(f"{BASE_URL}/tools/deepsearch/list", headers=headers)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert "sessions" in body
|
|
assert len(body["sessions"]) == 1
|
|
assert body["sessions"][0]["uid"] == uid
|
|
assert body["sessions"][0]["query"] == "first research"
|
|
assert body["sessions"][0]["status"] == "pending"
|
|
assert "created_at" in body["sessions"][0]
|
|
finally:
|
|
_clear()
|
|
|
|
|
|
def test_list_multiple_sessions_sorted_newest_first(app_server):
|
|
try:
|
|
owner_uid, api_key = _user()
|
|
uids = []
|
|
for i in range(3):
|
|
uids.append(_seed_session(owner_uid, query=f"research {i}"))
|
|
headers = {"X-API-KEY": api_key, **JSON}
|
|
r = requests.get(f"{BASE_URL}/tools/deepsearch/list", headers=headers)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert len(body["sessions"]) == 3
|
|
uids_in_response = [s["uid"] for s in body["sessions"]]
|
|
assert sorted(uids_in_response, reverse=True) == uids_in_response
|
|
actual_queries = [s["query"] for s in body["sessions"]]
|
|
assert len(set(actual_queries)) == 3
|
|
finally:
|
|
_clear()
|
|
|
|
|
|
def test_list_does_not_include_other_users_sessions(app_server):
|
|
try:
|
|
owner_uid, api_key = _user()
|
|
_seed_session(owner_uid, query="my research")
|
|
other_uid, _ = _user()
|
|
_seed_session(other_uid, query="other research")
|
|
headers = {"X-API-KEY": api_key, **JSON}
|
|
r = requests.get(f"{BASE_URL}/tools/deepsearch/list", headers=headers)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert len(body["sessions"]) == 1
|
|
assert body["sessions"][0]["query"] == "my research"
|
|
finally:
|
|
_clear()
|
|
|
|
|
|
def test_list_empty_for_no_sessions(app_server):
|
|
owner_uid, api_key = _user()
|
|
headers = {"X-API-KEY": api_key, **JSON}
|
|
r = requests.get(f"{BASE_URL}/tools/deepsearch/list", headers=headers)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert "sessions" in body
|
|
assert len(body["sessions"]) == 0
|