168 lines
5.1 KiB
Python
168 lines
5.1 KiB
Python
"""Unit tests for index/cleaning.py"""
|
|
import sys
|
|
import os
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
import pytest
|
|
from index.cleaning import (
|
|
normalize_unicode,
|
|
parse_file_snippets,
|
|
normalize_member_event,
|
|
normalize_part,
|
|
clean_message,
|
|
)
|
|
from index.schemas import Message
|
|
|
|
|
|
def _make_message(**kwargs) -> Message:
|
|
defaults = dict(
|
|
id="msg1",
|
|
thread_sn=None,
|
|
time=1000000,
|
|
text="",
|
|
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(**defaults)
|
|
|
|
|
|
class TestNormalizeUnicode:
|
|
def test_removes_zero_width(self):
|
|
assert "\u200b" not in normalize_unicode("hello\u200bworld")
|
|
assert "\u200c" not in normalize_unicode("a\u200cb")
|
|
assert "\ufeff" not in normalize_unicode("\ufefftext")
|
|
|
|
def test_collapses_whitespace(self):
|
|
result = normalize_unicode(" too many spaces ")
|
|
assert " " not in result
|
|
|
|
def test_normalizes_newlines(self):
|
|
result = normalize_unicode("line1\r\nline2\rline3")
|
|
assert "\r" not in result
|
|
|
|
def test_collapses_multiple_newlines(self):
|
|
result = normalize_unicode("a\n\n\n\nb")
|
|
assert "\n\n\n" not in result
|
|
|
|
def test_preserves_url(self):
|
|
url = "https://example.com/path?q=1&page=2"
|
|
assert url in normalize_unicode(url)
|
|
|
|
def test_preserves_email(self):
|
|
email = "user@corp.example"
|
|
assert email in normalize_unicode(email)
|
|
|
|
|
|
class TestParseFileSnippets:
|
|
def test_empty_string(self):
|
|
assert parse_file_snippets("") == []
|
|
|
|
def test_valid_list(self):
|
|
raw = '[{"name": "doc.pdf", "mime": "application/pdf"}]'
|
|
result = parse_file_snippets(raw)
|
|
assert len(result) == 1
|
|
assert result[0]["name"] == "doc.pdf"
|
|
|
|
def test_valid_dict(self):
|
|
raw = '{"name": "file.txt"}'
|
|
result = parse_file_snippets(raw)
|
|
assert len(result) == 1
|
|
|
|
def test_invalid_json(self):
|
|
assert parse_file_snippets("{broken json}") == []
|
|
|
|
def test_whitespace_only(self):
|
|
assert parse_file_snippets(" ") == []
|
|
|
|
|
|
class TestNormalizeMemberEvent:
|
|
def test_none_event(self):
|
|
assert normalize_member_event(None) == ""
|
|
|
|
def test_add_members(self):
|
|
event = {"type": "addMembers", "members": ["alice@example.com", "bob@example.com"]}
|
|
result = normalize_member_event(event)
|
|
assert "alice@example.com" in result
|
|
assert "added to chat" in result
|
|
|
|
def test_unknown_event(self):
|
|
event = {"type": "banUser", "member": "x@example.com"}
|
|
result = normalize_member_event(event)
|
|
assert "banUser" in result
|
|
|
|
|
|
class TestNormalizePart:
|
|
def test_text_part(self):
|
|
part = {"mediaType": "text", "text": "hello"}
|
|
result = normalize_part(part)
|
|
assert result["type"] == "text"
|
|
assert result["text"] == "hello"
|
|
|
|
def test_quote_part(self):
|
|
part = {"mediaType": "quote", "sn": "alice@x.com", "text": "original text"}
|
|
result = normalize_part(part)
|
|
assert result["type"] == "quote"
|
|
assert "alice@x.com" in result["text"]
|
|
assert "original text" in result["text"]
|
|
|
|
def test_forward_part(self):
|
|
part = {"mediaType": "forward", "sn": "channel@x.com", "text": "forwarded"}
|
|
result = normalize_part(part)
|
|
assert result["type"] == "forward"
|
|
assert "forwarded" in result["text"]
|
|
|
|
def test_unknown_media_type(self):
|
|
part = {"mediaType": "sticker", "text": ""}
|
|
result = normalize_part(part)
|
|
assert "sticker" in result["type"]
|
|
|
|
|
|
class TestCleanMessage:
|
|
def test_empty_message_is_empty(self):
|
|
msg = _make_message()
|
|
cleaned = clean_message(msg)
|
|
assert cleaned.is_empty
|
|
|
|
def test_text_message(self):
|
|
msg = _make_message(text="Hello world")
|
|
cleaned = clean_message(msg)
|
|
assert not cleaned.is_empty
|
|
assert cleaned.text == "Hello world"
|
|
|
|
def test_parts_extracted(self):
|
|
msg = _make_message(
|
|
parts=[{"mediaType": "text", "text": "from parts"}]
|
|
)
|
|
cleaned = clean_message(msg)
|
|
assert not cleaned.is_empty
|
|
assert any("from parts" in p["text"] for p in cleaned.parts)
|
|
|
|
def test_member_event_extracted(self):
|
|
msg = _make_message(
|
|
is_system=True,
|
|
member_event={"type": "addMembers", "members": ["u@x.com"]},
|
|
)
|
|
cleaned = clean_message(msg)
|
|
assert not cleaned.is_empty
|
|
assert "u@x.com" in cleaned.member_event_text
|
|
|
|
def test_file_snippets_parsed(self):
|
|
msg = _make_message(
|
|
file_snippets='[{"name": "report.pdf", "mime": "application/pdf"}]'
|
|
)
|
|
cleaned = clean_message(msg)
|
|
assert not cleaned.is_empty
|
|
assert cleaned.file_info[0]["name"] == "report.pdf"
|
|
|
|
def test_zero_width_stripped_from_text(self):
|
|
msg = _make_message(text="hello\u200bworld")
|
|
cleaned = clean_message(msg)
|
|
assert "\u200b" not in cleaned.text
|