"""TDD RED phase - MemoryIndexer 테스트 TC01 ~ TC12: SQLite FTS5 기반 메모리 인덱서 전체 기능 검증. 이 단계에서는 utils/memory_indexer.py 미구현 상태이므로 ImportError만 발생해야 한다 (RED phase). """ import os import sqlite3 import sys import textwrap from pathlib import Path import pytest sys.path.insert(0, "/home/jay/workspace") from utils.memory_indexer import MemoryIndexer # noqa: E402 (RED: ImportError 예상) # --------------------------------------------------------------------------- # 공통 헬퍼 # --------------------------------------------------------------------------- DIARY_CONTENT = textwrap.dedent("""\ --- date: "2026-04-05" session: 1 team_id: "dev3" task_id: "task-9999" tags: [auto-compact, pre-compact] auto_generated: true --- ## 수행한 작업 - 메모리 인덱서 TDD RED phase 작성 - SQLite FTS5 테이블 설계 검토 ## 피드백 및 수정 지시 - 한국어 검색 성능 확인 필요 """) MEMORY_CONTENT = textwrap.dedent("""\ --- name: 위임 규칙 종합 description: 파일 기반 위임, 논리적 팀, 한정승인 등 type: feedback team_id: "dev3" --- ### 파일 기반 위임 (절대 규칙) - 지시 내용 → 파일 작성 → `--task-file`로 전달. - 100자 이내만 `--task` 허용. ### 논리적 팀 - 마케팅/컨설팅/출판/디자인 라우팅 규칙 준수. """) MEMORY_CONTENT_UPDATED = textwrap.dedent("""\ --- name: 위임 규칙 종합 (업데이트) description: 파일 기반 위임, 논리적 팀, 한정승인 등 - 수정됨 type: feedback team_id: "dev3" --- ### 파일 기반 위임 (절대 규칙, 수정) - 지시 내용 → 파일 작성 → `--task-file`로 전달. - 200자 이내까지 `--task` 허용으로 변경. """) def make_diary_file(directory: Path, filename: str = "2026-04-05-session-01.md") -> Path: """임시 디렉토리에 diary 파일 생성.""" f = directory / filename f.write_text(DIARY_CONTENT, encoding="utf-8") return f def make_memory_file(directory: Path, filename: str = "delegation_rules.md") -> Path: """임시 디렉토리에 memory 파일 생성.""" f = directory / filename f.write_text(MEMORY_CONTENT, encoding="utf-8") return f # --------------------------------------------------------------------------- # TC01: DB 파일 생성 및 테이블 존재 확인 # --------------------------------------------------------------------------- def test_TC01_db_creation(tmp_path): """DB 파일이 생성되고 memories + memories_fts 테이블이 존재해야 한다.""" db_path = str(tmp_path / "test_index.db") indexer = MemoryIndexer(db_path=db_path) indexer.close() # DB 파일이 실제로 생성됐는지 확인 assert Path(db_path).exists(), "DB 파일이 생성되지 않았습니다." conn = sqlite3.connect(db_path) cursor = conn.cursor() # memories 테이블 존재 확인 cursor.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name='memories'" ) assert cursor.fetchone() is not None, "memories 테이블이 존재하지 않습니다." # memories_fts 가상 테이블 존재 확인 cursor.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name='memories_fts'" ) assert cursor.fetchone() is not None, "memories_fts FTS5 가상 테이블이 존재하지 않습니다." conn.close() # --------------------------------------------------------------------------- # TC02: diary 파일 frontmatter 파싱 # --------------------------------------------------------------------------- def test_TC02_parse_diary_file(tmp_path): """diary 파일에서 date, session, team_id, task_id, tags를 올바르게 파싱해야 한다.""" db_path = str(tmp_path / "test_index.db") diary_file = make_diary_file(tmp_path) indexer = MemoryIndexer(db_path=db_path) parsed = indexer._parse_file(str(diary_file)) indexer.close() assert parsed is not None, "_parse_file()이 None을 반환했습니다." assert parsed.get("team_id") == "dev3", f"team_id 불일치: {parsed.get('team_id')}" assert parsed.get("task_id") == "task-9999", f"task_id 불일치: {parsed.get('task_id')}" # tags는 리스트 또는 쉼표 구분 문자열로 반환될 수 있음 tags_raw = parsed.get("tags", "") tags_str = tags_raw if isinstance(tags_raw, str) else ",".join(tags_raw) assert "auto-compact" in tags_str, f"tags에 'auto-compact' 없음: {tags_str}" assert "pre-compact" in tags_str, f"tags에 'pre-compact' 없음: {tags_str}" # content가 있어야 함 assert parsed.get("content"), "content가 비어있습니다." # --------------------------------------------------------------------------- # TC03: 아누 메모리 파일 frontmatter 파싱 # --------------------------------------------------------------------------- def test_TC03_parse_memory_file(tmp_path): """메모리 파일에서 name(→title), type, description을 올바르게 파싱해야 한다.""" db_path = str(tmp_path / "test_index.db") mem_file = make_memory_file(tmp_path) indexer = MemoryIndexer(db_path=db_path) parsed = indexer._parse_file(str(mem_file)) indexer.close() assert parsed is not None, "_parse_file()이 None을 반환했습니다." # name 필드가 title로 매핑되거나 그대로 존재해야 함 title = parsed.get("title") or parsed.get("name") assert title, "title(또는 name) 필드가 없습니다." assert "위임 규칙" in title, f"title에 '위임 규칙' 없음: {title}" assert parsed.get("type") == "feedback", f"type 불일치: {parsed.get('type')}" # --------------------------------------------------------------------------- # TC04: 단일 파일 인덱싱 후 DB 레코드 확인 # --------------------------------------------------------------------------- def test_TC04_index_single_file(tmp_path): """단일 파일을 인덱싱하면 memories 테이블에 레코드가 삽입되어야 한다.""" db_path = str(tmp_path / "test_index.db") diary_file = make_diary_file(tmp_path) indexer = MemoryIndexer(db_path=db_path) result = indexer.index_file(str(diary_file)) indexer.close() assert result == "indexed", f"index_file() 반환값 불일치: {result!r} (기대: 'indexed')" # DB에 실제 레코드가 있는지 확인 conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("SELECT COUNT(*) FROM memories WHERE file_path = ?", (str(diary_file),)) count = cursor.fetchone()[0] conn.close() assert count == 1, f"DB에 레코드가 없습니다. count={count}" # --------------------------------------------------------------------------- # TC05: 동일 파일 재인덱싱 시 'skipped' 반환 (해시 동일) # --------------------------------------------------------------------------- def test_TC05_incremental_indexing(tmp_path): """파일 내용이 변경되지 않으면 재인덱싱 시 'skipped'를 반환해야 한다.""" db_path = str(tmp_path / "test_index.db") diary_file = make_diary_file(tmp_path) indexer = MemoryIndexer(db_path=db_path) first = indexer.index_file(str(diary_file)) second = indexer.index_file(str(diary_file)) indexer.close() assert first == "indexed", f"첫 번째 인덱싱 반환값 불일치: {first!r}" assert second == "skipped", f"재인덱싱(동일 해시) 반환값 불일치: {second!r} (기대: 'skipped')" # --------------------------------------------------------------------------- # TC06: 파일 내용 변경 후 재인덱싱 시 'updated' 반환 # --------------------------------------------------------------------------- def test_TC06_update_on_change(tmp_path): """파일 내용이 변경되면 재인덱싱 시 'updated'를 반환해야 한다.""" db_path = str(tmp_path / "test_index.db") mem_file = make_memory_file(tmp_path) indexer = MemoryIndexer(db_path=db_path) first = indexer.index_file(str(mem_file)) # 파일 내용 변경 mem_file.write_text(MEMORY_CONTENT_UPDATED, encoding="utf-8") second = indexer.index_file(str(mem_file)) indexer.close() assert first == "indexed", f"첫 번째 인덱싱 반환값 불일치: {first!r}" assert second == "updated", f"변경 후 재인덱싱 반환값 불일치: {second!r} (기대: 'updated')" # --------------------------------------------------------------------------- # TC07: 한국어 검색어로 FTS5 검색 결과 반환 # --------------------------------------------------------------------------- def test_TC07_fts5_search_korean(tmp_path): """한국어 검색어로 FTS5 검색 시 관련 결과를 반환해야 한다.""" db_path = str(tmp_path / "test_index.db") mem_file = make_memory_file(tmp_path) indexer = MemoryIndexer(db_path=db_path) indexer.index_file(str(mem_file)) results = indexer.search("위임 규칙") indexer.close() assert isinstance(results, list), "search()가 리스트를 반환하지 않습니다." assert len(results) > 0, "한국어 검색 결과가 비어있습니다." # 첫 번째 결과에 필수 키가 있어야 함 first = results[0] for key in ("file_path", "title", "type", "snippet"): assert key in first, f"결과에 '{key}' 키가 없습니다: {first.keys()}" # --------------------------------------------------------------------------- # TC08: type_filter, team_filter 적용 검색 # --------------------------------------------------------------------------- def test_TC08_search_with_filter(tmp_path): """type_filter와 team_filter를 적용하면 조건에 맞는 결과만 반환해야 한다.""" db_path = str(tmp_path / "test_index.db") # diary 파일 (type=diary, team_id=dev3) diary_file = make_diary_file(tmp_path, "diary_test.md") # memory 파일 (type=feedback, team_id=dev3) mem_file = make_memory_file(tmp_path, "mem_test.md") indexer = MemoryIndexer(db_path=db_path) indexer.index_file(str(diary_file)) indexer.index_file(str(mem_file)) # type=feedback 필터 feedback_results = indexer.search("규칙", type_filter="feedback") # type=diary 필터 (diary 파일에서 "메모리" 검색) diary_results = indexer.search("메모리", type_filter="diary") indexer.close() # feedback 결과는 type이 feedback이어야 함 for r in feedback_results: assert r.get("type") == "feedback", ( f"type_filter='feedback'인데 type={r.get('type')!r}가 반환됨" ) # diary 결과는 type이 diary이어야 함 for r in diary_results: assert r.get("type") == "diary", ( f"type_filter='diary'인데 type={r.get('type')!r}가 반환됨" ) # --------------------------------------------------------------------------- # TC09: reindex_all() 호출 시 전체 재구축 # --------------------------------------------------------------------------- def test_TC09_reindex_all(tmp_path, monkeypatch): """reindex_all() 호출 시 DIARY_DIR + ANU_MEMORY_DIR를 재인덱싱한다.""" db_path = str(tmp_path / "test_index.db") # 임시 디렉토리를 DIARY_DIR, ANU_MEMORY_DIR로 패치 diary_dir = tmp_path / "diary" diary_dir.mkdir() anu_dir = tmp_path / "anu_memory" anu_dir.mkdir() # 각 디렉토리에 파일 생성 (diary_dir / "2026-04-05-session-01.md").write_text(DIARY_CONTENT, encoding="utf-8") (diary_dir / "2026-04-05-session-02.md").write_text(DIARY_CONTENT, encoding="utf-8") (anu_dir / "delegation_rules.md").write_text(MEMORY_CONTENT, encoding="utf-8") monkeypatch.setattr(MemoryIndexer, "DIARY_DIR", str(diary_dir)) monkeypatch.setattr(MemoryIndexer, "ANU_MEMORY_DIR", str(anu_dir)) indexer = MemoryIndexer(db_path=db_path) stats = indexer.reindex_all() indexer.close() assert isinstance(stats, dict), "reindex_all()이 dict를 반환하지 않습니다." total_processed = stats.get("indexed", 0) + stats.get("updated", 0) assert total_processed >= 3, ( f"reindex_all() 후 처리된 파일 수가 3 미만입니다: {stats}" ) # --------------------------------------------------------------------------- # TC10: stats() 반환값 구조 검증 # --------------------------------------------------------------------------- def test_TC10_stats(tmp_path): """stats()는 total, by_type, by_team, last_indexed 키를 포함해야 한다.""" db_path = str(tmp_path / "test_index.db") diary_file = make_diary_file(tmp_path) mem_file = make_memory_file(tmp_path) indexer = MemoryIndexer(db_path=db_path) indexer.index_file(str(diary_file)) indexer.index_file(str(mem_file)) result = indexer.stats() indexer.close() assert isinstance(result, dict), "stats()가 dict를 반환하지 않습니다." assert "total" in result, "stats()에 'total' 키가 없습니다." assert "by_type" in result, "stats()에 'by_type' 키가 없습니다." assert "by_team" in result, "stats()에 'by_team' 키가 없습니다." assert "last_indexed" in result, "stats()에 'last_indexed' 키가 없습니다." assert result["total"] >= 2, f"total이 2 미만입니다: {result['total']}" assert isinstance(result["by_type"], dict), "by_type이 dict가 아닙니다." assert isinstance(result["by_team"], dict), "by_team이 dict가 아닙니다." # --------------------------------------------------------------------------- # TC11: 디렉토리 내 복수 파일 인덱싱 통계 검증 # --------------------------------------------------------------------------- def test_TC11_index_directory(tmp_path): """index_directory()는 디렉토리 내 모든 .md 파일을 인덱싱하고 통계를 반환해야 한다.""" db_path = str(tmp_path / "test_index.db") md_dir = tmp_path / "md_files" md_dir.mkdir() # .md 파일 3개 생성 (md_dir / "file_01.md").write_text(DIARY_CONTENT, encoding="utf-8") (md_dir / "file_02.md").write_text(MEMORY_CONTENT, encoding="utf-8") (md_dir / "file_03.md").write_text(DIARY_CONTENT, encoding="utf-8") # .txt 파일은 무시되어야 함 (md_dir / "ignore.txt").write_text("무시되어야 할 파일", encoding="utf-8") indexer = MemoryIndexer(db_path=db_path) stats = indexer.index_directory(str(md_dir)) indexer.close() assert isinstance(stats, dict), "index_directory()가 dict를 반환하지 않습니다." for key in ("indexed", "updated", "skipped", "errors"): assert key in stats, f"stats에 '{key}' 키가 없습니다: {stats.keys()}" total = stats["indexed"] + stats["updated"] + stats["skipped"] + stats["errors"] assert total == 3, ( f"처리된 파일 수가 3이어야 하는데 {total}입니다. stats={stats}" ) assert stats["indexed"] == 3, ( f"신규 인덱싱 파일 수가 3이어야 하는데 {stats['indexed']}입니다." ) # --------------------------------------------------------------------------- # TC12: 빈 DB에서 검색 시 빈 리스트 반환 # --------------------------------------------------------------------------- def test_TC12_search_empty_db(tmp_path): """아무것도 인덱싱하지 않은 빈 DB에서 검색 시 빈 리스트를 반환해야 한다.""" db_path = str(tmp_path / "test_index.db") indexer = MemoryIndexer(db_path=db_path) results = indexer.search("한국어 검색어") indexer.close() assert isinstance(results, list), "search()가 리스트를 반환하지 않습니다." assert results == [], f"빈 DB에서 검색 결과가 있습니다: {results}"