Spaces:
Configuration error
Configuration error
File size: 5,688 Bytes
ac98b25 a6e11a5 ac98b25 a6e11a5 ac98b25 a6e11a5 ac98b25 a6e11a5 ac98b25 a6e11a5 ac98b25 a6e11a5 ac98b25 a6e11a5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 | from agentcache.core import KV, SearchService
from agentcache.search import SearchIndex
def test_kv_scopes_import():
"""Verify KV class is importable and defines standard scope keys."""
assert KV.folders == "mem:folders"
assert KV.obs_lookup == "mem:obs_lookup"
assert KV.memories == "mem:memories"
assert KV.folder_obs("src/core", "agent1") == "mem:folder:src/core:agent1"
def test_search_service_index_remove_and_search(tmp_db):
"""Verify SearchService index, remove, and search operations work end-to-end."""
kv = tmp_db
bm25 = SearchIndex()
svc = SearchService(bm25_index=bm25, kv=kv)
obs1 = {
"id": "obs_1",
"title": "Auth Middleware",
"text": "Implemented JWT token validation in authentication middleware",
"folderPath": "src/auth",
"agentId": "agent_alpha",
}
obs2 = {
"id": "obs_2",
"title": "Database Connection",
"text": "SQLite database pool initialization and query optimization",
"folderPath": "src/db",
"agentId": "agent_beta",
}
# Populate KV store for hydration
kv.set(
KV.folders,
"src/auth:agent_alpha",
{"folderPath": "src/auth", "agentId": "agent_alpha"},
)
kv.set(KV.folder_obs("src/auth", "agent_alpha"), "obs_1", obs1)
kv.set(KV.obs_lookup, "obs_1", {"folderPath": "src/auth", "agentId": "agent_alpha"})
kv.set(
KV.folders,
"src/db:agent_beta",
{"folderPath": "src/db", "agentId": "agent_beta"},
)
kv.set(KV.folder_obs("src/db", "agent_beta"), "obs_2", obs2)
kv.set(KV.obs_lookup, "obs_2", {"folderPath": "src/db", "agentId": "agent_beta"})
# Index observations into SearchService
svc.index(obs1)
svc.index(obs2)
# Search for JWT token validation
results = svc.search("JWT token", limit=10)
assert len(results) >= 1
assert results[0]["id"] == "obs_1"
assert "score" in results[0]
# Filtered search by folder_path
filtered = svc.search("token", folder_path="src/db", limit=10)
assert len(filtered) == 0
filtered_auth = svc.search("token", folder_path="src/auth", limit=10)
assert len(filtered_auth) == 1
assert filtered_auth[0]["id"] == "obs_1"
# Filtered search by agent_id (returns only matching agent)
filtered_agent = svc.search("validation", agent_id="agent_alpha", limit=10)
assert len(filtered_agent) == 1
assert filtered_agent[0]["agentId"] == "agent_alpha"
filtered_wrong_agent = svc.search("validation", agent_id="agent_other", limit=10)
assert len(filtered_wrong_agent) == 0
# Remove obs1 and verify it no longer matches
svc.remove("obs_1")
post_remove = svc.search("JWT token", limit=10)
assert len(post_remove) == 0
def test_search_service_persistence(tmp_db):
"""Verify IndexPersistence saves and loads index data sharded in SQLite."""
kv = tmp_db
bm25 = SearchIndex()
svc = SearchService(bm25_index=bm25, kv=kv)
obs = {
"id": "obs_persist_1",
"title": "Cache Eviction",
"text": "LRU cache eviction policy implemented for high throughput",
}
kv.set(KV.memories, "obs_persist_1", obs)
svc.index(obs)
svc.flush_persist()
# Create fresh SearchService and load persisted index
new_bm25 = SearchIndex()
new_svc = SearchService(bm25_index=new_bm25, kv=kv)
loaded = new_svc.load_persisted()
assert loaded["bm25"] is True
assert new_svc.bm25_size > 0
results = new_svc.search("LRU cache", limit=10)
assert len(results) >= 1
# Search for a term never indexed returns an empty list (no error/exception)
non_existent = new_svc.search("nonexistenttermxyz", limit=10)
assert isinstance(non_existent, list)
assert len(non_existent) == 0
def test_http_search_routes_end_to_end(app_client):
"""Verify Flask HTTP search endpoints work end-to-end with real SQLite DB."""
client = app_client
import agentcache.app as app_mod
kv = app_mod.kv
search_svc = app_mod.search_service
obs = {
"id": "obs_http_1",
"title": "API Authentication Strategy",
"text": "Secure OAuth2 bearer token authorization for external endpoints",
"folderPath": "src/api",
"agentId": "agent_gamma",
}
kv.set(
KV.folders,
"src/api:agent_gamma",
{"folderPath": "src/api", "agentId": "agent_gamma"},
)
kv.set(KV.folder_obs("src/api", "agent_gamma"), "obs_http_1", obs)
kv.set(
KV.obs_lookup, "obs_http_1", {"folderPath": "src/api", "agentId": "agent_gamma"}
)
search_svc.index(obs)
# POST /agentcache/search
res = client.post("/agentcache/search", json={"query": "OAuth2 bearer token"})
assert res.status_code == 200
data = res.get_json()
assert isinstance(data, list)
assert len(data) == 1
assert data[0]["id"] == "obs_http_1"
assert data[0]["folderPath"] == "src/api"
# POST /agentmemory/search (alias endpoint)
res_alias = client.post(
"/agentmemory/search", json={"query": "OAuth2 bearer token", "limit": 5}
)
assert res_alias.status_code == 200
data_alias = res_alias.get_json()
assert len(data_alias) == 1
assert data_alias[0]["id"] == "obs_http_1"
# Validation paths: missing or empty query -> 400
res_no_query = client.post("/agentcache/search", json={})
assert res_no_query.status_code == 400
res_empty_query = client.post("/agentcache/search", json={"query": ""})
assert res_empty_query.status_code == 400
res_alias_no_query = client.post("/agentmemory/search", json={})
assert res_alias_no_query.status_code == 400
|