145 lines
4.4 KiB
Python
145 lines
4.4 KiB
Python
import fnmatch
|
|
import os
|
|
import sys
|
|
import threading
|
|
|
|
|
|
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
if PROJECT_ROOT not in sys.path:
|
|
sys.path.insert(0, PROJECT_ROOT)
|
|
|
|
|
|
class _FakeRedis:
|
|
_store: dict[str, object] = {}
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
@classmethod
|
|
def reset(cls) -> None:
|
|
cls._store = {}
|
|
|
|
def keys(self, pattern: str):
|
|
return [key for key in self._store.keys() if fnmatch.fnmatch(key, pattern)]
|
|
|
|
def delete(self, *keys: str):
|
|
deleted = 0
|
|
for key in keys:
|
|
if key in self._store:
|
|
deleted += 1
|
|
del self._store[key]
|
|
return deleted
|
|
|
|
|
|
def test_delete_context_removes_file_and_redis_keys(tmp_path, monkeypatch):
|
|
import context.context_manager as context_manager_module
|
|
|
|
_FakeRedis.reset()
|
|
monkeypatch.setattr(context_manager_module.redis, "Redis", _FakeRedis)
|
|
|
|
context_id = "conv-1"
|
|
context_file = tmp_path / f"ctx_{context_id}.json"
|
|
context_file.write_text("{}", encoding="utf-8")
|
|
|
|
_FakeRedis._store = {
|
|
f"context:{context_id}:messages": ["message"],
|
|
f"context:{context_id}:metadata": "{}",
|
|
}
|
|
|
|
manager = object.__new__(context_manager_module.ContextManager)
|
|
manager._lock = threading.RLock()
|
|
manager._active_contexts = {}
|
|
manager.context_dir = str(tmp_path)
|
|
manager.config = type(
|
|
"Cfg",
|
|
(),
|
|
{"REDIS_HOST": "localhost", "REDIS_PORT": 9736, "REDIS_DB": 0},
|
|
)()
|
|
|
|
assert manager.delete_context(context_id) is True
|
|
assert context_file.exists() is False
|
|
assert not any(
|
|
key.startswith(f"context:{context_id}:") for key in _FakeRedis._store
|
|
)
|
|
|
|
|
|
def test_delete_sketch_pad_removes_file_and_redis_keys(tmp_path, monkeypatch):
|
|
import context.sketch_manager as sketch_manager_module
|
|
|
|
_FakeRedis.reset()
|
|
monkeypatch.setattr(sketch_manager_module.redis, "Redis", _FakeRedis)
|
|
|
|
sketch_id = "conv-1"
|
|
sketch_file = tmp_path / f"skt_{sketch_id}.json"
|
|
sketch_file.write_text("{}", encoding="utf-8")
|
|
|
|
_FakeRedis._store = {
|
|
f"sketch_pad:{sketch_id}:code": "print('x')",
|
|
f"sketch_pad:{sketch_id}:tag:model": ["code"],
|
|
}
|
|
|
|
manager = object.__new__(sketch_manager_module.SketchManager)
|
|
manager._lock = threading.RLock()
|
|
manager._active_sketches = {}
|
|
manager.sketch_dir = str(tmp_path)
|
|
manager.config = type(
|
|
"Cfg",
|
|
(),
|
|
{"REDIS_HOST": "localhost", "REDIS_PORT": 9736, "REDIS_DB": 0},
|
|
)()
|
|
|
|
assert manager.delete_sketch_pad(sketch_id) is True
|
|
assert sketch_file.exists() is False
|
|
assert not any(
|
|
key.startswith(f"sketch_pad:{sketch_id}:") for key in _FakeRedis._store
|
|
)
|
|
|
|
|
|
def test_delete_all_conversations_discovers_ids_across_sources(tmp_path):
|
|
from context.conversation_manager import ConversationManager
|
|
|
|
marker_id = "marker-only"
|
|
active_id = "active-only"
|
|
context_id = "context-only"
|
|
sketch_id = "sketch-only"
|
|
|
|
conversations_dir = tmp_path / "conversations"
|
|
conversations_dir.mkdir()
|
|
(conversations_dir / f"conv_{marker_id}.marker").write_text("", encoding="utf-8")
|
|
|
|
deleted_contexts: list[str] = []
|
|
deleted_sketches: list[str] = []
|
|
|
|
class _FakeContextManager:
|
|
def list_context_ids(self):
|
|
return [context_id]
|
|
|
|
def delete_context(self, conversation_id: str):
|
|
deleted_contexts.append(conversation_id)
|
|
return conversation_id in {marker_id, active_id, context_id, sketch_id}
|
|
|
|
class _FakeSketchManager:
|
|
def list_sketch_ids(self):
|
|
return [sketch_id]
|
|
|
|
def delete_sketch_pad(self, conversation_id: str):
|
|
deleted_sketches.append(conversation_id)
|
|
return conversation_id in {marker_id, active_id, context_id, sketch_id}
|
|
|
|
manager = object.__new__(ConversationManager)
|
|
manager._lock = threading.RLock()
|
|
manager._active_conversations = {active_id: object()}
|
|
manager.context_manager = _FakeContextManager()
|
|
manager.sketch_manager = _FakeSketchManager()
|
|
manager.conversations_dir = str(conversations_dir)
|
|
|
|
deleted_ids = manager.delete_all_conversations()
|
|
|
|
assert sorted(deleted_ids) == sorted([marker_id, active_id, context_id, sketch_id])
|
|
assert sorted(deleted_contexts) == sorted(
|
|
[marker_id, active_id, context_id, sketch_id]
|
|
)
|
|
assert sorted(deleted_sketches) == sorted(
|
|
[marker_id, active_id, context_id, sketch_id]
|
|
)
|