OffGridSchedula / tests /test_threads.py
ParetoOptimal's picture
Initial Commit
0366d65
Raw
History Blame Contribute Delete
1.6 kB
"""Thread assembly (server/threads.py) — pure, stub-safe."""
from server.threads import format_thread, rolling_thread
def test_format_thread_skips_empty():
msgs = [
{"sender": "Alex", "text": "dinner 7pm?"},
{"sender": "Me", "text": ""},
{"sender": "Me", "text": "works"},
]
assert format_thread(msgs) == "Alex: dinner 7pm?\nMe: works"
def test_rolling_thread_filters_by_chat():
feed = [
{"chat": "Mom", "sender": "Mom", "text": "a", "timestamp": "2026-06-05T10:00:00"},
{"chat": "Work", "sender": "Boss", "text": "b", "timestamp": "2026-06-05T10:01:00"},
{"chat": "Mom", "sender": "Mom", "text": "c", "timestamp": "2026-06-05T10:02:00"},
]
out = rolling_thread(feed, "Mom")
assert "a" in out and "c" in out and "b" not in out
def test_rolling_thread_window_limit():
feed = [
{"chat": "Mom", "sender": "Mom", "text": str(i), "timestamp": f"2026-06-05T10:{i:02d}:00"}
for i in range(10)
]
out = rolling_thread(feed, "Mom", window=3)
assert out.splitlines() == ["Mom: 7", "Mom: 8", "Mom: 9"]
def test_rolling_thread_drops_old_messages():
feed = [
{"chat": "Mom", "sender": "Mom", "text": "old", "timestamp": "2026-06-05T00:00:00"},
{"chat": "Mom", "sender": "Mom", "text": "new", "timestamp": "2026-06-05T23:00:00"},
]
out = rolling_thread(feed, "Mom", minutes=60)
assert "new" in out and "old" not in out
def test_rolling_thread_empty_for_unknown_chat():
assert rolling_thread([], "Nobody") == ""