ReviewOS
Reload

also looking at this

zig-utils/zig-test-framework

Implement the web UI server with Zig I/O

#26
Merged glennmichael123 wants to merge codex/issue-9-ui-server into main
8 files +513 -114
src/ui_server.zigmodified+360-43
@@ -14,47 +14,273 @@ pub const UIServerOptions = struct {
1414 verbose: bool = false,
1515};
1616
17/// UI Server for web-based test visualization
18/// Note: std.net was removed in Zig 0.16. Server functionality is stubbed out
19/// until the new Io-based networking API stabilizes.
17/// UI Server for web-based test visualization.
2018pub const UIServer = struct {
2119 allocator: std.mem.Allocator,
2220 options: UIServerOptions,
2321 mutex: compat.Mutex = .{},
2422 history: ?*test_history.TestHistory = null,
23 threaded_io: std.Io.Threaded,
24 listener: ?std.Io.net.Server = null,
25 thread: ?std.Thread = null,
26 running: std.atomic.Value(bool) = .init(false),
27 active_client: ?std.Io.net.Stream = null,
28 clients: std.ArrayList(std.Io.net.Stream) = .empty,
2529
2630 const Self = @This();
31 const index_html = @embedFile("ui/index.html");
32
33 pub const Error = error{
34 AlreadyStarted,
35 InvalidRequest,
36 ServerStopped,
37 };
2738
2839 pub fn init(allocator: std.mem.Allocator, options: UIServerOptions) Self {
2940 return .{
3041 .allocator = allocator,
3142 .options = options,
43 .threaded_io = .init(allocator, .{ .environ = .empty }),
3244 };
3345 }
3446
3547 pub fn deinit(self: *Self) void {
36 _ = self;
48 self.stop();
49 self.clients.deinit(self.allocator);
50 self.threaded_io.deinit();
3751 }
3852
39 /// Start the UI server (stubbed - std.net removed in Zig 0.16)
53 /// Bind the configured address and start accepting HTTP clients.
4054 pub fn start(self: *Self) !void {
55 if (self.running.load(.acquire) or self.listener != null) return Error.AlreadyStarted;
56
57 const io = self.threaded_io.io();
58 const address = if (std.ascii.eqlIgnoreCase(self.options.host, "localhost"))
59 std.Io.net.IpAddress{ .ip4 = .loopback(self.options.port) }
60 else
61 try std.Io.net.IpAddress.parse(self.options.host, self.options.port);
62
63 self.listener = try address.listen(io, .{ .reuse_address = true });
64 self.options.port = self.listener.?.socket.address.getPort();
65 self.running.store(true, .release);
66 self.thread = std.Thread.spawn(.{}, serve, .{self}) catch |err| {
67 self.running.store(false, .release);
68 self.listener.?.deinit(io);
69 self.listener = null;
70 return err;
71 };
72
4173 if (self.options.verbose) {
42 std.debug.print("UI Server: networking stubbed out (std.net removed in Zig 0.16)\n", .{});
74 std.debug.print("UI Server: listening on http://{s}:{d}\n", .{ self.options.host, self.options.port });
75 }
76 }
77
78 /// Stop accepting connections and close every active event stream.
79 pub fn stop(self: *Self) void {
80 const was_running = self.running.swap(false, .acq_rel);
81 var wake_io: ?std.Io.Threaded = null;
82 var wake_stream: ?std.Io.net.Stream = null;
83
84 if (was_running) {
85 // A client may have connected without finishing its request
86 // headers. Shutdown unblocks the parser while leaving ownership of
87 // the socket with the accept thread.
88 self.mutex.lock();
89 if (self.active_client) |client| client.shutdown(self.threaded_io.io(), .both) catch {};
90 self.mutex.unlock();
91
92 // Closing a listening socket from another thread makes the current
93 // Zig Windows backend treat the canceled accept as unreachable.
94 // Wake it with a loopback connection on an independent I/O runtime:
95 // sharing the server runtime here can deadlock its blocking accept
96 // on Linux and Windows. Keep the connection alive until accept has
97 // returned so the peer cannot disappear from the listen queue.
98 wake_io = .init(self.allocator, .{ .environ = .empty });
99 if (wake_io) |*threaded| {
100 wake_stream = self.connectWake(threaded.io());
101 }
102 }
103
104 if (self.thread) |thread| {
105 thread.join();
106 self.thread = null;
107 }
108
109 if (wake_stream) |stream| {
110 if (wake_io) |*threaded| stream.close(threaded.io());
43111 }
112 if (wake_io) |*threaded| threaded.deinit();
113
114 if (self.listener) |*server| {
115 server.deinit(self.threaded_io.io());
116 self.listener = null;
117 }
118
119 self.mutex.lock();
120 defer self.mutex.unlock();
121 for (self.clients.items) |client| client.close(self.threaded_io.io());
122 self.clients.clearRetainingCapacity();
123 }
124
125 fn connectWake(self: *Self, io: std.Io) ?std.Io.net.Stream {
126 const server = self.listener orelse return null;
127 var address = server.socket.address;
128 switch (address) {
129 .ip4 => |ip4| {
130 if (std.mem.eql(u8, &ip4.bytes, &.{ 0, 0, 0, 0 })) {
131 address = .{ .ip4 = .loopback(ip4.port) };
132 }
133 },
134 .ip6 => |ip6| {
135 const unspecified: [16]u8 = @splat(0);
136 if (std.mem.eql(u8, &ip6.bytes, &unspecified)) {
137 address = .{ .ip6 = .loopback(ip6.port) };
138 }
139 },
140 }
141
142 return address.connect(io, .{ .mode = .stream }) catch null;
44143 }
45144
46 /// Accept a client connection (stubbed)
145 /// Return the actual bound port. This differs from the requested port when
146 /// port zero asks the operating system for an ephemeral port.
147 pub fn port(self: *const Self) u16 {
148 return self.options.port;
149 }
150
151 fn serve(self: *Self) void {
152 while (self.running.load(.acquire)) {
153 self.acceptClient() catch |err| {
154 if (!self.running.load(.acquire)) return;
155 if (self.options.verbose) {
156 std.debug.print("UI Server error: {s}\n", .{@errorName(err)});
157 }
158 compat.sleep(10 * std.time.ns_per_ms);
159 };
160 }
161 }
162
163 /// Accept and serve one HTTP client connection.
47164 pub fn acceptClient(self: *Self) !void {
48 _ = self;
49 // Stubbed: no networking available
50 compat.sleep(100 * std.time.ns_per_ms);
165 if (!self.running.load(.acquire)) return Error.ServerStopped;
166 const stream = if (self.listener) |*listener|
167 try listener.accept(self.threaded_io.io())
168 else
169 return Error.ServerStopped;
170 if (!self.running.load(.acquire)) {
171 stream.close(self.threaded_io.io());
172 return Error.ServerStopped;
173 }
174
175 self.mutex.lock();
176 self.active_client = stream;
177 self.mutex.unlock();
178
179 const disposition = self.handleClient(stream) catch |err| {
180 self.finishClient(false) catch {};
181 return err;
182 };
183 try self.finishClient(disposition == .event_stream);
51184 }
52185
53 /// Broadcast an event to all connected clients (stubbed)
186 const ClientDisposition = enum { close, event_stream };
187
188 fn handleClient(self: *Self, stream: std.Io.net.Stream) !ClientDisposition {
189 var read_buffer: [4096]u8 = undefined;
190 var stream_reader = stream.reader(self.threaded_io.io(), &read_buffer);
191 const request_line_raw = (try stream_reader.interface.takeDelimiter('\n')) orelse return Error.InvalidRequest;
192 const request_line = std.mem.trimEnd(u8, request_line_raw, "\r");
193 var parts = std.mem.splitScalar(u8, request_line, ' ');
194 const method = parts.next() orelse return Error.InvalidRequest;
195 const target = parts.next() orelse return Error.InvalidRequest;
196 _ = parts.next() orelse return Error.InvalidRequest;
197
198 const is_get = std.mem.eql(u8, method, "GET");
199 const is_events = std.mem.eql(u8, target, "/events");
200 const is_index = std.mem.eql(u8, target, "/") or std.mem.eql(u8, target, "/index.html");
201
202 while (true) {
203 const header_raw = (try stream_reader.interface.takeDelimiter('\n')) orelse return Error.InvalidRequest;
204 if (std.mem.trimEnd(u8, header_raw, "\r").len == 0) break;
205 }
206
207 if (!is_get) {
208 try self.respond(stream, "405 Method Not Allowed", "text/plain; charset=utf-8", "Method Not Allowed\n");
209 } else if (is_events) {
210 try self.openEventStream(stream);
211 return .event_stream;
212 } else if (is_index) {
213 try self.respond(stream, "200 OK", "text/html; charset=utf-8", index_html);
214 } else {
215 try self.respond(stream, "404 Not Found", "text/plain; charset=utf-8", "Not Found\n");
216 }
217 return .close;
218 }
219
220 fn finishClient(self: *Self, keep_open: bool) !void {
221 self.mutex.lock();
222 defer self.mutex.unlock();
223 const client = self.active_client orelse return;
224 self.active_client = null;
225
226 if (keep_open and self.running.load(.acquire)) {
227 self.clients.append(self.allocator, client) catch |err| {
228 client.close(self.threaded_io.io());
229 return err;
230 };
231 } else {
232 client.close(self.threaded_io.io());
233 }
234 }
235
236 fn respond(self: *Self, stream: std.Io.net.Stream, status: []const u8, content_type: []const u8, body: []const u8) !void {
237 var write_buffer: [2048]u8 = undefined;
238 var stream_writer = stream.writer(self.threaded_io.io(), &write_buffer);
239 try stream_writer.interface.print(
240 "HTTP/1.1 {s}\r\nContent-Type: {s}\r\nContent-Length: {d}\r\nConnection: close\r\n\r\n",
241 .{ status, content_type, body.len },
242 );
243 try stream_writer.interface.writeAll(body);
244 try stream_writer.interface.flush();
245 }
246
247 fn openEventStream(self: *Self, stream: std.Io.net.Stream) !void {
248 var write_buffer: [512]u8 = undefined;
249 var stream_writer = stream.writer(self.threaded_io.io(), &write_buffer);
250 try stream_writer.interface.writeAll(
251 "HTTP/1.1 200 OK\r\n" ++
252 "Content-Type: text/event-stream\r\n" ++
253 "Cache-Control: no-cache\r\n" ++
254 "Connection: keep-alive\r\n" ++
255 "X-Accel-Buffering: no\r\n\r\n" ++
256 "event: connected\ndata: {}\n\n",
257 );
258 try stream_writer.interface.flush();
259 }
260
261 fn sendEvent(self: *Self, stream: std.Io.net.Stream, event: []const u8, data: []const u8) !void {
262 var write_buffer: [1024]u8 = undefined;
263 var stream_writer = stream.writer(self.threaded_io.io(), &write_buffer);
264 try stream_writer.interface.print("event: {s}\ndata: {s}\n\n", .{ event, data });
265 try stream_writer.interface.flush();
266 }
267
268 /// Broadcast an event to all connected clients. A disconnected browser is
269 /// removed without disrupting the test run or the remaining clients.
54270 pub fn broadcast(self: *Self, event: []const u8, data: []const u8) !void {
55 _ = self;
56 _ = event;
57 _ = data;
271 self.mutex.lock();
272 defer self.mutex.unlock();
273
274 var index: usize = 0;
275 while (index < self.clients.items.len) {
276 const client = self.clients.items[index];
277 self.sendEvent(client, event, data) catch {
278 client.close(self.threaded_io.io());
279 _ = self.clients.swapRemove(index);
280 continue;
281 };
282 index += 1;
283 }
58284 }
59285};
60286
@@ -89,69 +315,62 @@ pub const UIReporter = struct {
89315 self.buffer.deinit(self.reporter.allocator);
90316 }
91317
92 fn onRunStart(reporter: *reporter_mod.Reporter, total: usize) !void {
93 const self: *Self = @fieldParentPtr("reporter", reporter);
94
318 fn encode(self: *Self, value: anytype) !void {
95319 self.buffer.clearRetainingCapacity();
96 try self.buffer.print(reporter.allocator, "", .{total});
320 var writer: std.Io.Writer.Allocating = .fromArrayList(self.reporter.allocator, &self.buffer);
321 defer self.buffer = writer.toArrayList();
322 try std.json.Stringify.value(value, .{}, &writer.writer);
323 }
97324
325 fn onRunStart(reporter: *reporter_mod.Reporter, total: usize) !void {
326 const self: *Self = @fieldParentPtr("reporter", reporter);
327 try self.encode(.{ .total = total });
98328 try self.server.broadcast("run_start", self.buffer.items);
99329 }
100330
101331 fn onRunEnd(reporter: *reporter_mod.Reporter, results: *reporter_mod.TestResults) !void {
102332 const self: *Self = @fieldParentPtr("reporter", reporter);
103333
104 self.buffer.clearRetainingCapacity();
105 try self.buffer.print(reporter.allocator, "", .{
106 results.total,
107 results.passed,
108 results.failed,
109 results.skipped,
334 try self.encode(.{
335 .total = results.total,
336 .passed = results.passed,
337 .flaky = results.flaky,
338 .failed = results.failed,
339 .skipped = results.skipped,
110340 });
111
112341 try self.server.broadcast("run_end", self.buffer.items);
113342 }
114343
115344 fn onSuiteStart(reporter: *reporter_mod.Reporter, suite_name: []const u8) !void {
116345 const self: *Self = @fieldParentPtr("reporter", reporter);
117346
118 self.buffer.clearRetainingCapacity();
119 try self.buffer.print(reporter.allocator, "", .{suite_name});
120
347 try self.encode(.{ .name = suite_name });
121348 try self.server.broadcast("suite_start", self.buffer.items);
122349 }
123350
124351 fn onSuiteEnd(reporter: *reporter_mod.Reporter, suite_name: []const u8) !void {
125352 const self: *Self = @fieldParentPtr("reporter", reporter);
126353
127 self.buffer.clearRetainingCapacity();
128 try self.buffer.print(reporter.allocator, "", .{suite_name});
129
354 try self.encode(.{ .name = suite_name });
130355 try self.server.broadcast("suite_end", self.buffer.items);
131356 }
132357
133358 fn onTestStart(reporter: *reporter_mod.Reporter, test_name: []const u8) !void {
134359 const self: *Self = @fieldParentPtr("reporter", reporter);
135360
136 self.buffer.clearRetainingCapacity();
137 try self.buffer.print(reporter.allocator, "", .{test_name});
138
361 try self.encode(.{ .name = test_name });
139362 try self.server.broadcast("test_start", self.buffer.items);
140363 }
141364
142365 fn onTestEnd(reporter: *reporter_mod.Reporter, test_case: *const suite.TestCase) !void {
143366 const self: *Self = @fieldParentPtr("reporter", reporter);
144367
145 self.buffer.clearRetainingCapacity();
146
147 const error_msg = test_case.error_message orelse "";
148 try self.buffer.print(reporter.allocator, "", .{
149 test_case.name,
150 @tagName(test_case.status),
151 test_case.execution_time_ns,
152 error_msg,
368 try self.encode(.{
369 .name = test_case.name,
370 .status = @tagName(test_case.status),
371 .execution_time_ns = test_case.execution_time_ns,
372 .error_message = test_case.error_message orelse "",
153373 });
154
155374 try self.server.broadcast("test_end", self.buffer.items);
156375 }
157376};
@@ -253,4 +472,102 @@ test "UIReporter initialization" {
253472
254473 try std.testing.expectEqual(false, ui_reporter.reporter.use_colors);
255474 try std.testing.expectEqual(@as(usize, 0), ui_reporter.buffer.items.len);
475
476 try ui_reporter.reporter.onSuiteStart("quoted \"suite\"\nname");
477 try std.testing.expectEqualStrings("{\"name\":\"quoted \\\"suite\\\"\\nname\"}", ui_reporter.buffer.items);
478}
479
480fn writeTestRequest(io: std.Io, stream: std.Io.net.Stream, target: []const u8) !void {
481 var write_buffer: [512]u8 = undefined;
482 var writer = stream.writer(io, &write_buffer);
483 try writer.interface.print("GET {s} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n", .{target});
484 try writer.interface.flush();
485}
486
487fn expectLine(reader: *std.Io.Reader, expected: []const u8) !void {
488 const raw = (try reader.takeDelimiter('\n')) orelse return error.EndOfStream;
489 try std.testing.expectEqualStrings(expected, std.mem.trimEnd(u8, raw, "\r"));
490}
491
492test "UI server serves the embedded page on an ephemeral port" {
493 var server = UIServer.init(std.testing.allocator, .{ .port = 0 });
494 defer server.deinit();
495 try server.start();
496 try std.testing.expect(server.port() != 0);
497
498 var client_io: std.Io.Threaded = .init(std.testing.allocator, .{ .environ = .empty });
499 defer client_io.deinit();
500 const io = client_io.io();
501 const address = std.Io.net.IpAddress{ .ip4 = .loopback(server.port()) };
502 const stream = try address.connect(io, .{ .mode = .stream });
503 defer stream.close(io);
504 try writeTestRequest(io, stream, "/");
505
506 var read_buffer: [4096]u8 = undefined;
507 var stream_reader = stream.reader(io, &read_buffer);
508 const response = try stream_reader.interface.allocRemaining(std.testing.allocator, .limited(UIServer.index_html.len + 1024));
509 defer std.testing.allocator.free(response);
510 try std.testing.expect(std.mem.startsWith(u8, response, "HTTP/1.1 200 OK\r\n"));
511 try std.testing.expect(std.mem.indexOf(u8, response, "Zig Test Framework - Live Results") != null);
512}
513
514test "UI server streams broadcast events over SSE" {
515 var server = UIServer.init(std.testing.allocator, .{ .port = 0 });
516 defer server.deinit();
517 try server.start();
518
519 var client_io: std.Io.Threaded = .init(std.testing.allocator, .{ .environ = .empty });
520 defer client_io.deinit();
521 const io = client_io.io();
522 const address = std.Io.net.IpAddress{ .ip4 = .loopback(server.port()) };
523 const stream = try address.connect(io, .{ .mode = .stream });
524 defer stream.close(io);
525 try writeTestRequest(io, stream, "/events");
526
527 var read_buffer: [4096]u8 = undefined;
528 var stream_reader = stream.reader(io, &read_buffer);
529 try expectLine(&stream_reader.interface, "HTTP/1.1 200 OK");
530 while (true) {
531 const raw = (try stream_reader.interface.takeDelimiter('\n')) orelse return error.EndOfStream;
532 if (std.mem.trimEnd(u8, raw, "\r").len == 0) break;
533 }
534 try expectLine(&stream_reader.interface, "event: connected");
535 try expectLine(&stream_reader.interface, "data: {}");
536 try expectLine(&stream_reader.interface, "");
537
538 try server.broadcast("run_start", "{\"total\":2}");
539 try expectLine(&stream_reader.interface, "event: run_start");
540 try expectLine(&stream_reader.interface, "data: {\"total\":2}");
541 try expectLine(&stream_reader.interface, "");
542}
543
544test "UI server shutdown cancels an incomplete HTTP request" {
545 var server = UIServer.init(std.testing.allocator, .{ .port = 0 });
546 defer server.deinit();
547 try server.start();
548
549 var client_io: std.Io.Threaded = .init(std.testing.allocator, .{ .environ = .empty });
550 defer client_io.deinit();
551 const io = client_io.io();
552 const address = std.Io.net.IpAddress{ .ip4 = .loopback(server.port()) };
553 const stream = try address.connect(io, .{ .mode = .stream });
554 defer stream.close(io);
555
556 var write_buffer: [128]u8 = undefined;
557 var writer = stream.writer(io, &write_buffer);
558 try writer.interface.writeAll("GET / HTTP/1.1\r\n");
559 try writer.interface.flush();
560
561 var request_is_active = false;
562 for (0..100) |_| {
563 server.mutex.lock();
564 request_is_active = server.active_client != null;
565 server.mutex.unlock();
566 if (request_is_active) break;
567 compat.sleep(std.time.ns_per_ms);
568 }
569 try std.testing.expect(request_is_active);
570
571 server.stop();
572 try std.testing.expect(!server.running.load(.acquire));
256573}