test_s4_sse.py
3.65 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
#!/usr/bin/env python3
"""Test S4: SSE streaming endpoint and call_hermes_stream()."""
import sys, os, unittest
from unittest.mock import patch, MagicMock
from io import StringIO
# Add wechat-viewer to path
sys.path.insert(0, os.path.expanduser("~/wechat-viewer"))
class TestHermesStream(unittest.TestCase):
"""Test the call_hermes_stream() generator."""
def test_call_hermes_stream_exists(self):
"""call_hermes_stream() should exist as a callable function in app module."""
import app
self.assertTrue(hasattr(app, 'call_hermes_stream'),
"app must have call_hermes_stream function")
self.assertTrue(callable(app.call_hermes_stream),
"call_hermes_stream must be callable")
@patch('app.subprocess.Popen')
def test_stream_yields_chunks(self, mock_popen):
"""call_hermes_stream should yield text chunks from subprocess stdout."""
lines = ["Hello", " ", "world\n", "How are you?\n"]
mock_proc = MagicMock()
mock_proc.stdout.__iter__.return_value = iter(lines)
mock_proc.poll.return_value = None
mock_proc.returncode = 0
mock_popen.return_value = mock_proc
import app
chunks = list(app.call_hermes_stream("test input"))
self.assertGreater(len(chunks), 0, "Should yield at least one chunk")
self.assertIn("Hello", chunks[0])
mock_popen.assert_called_once()
@patch('app.subprocess.Popen')
def test_stream_handles_empty_output(self, mock_popen):
"""call_hermes_stream should handle empty stdout gracefully."""
mock_proc = MagicMock()
mock_proc.stdout.readline.return_value = ""
mock_proc.poll.return_value = None
mock_popen.return_value = mock_proc
import app
chunks = list(app.call_hermes_stream("test"))
self.assertEqual(len(chunks), 0, "Empty output should yield no chunks")
def test_stream_route_exists(self):
"""GET /api/conversations/<id>/stream should exist in app.py."""
app_path = os.path.expanduser("~/wechat-viewer/app.py")
code = open(app_path).read()
self.assertIn("/stream", code, "Stream route should exist in app.py")
self.assertIn("text/event-stream", code,
"Should use text/event-stream content type")
def test_response_class_import(self):
"""Flask Response class should be importable for SSE."""
from flask import Response
self.assertTrue(callable(Response), "Flask Response must be callable")
class TestSSEFormat(unittest.TestCase):
"""Test SSE wire format."""
def test_sse_format(self):
"""SSE format should be 'data: <content>\\n\\n'."""
chunks = ["Hello", " ", "world"]
sse_events = []
for chunk in chunks:
sse_events.append(f"data: {chunk}\n\n")
sse_events.append("data: [DONE]\n\n")
output = "".join(sse_events)
self.assertIn("data: Hello", output)
self.assertIn("data: [DONE]", output)
self.assertTrue(output.endswith("\n\n"))
if __name__ == '__main__':
# Quick check if the function/route exists
app_path = os.path.expanduser("~/wechat-viewer/app.py")
code = open(app_path).read()
has_stream_func = 'def call_hermes_stream' in code
has_stream_route = '/stream' in code and 'text/event-stream' in code
print(f"call_hermes_stream defined: {has_stream_func}")
print(f"SSE route + text/event-stream: {has_stream_route}")
if not has_stream_func or not has_stream_route:
print("RED: Missing SSE implementation — test expected to fail")
sys.exit(1)
unittest.main(verbosity=2)