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