import asyncio import os import sqlite3 from pathlib import Path from unittest.mock import patch from agent.memory.storage import MemoryChunk, MemoryStorage from agent.knowledge.service import KnowledgeService class FakeStorage: def __init__(self): self.deleted = [] def delete_by_path(self, path): self.deleted.append(path) class FakeMemoryManager: def __init__(self): self.storage = FakeStorage() self.dirty = 0 self.synced = 0 def mark_dirty(self): self.dirty += 1 async def sync(self): self.synced += 1 def service(tmp_path): (tmp_path / "knowledge").mkdir() manager = FakeMemoryManager() return KnowledgeService(str(tmp_path), manager), manager def test_category_lifecycle_and_confirmation(tmp_path): svc, manager = service(tmp_path) assert svc.dispatch("create_category", {"path": "notes"})["payload"]["created"] (tmp_path / "knowledge/notes/a.md").write_text("# A", encoding="utf-8") denied = svc.dispatch("delete_category", {"path": "notes"}) assert denied["code"] == 403 result = svc.dispatch("delete_category", {"path": "notes", "confirm": True}) assert result["payload"]["deleted_documents"] == 1 assert manager.storage.deleted == ["knowledge/notes/a.md"] assert manager.synced == 1 def test_rename_category_reindexes_documents(tmp_path): svc, manager = service(tmp_path) (tmp_path / "knowledge/old/sub").mkdir(parents=True) (tmp_path / "knowledge/old/a.md").write_text("a", encoding="utf-8") (tmp_path / "knowledge/old/sub/b.md").write_text("b", encoding="utf-8") result = svc.dispatch("rename_category", {"path": "old", "new_path": "new"}) assert result["code"] == 200 assert sorted(manager.storage.deleted) == [ "knowledge/old/a.md", "knowledge/old/sub/b.md" ] assert manager.synced == 1 assert (tmp_path / "knowledge/new/sub/b.md").exists() def test_delete_documents_is_idempotent_and_protects_metadata(tmp_path): svc, manager = service(tmp_path) (tmp_path / "knowledge/index.md").write_text("index", encoding="utf-8") (tmp_path / "knowledge/a.md").write_text("a", encoding="utf-8") protected = svc.dispatch("delete_documents", {"paths": ["index.md"]}) assert protected["code"] == 403 first = svc.dispatch("delete_documents", {"paths": ["a.md", "missing.md"]}) assert first["payload"]["deleted"] == 1 second = svc.dispatch("delete_documents", {"paths": ["a.md"]}) assert second["payload"]["deleted"] == 0 assert manager.storage.deleted == ["knowledge/a.md", "knowledge/missing.md", "knowledge/a.md"] assert manager.synced == 2 def test_move_documents_rejects_overwrite_and_syncs(tmp_path): svc, manager = service(tmp_path) (tmp_path / "knowledge/source").mkdir() (tmp_path / "knowledge/target").mkdir() (tmp_path / "knowledge/source/a.md").write_text("a", encoding="utf-8") (tmp_path / "knowledge/source/b.md").write_text("b", encoding="utf-8") (tmp_path / "knowledge/target/b.md").write_text("existing", encoding="utf-8") result = svc.dispatch("move_documents", { "paths": ["source/a.md", "source/b.md"], "target_category": "target" }) assert result["payload"]["moved"] == 1 assert result["payload"]["results"][1]["reason"] == "target_exists" assert manager.storage.deleted == ["knowledge/source/a.md"] assert manager.synced == 1 def test_path_traversal_and_symlink_escape_are_rejected(tmp_path): svc, _ = service(tmp_path) outside = tmp_path / "outside" outside.mkdir() (tmp_path / "knowledge/link").symlink_to(outside, target_is_directory=True) assert svc.dispatch("create_category", {"path": "../bad"})["code"] == 403 assert svc.dispatch("create_category", {"path": "link/bad"})["code"] == 403 def test_dispatch_sync_works_inside_running_event_loop(tmp_path): svc, manager = service(tmp_path) (tmp_path / "knowledge/source").mkdir() (tmp_path / "knowledge/target").mkdir() (tmp_path / "knowledge/source/a.md").write_text("a", encoding="utf-8") async def run(): return svc.dispatch("move_documents", { "paths": ["source/a.md"], "target_category": "target" }) assert asyncio.run(run())["code"] == 200 assert manager.synced == 1 def test_real_storage_delete_by_path_removes_chunks_and_file_metadata(tmp_path): storage = MemoryStorage(tmp_path / "index.db") path = "knowledge/category/a.md" storage.save_chunks_batch([MemoryChunk( id="chunk-1", user_id=None, scope="shared", source="knowledge", path=path, start_line=1, end_line=1, text="unique content", embedding=None, hash="hash-1", )]) storage.update_file_metadata(path, "knowledge", "file-hash", 1, 14) storage.delete_by_path(path) assert storage.conn.execute("SELECT COUNT(*) FROM chunks WHERE path = ?", (path,)).fetchone()[0] == 0 assert storage.conn.execute("SELECT COUNT(*) FROM files WHERE path = ?", (path,)).fetchone()[0] == 0 storage.close() def test_missing_document_still_cleans_stale_index(tmp_path): svc, manager = service(tmp_path) result = svc.dispatch("delete_documents", {"paths": ["removed-by-agent.md"]}) assert result["code"] == 200 assert result["payload"]["results"][0]["reason"] == "not_found" assert manager.storage.deleted == ["knowledge/removed-by-agent.md"] assert manager.dirty == 1 assert manager.synced == 1 def test_category_rename_handles_concurrent_disappearance(tmp_path): svc, manager = service(tmp_path) category = tmp_path / "knowledge/source" category.mkdir() (category / "a.md").write_text("a", encoding="utf-8") def disappear_then_rename(path, target): (category / "a.md").unlink() category.rmdir() raise FileNotFoundError(path) with patch.object(Path, "rename", disappear_then_rename): result = svc.dispatch("rename_category", {"path": "source", "new_path": "target"}) assert result["code"] == 200 assert result["payload"]["reason"] == "not_found" assert manager.storage.deleted == [] def test_category_delete_handles_concurrent_disappearance(tmp_path): svc, manager = service(tmp_path) category = tmp_path / "knowledge/source" category.mkdir() (category / "a.md").write_text("a", encoding="utf-8") def disappear_then_delete(path): (category / "a.md").unlink() category.rmdir() raise FileNotFoundError(path) with patch("agent.knowledge.service.shutil.rmtree", side_effect=disappear_then_delete): result = svc.dispatch("delete_category", {"path": "source", "confirm": True}) assert result["code"] == 200 assert result["payload"]["reason"] == "not_found" assert manager.storage.deleted == [] def test_move_does_not_overwrite_target_created_concurrently(tmp_path): svc, manager = service(tmp_path) (tmp_path / "knowledge/source").mkdir() (tmp_path / "knowledge/target").mkdir() source = tmp_path / "knowledge/source/a.md" target = tmp_path / "knowledge/target/a.md" source.write_text("source", encoding="utf-8") real_link = os.link def create_target_then_link(src, dst): target.write_text("concurrent", encoding="utf-8") return real_link(src, dst) with patch("agent.knowledge.service.os.link", side_effect=create_target_then_link): result = svc.dispatch("move_documents", { "paths": ["source/a.md"], "target_category": "target", }) assert result["payload"]["results"][0]["reason"] == "target_exists" assert source.read_text(encoding="utf-8") == "source" assert target.read_text(encoding="utf-8") == "concurrent" assert manager.storage.deleted == []