"""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") == ""