14873191299

Committed 07 May 2025 01:34AM UTC coverage: 93.297% (+0.7%) from 92.587%

Build # 14873191299

Build Type

Pull #1167

github

Committed by

web-flow

Commit Message

Merge 5b34e9b2a into 4a6725f4e

Pull Request Pull Request #1167: Initialized the Websocket Functionality

Run Details

63 of 64 new or added lines in 2 files covered. (98.44%)

3438 of 3685 relevant lines covered (93.3%)

17.67 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.92

/python_ta/reporters/persistent_server.py

"""This module defines a HTML server using aiohttp that supports reloading via WebSockets."""

import asyncio
import logging
import threading
import webbrowser

from aiohttp import WSMsgType, web

LOADING_HTML = b"<h1>Loading report...</h1>"


class PersistentHTMLServer:
    """A persistent HTML server that serves HTML content and supports WebSocket connections."""

    def __init__(self, port: int):
        self.port = port
        self.latest_html = LOADING_HTML
        self.websockets = set()
        self.server_started = False
        self.loop = asyncio.get_event_loop()

    async def handle_report(self, request: web.Request) -> web.Response:
        """Serve the current HTML content at the root endpoint ('/')."""
        return web.Response(body=self.latest_html, content_type="text/html")

    async def handle_websocket(self, request: web.Request) -> web.WebSocketResponse:
        """Handle WebSocket connections, tracking clients and listening for errors."""
        ws = web.WebSocketResponse()
        await ws.prepare(request)

        self.websockets.add(ws)
        try:
            async for msg in ws:
                if msg.type == WSMsgType.ERROR:
                    print(f"WebSocket error: {ws.exception()}")
        finally:
            self.websockets.remove(ws)
        return ws

    async def update_report(self, new_html: bytes) -> None:
        """Update the served HTML content and notify connected WebSocket clients to reload."""
        self.latest_html = new_html

        active = [ws for ws in self.websockets if not ws.closed]
        if active:
            for ws in active:
                await ws.send_str("reload")
        else:
            webbrowser.open(f"http://localhost:{self.port}", new=2)

    def start_server_once(self, initial_html: bytes) -> None:
        """Start the HTML and WebSocket server if not already running, or update content if running."""
        logging.getLogger("aiohttp.access").disabled = True
        if not self.server_started:
            self.server_started = True
            self.loop.create_task(self.run_server())
            self.loop.create_task(self.update_report(initial_html))
            threading.Thread(target=self.loop.run_forever, daemon=True).start()
        else:
            asyncio.run_coroutine_threadsafe(self.update_report(initial_html), self.loop)

    async def run_server(self) -> None:
        """Launch the aiohttp web server on the specified port with routes for HTML and WebSocket."""
        app = web.Application()
        app.router.add_get("/", self.handle_report)
        app.router.add_get("/ws", self.handle_websocket)

        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, "localhost", self.port)
        await site.start()

1	"""This module defines a HTML server using aiohttp that supports reloading via WebSockets."""
2
3	import asyncio	20✔
4	import logging	20✔
5	import threading	20✔
6	import webbrowser	20✔
7
8	from aiohttp import WSMsgType, web	20✔
9
10	LOADING_HTML = b"<h1>Loading report...</h1>"	20✔
11
12
13	class PersistentHTMLServer:	20✔
14	"""A persistent HTML server that serves HTML content and supports WebSocket connections."""
15
16	def __init__(self, port: int):	20✔
17	self.port = port	20✔
18	self.latest_html = LOADING_HTML	20✔
19	self.websockets = set()	20✔
20	self.server_started = False	20✔
21	self.loop = asyncio.get_event_loop()	20✔
22
23	async def handle_report(self, request: web.Request) -> web.Response:	20✔
24	"""Serve the current HTML content at the root endpoint ('/')."""
25	return web.Response(body=self.latest_html, content_type="text/html")	10✔
26
27	async def handle_websocket(self, request: web.Request) -> web.WebSocketResponse:	20✔
28	"""Handle WebSocket connections, tracking clients and listening for errors."""
29	ws = web.WebSocketResponse()	10✔
30	await ws.prepare(request)	10✔
31
32	self.websockets.add(ws)	10✔
33	try:	10✔
34	async for msg in ws:	10✔
NEW 35	if msg.type == WSMsgType.ERROR:	×
36	print(f"WebSocket error: {ws.exception()}")	2✔
37	finally:
38	self.websockets.remove(ws)	10✔
39	return ws	10✔
40
41	async def update_report(self, new_html: bytes) -> None:	20✔
42	"""Update the served HTML content and notify connected WebSocket clients to reload."""
43	self.latest_html = new_html	10✔
44
45	active = [ws for ws in self.websockets if not ws.closed]	10✔
46	if active:	10✔
47	for ws in active:	10✔
48	await ws.send_str("reload")	10✔
49	else:
50	webbrowser.open(f"http://localhost:{self.port}", new=2)	10✔
51
52	def start_server_once(self, initial_html: bytes) -> None:	20✔
53	"""Start the HTML and WebSocket server if not already running, or update content if running."""
54	logging.getLogger("aiohttp.access").disabled = True	10✔
55	if not self.server_started:	10✔
56	self.server_started = True	10✔
57	self.loop.create_task(self.run_server())	10✔
58	self.loop.create_task(self.update_report(initial_html))	10✔
59	threading.Thread(target=self.loop.run_forever, daemon=True).start()	10✔
60	else:
61	asyncio.run_coroutine_threadsafe(self.update_report(initial_html), self.loop)	10✔
62
63	async def run_server(self) -> None:	20✔
64	"""Launch the aiohttp web server on the specified port with routes for HTML and WebSocket."""
65	app = web.Application()	10✔
66	app.router.add_get("/", self.handle_report)	10✔
67	app.router.add_get("/ws", self.handle_websocket)	10✔
68
69	runner = web.AppRunner(app)	10✔
70	await runner.setup()	10✔
71	site = web.TCPSite(runner, "localhost", self.port)	10✔
72	await site.start()	10✔

pyta-uoft / pyta / 14873191299

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous