test backend typing packets

backend/src/network_sniffer.py

@@ -7,7 +7,7 @@ import socket
 import selectors
 import errno
 import struct
-from typing import Dict, List, Optional, Any
+from typing import Dict, List, Optional, Any, TypedDict, Union
 from uuid import uuid4
 
 # NOTE: ensure this path points to your shared runtime module
@@ -50,8 +50,33 @@ sessions: Dict[str, Dict[str, Any]] = {}
 #   "ports": List[str],
 # }
 
+# -------------------------
+# PacketInfo typing
+# -------------------------
+class PacketInfo(TypedDict, total=False):
+    """
+    TypedDict for the parsed packet info produced by parse_packet.
+    Fields marked optional (total=False) for flexibility across contexts.
+    """
+    iface: str
+    length: int
+    raw: bytes  # original raw bytes (kept for buffering; DB helper may convert to base64)
+    src_mac: Optional[str]
+    dst_mac: Optional[str]
+    eth_type_raw: Optional[int]
+    eth_type: Optional[Union[str, EtherTypeEnum]]
+    vlan_id: Optional[int]
+    protocol_raw: Optional[int]
+    protocol: Optional[Union[str, IPProtocolEnum, int]]
+    protocol_name: Optional[str]
+    src_ip: Optional[str]
+    dst_ip: Optional[str]
+    src_port: Optional[int]
+    dst_port: Optional[int]
+
+
 # small bounded buffer for packets produced before shared_objects is ready
-_PACKET_BUFFER: List[Dict[str, Any]] = []
+_PACKET_BUFFER: List[PacketInfo] = []
 _BUFFER_CAPACITY = 20000
 
 # Background asyncio loop used for internal tasks in this module (kept but not used for DB pool)
@@ -113,7 +138,7 @@ def _safe_get_attr(layer, attr: str):
 
 def parse_packet(pkt, bridge_label: str) -> None:
     """
-    Parse a scapy Packet object into a normalized dict and schedule DB insert.
+    Parse a scapy Packet object into a normalized PacketInfo and schedule DB insert.
     bridge_label indicates whether the packet was captured as part of a bridge-snapshot or single-interface.
     """
     pkt_iface = getattr(pkt, "sniffed_on", None)
@@ -122,7 +147,7 @@ def parse_packet(pkt, bridge_label: str) -> None:
 
     logger.debug("Packet captured on %s (bridge_label %s)", pkt_iface, bridge_label)
 
-    pkt_info: Dict[str, Any] = {
+    pkt_info: PacketInfo = {
         "iface": pkt_iface,
         "length": len(pkt),
         "raw": bytes(pkt),
@@ -210,7 +235,7 @@ def parse_packet(pkt, bridge_label: str) -> None:
         elif proto_num == 1 and ICMP in pkt:
             pkt_info["protocol_name"] = "ICMP"
         else:
-            if pkt_info["protocol_name"] is None:
+            if pkt_info.get("protocol_name") is None:
                 pkt_info["protocol_name"] = f"IP_PROTO_{proto_num}" if proto_num is not None else None
 
     # IPv6
@@ -242,11 +267,11 @@ def parse_packet(pkt, bridge_label: str) -> None:
         elif ICMPv6Unknown in pkt:
             pkt_info["protocol_name"] = "ICMPv6"
         else:
-            if pkt_info["protocol_name"] is None:
+            if pkt_info.get("protocol_name") is None:
                 pkt_info["protocol_name"] = f"IPV6_PROTO_{nh}" if nh is not None else None
 
     # Raw fallback label
-    if Raw in pkt and not pkt_info["protocol_name"]:
+    if Raw in pkt and not pkt_info.get("protocol_name"):
         pkt_info["protocol_name"] = "RAW"
 
     # Submit DB insert to shared web loop if available, otherwise buffer