e49e3417eb
- Add logviewer/: Dozzle web UI (port 9999) + analyze.py CLI tool - docker-compose.yml: add json-file logging with rotation and labels for index/search - Fix Dockerfiles: COPY *.py . so all modules are included in image - Convert all relative imports to flat absolute imports for Docker flat layout - Rename index/schemas.py → index/index_schemas.py to avoid module name collision with search/schemas.py in test runner - Update all tests to add service dir to sys.path and use flat imports Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
125 lines
3.9 KiB
Python
125 lines
3.9 KiB
Python
"""Unit tests for index/chunking.py"""
|
|
import sys
|
|
import os
|
|
|
|
_INDEX_DIR = os.path.join(os.path.dirname(__file__), "..", "index")
|
|
sys.path.insert(0, _INDEX_DIR)
|
|
|
|
from chunking import build_chunks, _split_windows, WINDOW_MAX_MESSAGES, TIME_GAP_SECONDS
|
|
from cleaning import CleanedMessage
|
|
from index_schemas import Message
|
|
|
|
|
|
def _make_message(id: str, time: int, text: str = "hello", **kwargs) -> Message:
|
|
defaults = dict(
|
|
thread_sn=None,
|
|
sender_id="user@example.com",
|
|
file_snippets="",
|
|
parts=None,
|
|
mentions=None,
|
|
member_event=None,
|
|
is_system=False,
|
|
is_hidden=False,
|
|
is_forward=False,
|
|
is_quote=False,
|
|
)
|
|
defaults.update(kwargs)
|
|
return Message(id=id, time=time, text=text, **defaults)
|
|
|
|
|
|
def _make_cleaned(id: str, time: int, text: str = "hello") -> CleanedMessage:
|
|
return CleanedMessage(
|
|
id=id,
|
|
sender_id="user@x.com",
|
|
time=time,
|
|
thread_sn=None,
|
|
text=text,
|
|
parts=[],
|
|
mentions=[],
|
|
member_event_text="",
|
|
file_info=[],
|
|
is_system=False,
|
|
is_forward=False,
|
|
is_quote=False,
|
|
)
|
|
|
|
|
|
class TestBuildChunks:
|
|
def test_empty_new_messages(self):
|
|
result = build_chunks([], [])
|
|
assert result == []
|
|
|
|
def test_single_message(self):
|
|
msgs = [_make_message("m1", 1000000, text="A simple message")]
|
|
result = build_chunks([], msgs)
|
|
assert len(result) == 1
|
|
assert "m1" in result[0].message_ids
|
|
|
|
def test_message_ids_preserved(self):
|
|
msgs = [
|
|
_make_message("m1", 1000000, text="First"),
|
|
_make_message("m2", 1000100, text="Second"),
|
|
]
|
|
result = build_chunks([], msgs)
|
|
all_ids = [mid for chunk in result for mid in chunk.message_ids]
|
|
assert "m1" in all_ids
|
|
assert "m2" in all_ids
|
|
|
|
def test_different_content_fields(self):
|
|
msgs = [_make_message("m1", 1000000, text="test")]
|
|
result = build_chunks([], msgs)
|
|
chunk = result[0]
|
|
assert chunk.page_content
|
|
assert chunk.dense_content
|
|
assert chunk.sparse_content
|
|
|
|
def test_overlap_appears_in_chunk(self):
|
|
overlap = [_make_message("o1", 999000, text="overlap message")]
|
|
new_msgs = [_make_message("m1", 1000000, text="new message")]
|
|
result = build_chunks(overlap, new_msgs)
|
|
assert len(result) >= 1
|
|
# overlap ids should NOT be in message_ids (they're context only)
|
|
assert "o1" not in result[0].message_ids
|
|
assert "m1" in result[0].message_ids
|
|
|
|
def test_time_gap_splits_window(self):
|
|
msgs = [
|
|
_make_message("m1", 1000000, text="morning message"),
|
|
_make_message("m2", 1000000 + TIME_GAP_SECONDS + 1, text="evening message"),
|
|
]
|
|
result = build_chunks([], msgs)
|
|
# large time gap should create 2 chunks
|
|
assert len(result) == 2
|
|
|
|
def test_empty_messages_skipped(self):
|
|
msgs = [
|
|
_make_message("m1", 1000000, text=""),
|
|
_make_message("m2", 1000100, text="real content"),
|
|
]
|
|
result = build_chunks([], msgs)
|
|
all_ids = [mid for chunk in result for mid in chunk.message_ids]
|
|
assert "m1" not in all_ids
|
|
assert "m2" in all_ids
|
|
|
|
|
|
class TestSplitWindows:
|
|
def test_empty(self):
|
|
assert _split_windows([]) == []
|
|
|
|
def test_single(self):
|
|
msgs = [_make_cleaned("m1", 1000000)]
|
|
windows = _split_windows(msgs)
|
|
assert len(windows) == 1
|
|
|
|
def test_time_gap_splits(self):
|
|
msgs = [
|
|
_make_cleaned("m1", 1000000),
|
|
_make_cleaned("m2", 1000000 + TIME_GAP_SECONDS + 1),
|
|
]
|
|
windows = _split_windows(msgs)
|
|
assert len(windows) == 2
|
|
|
|
def test_max_messages_splits(self):
|
|
msgs = [_make_cleaned(f"m{i}", 1000000 + i * 10) for i in range(WINDOW_MAX_MESSAGES + 2)]
|
|
windows = _split_windows(msgs)
|
|
assert len(windows) >= 2
|