Add SSDPResponse::sendto method and tests (#96)

Author: tibotix

ssdp/messages.py

@@ -53,7 +53,7 @@ def __str__(self):
 
     def __bytes__(self):
         """Return full HTTP message as bytes."""
-        return self.__str__().encode().replace(b"\n", b"\r\n") + b"\r\n\r\n"
+        return self.__str__().encode() + b"\r\n\r\n"
 
 
 class SSDPResponse(SSDPMessage):
@@ -74,6 +74,20 @@ def parse(cls, msg: str):
             version=version, status_code=status_code, reason=reason, headers=headers
         )
 
+    def sendto(self, transport, addr):
+        """
+        Send response to a given address via given transport.
+
+        Args:
+            transport (asyncio.DatagramTransport):
+                Write transport to send the message on.
+            addr (Tuple[str, int]):
+                IP address and port pair to send the message to.
+
+        """
+        logger.debug("%s:%s - - %s", *(addr + (self,)))
+        transport.sendto(bytes(self), addr)
+
     def __str__(self):
         """Return complete SSDP response."""
         lines = list()

tests/test_messages.py

@@ -39,7 +39,7 @@ def test_str(self):
     def test_bytes(self):
         class MyMessage(SSDPMessage):
             def __str__(self):
-                return "NOTIFY * HTTP/1.1\n" "Cache-Control: max-age=3600"
+                return "NOTIFY * HTTP/1.1\r\n" "Cache-Control: max-age=3600"
 
         msg = MyMessage()
         assert bytes(msg) == (
@@ -61,6 +61,16 @@ def test_str(self):
             "HTTP/1.1 200 OK\r\nLocation: http://192.168.1.239:55443"
         )
 
+    def test_sendto(self):
+        transport = Mock()
+        addr = network.MULTICAST_ADDRESS_IPV4, network.PORT
+        SSDPResponse(
+            200, "OK", headers=[("Location", "http://192.168.1.239:55443")]
+        ).sendto(transport, addr)
+        transport.sendto.assert_called_once_with(
+            b"HTTP/1.1 200 OK\r\nLocation: http://192.168.1.239:55443\r\n\r\n", addr
+        )
+
 
 class TestSSDPRequest:
     def test_parse(self):