13928291307

Committed 18 Mar 2025 04:21PM UTC coverage: 94.049% (-0.3%) from 94.319%

Build # 13928291307

Build Type

Pull #134

github

Committed by

web-flow

Commit Message

Merge 05605cc40 into 1df94ad9c

Pull Request Pull Request #134: fix(migrations): set a proper default value for originating instance migration

Run Details

47 of 52 branches covered (90.38%)

Branch coverage included in aggregate %.

996 of 1057 relevant lines covered (94.23%)

0.94 hits per line

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

76.47

/config/consumers.py

"""Define logic for the WebSocket consumers."""

import logging

from asgiref.sync import sync_to_async
from channels.generic.websocket import AsyncJsonWebsocketConsumer

from scram.route_manager.models import Entry, WebSocketSequenceElement

logger = logging.getLogger(__name__)


class TranslatorConsumer(AsyncJsonWebsocketConsumer):
    """Handle messages from the Translator(s)."""

    async def connect(self):
        """Handle the initial connection with adding to the right group."""
        logger.info("Translator connected")
        self.actiontype = self.scope["url_route"]["kwargs"]["actiontype"]
        self.translator_group = f"translator_{self.actiontype}"

        await self.channel_layer.group_add(self.translator_group, self.channel_name)
        await self.accept()

        # Filter WebSocketSequenceElements by actiontype
        elements = await sync_to_async(list)(
            WebSocketSequenceElement.objects.filter(action_type__name=self.actiontype).order_by("order_num"),
        )
        if not elements:
            logger.warning("No elements found for actiontype=%s.", self.actiontype)
            return

        # Avoid lazy evaluation
        routes = await sync_to_async(list)(Entry.objects.filter(is_active=True).values_list("route__route", flat=True))

        for route in routes:
            for element in elements:
                msg = await sync_to_async(lambda e: e.websocketmessage)(element)
                msg.msg_data[msg.msg_data_route_field] = str(route)
                await self.send_json({"type": msg.msg_type, "message": msg.msg_data})

    async def disconnect(self, close_code):
        """Discard any remaining messages on disconnect."""
        logger.info("Disconnect received: %s", close_code)
        await self.channel_layer.group_discard(self.translator_group, self.channel_name)

    async def receive_json(self, content):
        """Handle a WebSocket message."""
        if content["type"] == "translator_check_resp":
            # We received a check response from a translator, forward to web UI.
            channel = content.pop("channel")
            content["type"] = "wui_check_resp"
            await self.channel_layer.send(channel, content)

    async def _send_event(self, event):
        await self.send_json(event)

    # Tell all translators of this actiontype of an addition of a route.
    translator_add = _send_event
    # Tell all translators of this actiontype of a withdrawal of a route.
    translator_remove = _send_event
    # Tell all translators of this actiontype to withdraw ALL routes.
    translator_remove_all = _send_event
    # Send a query to all translators if a route is announced.
    translator_check = _send_event


class WebUIConsumer(AsyncJsonWebsocketConsumer):
    """Handle messages from the Web UI."""

    async def connect(self):
        """Handle the initial connection with adding to the right group."""
        self.actiontype = self.scope["url_route"]["kwargs"]["actiontype"]
        self.translator_group = f"translator_{self.actiontype}"

        await self.accept()

    async def receive_json(self, content):
        """Receive message from WebSocket."""
        if content["type"] == "wui_check_req":
            # Web UI asks us to check; forward to translator(s)
            await self.channel_layer.group_send(
                self.translator_group,
                {
                    "type": "translator_check",
                    "channel": self.channel_name,
                    "message": content["message"],
                },
            )

    async def wui_check_resp(self, event):
        """Forward a message to the correct Websocket."""
        await self.send_json(event)

1	"""Define logic for the WebSocket consumers."""
2
3	import logging	1✔
4
5	from asgiref.sync import sync_to_async	1✔
6	from channels.generic.websocket import AsyncJsonWebsocketConsumer	1✔
7
8	from scram.route_manager.models import Entry, WebSocketSequenceElement	1✔
9
10	logger = logging.getLogger(__name__)	1✔
11
12
13	class TranslatorConsumer(AsyncJsonWebsocketConsumer):	1✔
14	"""Handle messages from the Translator(s)."""
15
16	async def connect(self):	1✔
17	"""Handle the initial connection with adding to the right group."""
18	logger.info("Translator connected")	1✔
19	self.actiontype = self.scope["url_route"]["kwargs"]["actiontype"]	1✔
20	self.translator_group = f"translator_{self.actiontype}"	1✔
21
22	await self.channel_layer.group_add(self.translator_group, self.channel_name)	1✔
23	await self.accept()	1✔
24
25	# Filter WebSocketSequenceElements by actiontype
26	elements = await sync_to_async(list)(	1✔
27	WebSocketSequenceElement.objects.filter(action_type__name=self.actiontype).order_by("order_num"),
28	)
29	if not elements:	1✔
30	logger.warning("No elements found for actiontype=%s.", self.actiontype)	1✔
31	return	1✔
32
33	# Avoid lazy evaluation
34	routes = await sync_to_async(list)(Entry.objects.filter(is_active=True).values_list("route__route", flat=True))	1✔
35
36	for route in routes:	1✔
37	for element in elements:	×
38	msg = await sync_to_async(lambda e: e.websocketmessage)(element)	×
39	msg.msg_data[msg.msg_data_route_field] = str(route)	×
40	await self.send_json({"type": msg.msg_type, "message": msg.msg_data})	×
41
42	async def disconnect(self, close_code):	1✔
43	"""Discard any remaining messages on disconnect."""
44	logger.info("Disconnect received: %s", close_code)	×
45	await self.channel_layer.group_discard(self.translator_group, self.channel_name)	×
46
47	async def receive_json(self, content):	1✔
48	"""Handle a WebSocket message."""
49	if content["type"] == "translator_check_resp":	×
50	# We received a check response from a translator, forward to web UI.
51	channel = content.pop("channel")	×
52	content["type"] = "wui_check_resp"	×
53	await self.channel_layer.send(channel, content)	×
54
55	async def _send_event(self, event):	1✔
56	await self.send_json(event)	1✔
57
58	# Tell all translators of this actiontype of an addition of a route.
59	translator_add = _send_event	1✔
60	# Tell all translators of this actiontype of a withdrawal of a route.
61	translator_remove = _send_event	1✔
62	# Tell all translators of this actiontype to withdraw ALL routes.
63	translator_remove_all = _send_event	1✔
64	# Send a query to all translators if a route is announced.
65	translator_check = _send_event	1✔
66
67
68	class WebUIConsumer(AsyncJsonWebsocketConsumer):	1✔
69	"""Handle messages from the Web UI."""
70
71	async def connect(self):	1✔
72	"""Handle the initial connection with adding to the right group."""
73	self.actiontype = self.scope["url_route"]["kwargs"]["actiontype"]	1✔
74	self.translator_group = f"translator_{self.actiontype}"	1✔
75
76	await self.accept()	1✔
77
78	async def receive_json(self, content):	1✔
79	"""Receive message from WebSocket."""
80	if content["type"] == "wui_check_req":	1✔
81	# Web UI asks us to check; forward to translator(s)
82	await self.channel_layer.group_send(	1✔
83	self.translator_group,
84	{
85	"type": "translator_check",
86	"channel": self.channel_name,
87	"message": content["message"],
88	},
89	)
90
91	async def wui_check_resp(self, event):	1✔
92	"""Forward a message to the correct Websocket."""
93	await self.send_json(event)	1✔

esnet-security / SCRAM / 13928291307

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