def test_unknownParams(self):
    """
    Parsing and serializing Via headers with unknown parameters should work.
    """
    s = "SIP/2.0/UDP example.com:5060;branch=a12345b;bogus;pie=delicious"
    v = sip.parseViaHeader(s)
    self.assertEqual(v.toString(), s)

    def deliverResponse(self, responseMessage):
        """Deliver response.
        Destination is based on topmost Via header."""
        destVia = sip.parseViaHeader(responseMessage.headers["via"][0])
        # XXX we don't do multicast yet
        host = destVia.received or destVia.host
        port = destVia.rport or destVia.port or self.PORT

        destAddr = sip.URL(host=host, port=port)
        print str(datetime.now()), responseMessage.toString()
        self.sendMessage(destAddr, responseMessage)

 def testSimple(self):
     s = "SIP/2.0/UDP example.com;hidden"
     v = sip.parseViaHeader(s)
     self.assertEquals(v.transport, "UDP")
     self.assertEquals(v.host, "example.com")
     self.assertEquals(v.port, 5060)
     self.assertEquals(v.ttl, None)
     self.assertEquals(v.maddr, None)
     self.assertEquals(v.branch, None)
     self.assertEquals(v.hidden, 1)
     self.assertEquals(v.toString(),
                       "SIP/2.0/UDP example.com:5060;hidden")
     self.checkRoundtrip(v)

 def testComplex(self):
     s = "SIP/2.0/UDP first.example.com:4000;ttl=16;maddr=224.2.0.1 ;branch=a7c6a8dlze (Example)"
     v = sip.parseViaHeader(s)
     self.assertEquals(v.transport, "UDP")
     self.assertEquals(v.host, "first.example.com")
     self.assertEquals(v.port, 4000)
     self.assertEquals(v.ttl, 16)
     self.assertEquals(v.maddr, "224.2.0.1")
     self.assertEquals(v.branch, "a7c6a8dlze")
     self.assertEquals(v.hidden, 0)
     self.assertEquals(v.toString(),
                       "SIP/2.0/UDP first.example.com:4000;ttl=16;branch=a7c6a8dlze;maddr=224.2.0.1")
     self.checkRoundtrip(v)

 def handle_response(self, rsp, addr):
     print "Received SIP response from %s:%d:" % addr
     print rsp.toString()
     # Strip the top via
     rsp.headers["via"].pop(0)
     # Determine the next hop from the next via
     via = sip.parseViaHeader(rsp.headers["via"][0])
     # rport handling is a bit mysterious - rport should be filled
     # in by Twisted, but apparently it's not. This seems to work
     # for now, though, presumably by defaulting to port 5060.
     uri = sip.URL(host=via.received or via.host, port=via.rportValue if via.rportRequested else self.PORT)
     print "Sending SIP response to %s:" % uri
     print rsp.toString()
     print "==============================================================================="
     # Send the message on
     self.sendMessage(uri, rsp)

 def test_simple(self):
     """
     Test parsing a simple Via header.
     """
     s = "SIP/2.0/UDP example.com;hidden"
     v = sip.parseViaHeader(s)
     self.assertEqual(v.transport, "UDP")
     self.assertEqual(v.host, "example.com")
     self.assertEqual(v.port, 5060)
     self.assertEqual(v.rport, None)
     self.assertEqual(v.rportValue, None)
     self.assertEqual(v.rportRequested, False)
     self.assertEqual(v.ttl, None)
     self.assertEqual(v.maddr, None)
     self.assertEqual(v.branch, None)
     self.assertEqual(v.hidden, True)
     self.assertEqual(v.toString(), "SIP/2.0/UDP example.com:5060;hidden")
     self.checkRoundtrip(v)

 def test_complex(self):
     """
     Test parsing a Via header with one of everything.
     """
     s = "SIP/2.0/UDP first.example.com:4000;ttl=16;maddr=224.2.0.1" " ;branch=a7c6a8dlze (Example)"
     v = sip.parseViaHeader(s)
     self.assertEqual(v.transport, "UDP")
     self.assertEqual(v.host, "first.example.com")
     self.assertEqual(v.port, 4000)
     self.assertEqual(v.rport, None)
     self.assertEqual(v.rportValue, None)
     self.assertEqual(v.rportRequested, False)
     self.assertEqual(v.ttl, 16)
     self.assertEqual(v.maddr, "224.2.0.1")
     self.assertEqual(v.branch, "a7c6a8dlze")
     self.assertEqual(v.hidden, 0)
     self.assertEqual(v.toString(), "SIP/2.0/UDP first.example.com:4000" ";ttl=16;branch=a7c6a8dlze;maddr=224.2.0.1")
     self.checkRoundtrip(v)

 def handle_request(self, req, addr):
     print "Received SIP request from %s:%d:" % addr
     print req.toString()
     # Add a via header for us
     via = self.getVia()
     if not via.branch:
         # Twisted SIP is still not compliant with RFC3261 s16.11.
         # Work around this, assuming our caller *is* compliant.
         topVia = sip.parseViaHeader(req.headers["via"][0])
         via.branch = topVia.branch + "-AS-059"  # chosen by fair random.org. guaranteed to be random.
     req.headers["via"].insert(0, via.toString())
     # Determine the next hop - use the route headers if present, else the request URI
     uri = req.uri
     if "route" in req.headers:
         # If we're the first route header, strip it
         _, route, _ = sip.parseAddress(req.headers["route"][0])
         if route.host == self.host:
             req.headers["route"].pop(0)
         # Now find the next hop
         if len(req.headers["route"]) > 0:
             _, uri, _ = sip.parseAddress(req.headers["route"][0])
     # Insert a record-route header to keep us in the path
     req.headers.setdefault("record-route", []).insert(
         0, sip.URL(host=self.host, port=self.port, other=["transport=UDP", "lr"])
     )
     # Assert that we're authorized. Workaround pending sto131 completion.
     if "authorization" not in req.headers:
         req.addHeader(
             "authorization",
             'Digest username="[email protected]", realm="cw-ngv.com", nonce="0000000000000000", uri="sip:cw-ngv.com", response="00000000000000000000000000000000", algorithm=md5, opaque="0000000000000000",integrity-protected="yes"',
         )
     print "Sending SIP request to %s:" % uri
     print req.toString()
     print "==============================================================================="
     # Send the message on
     self.sendMessage(uri, req)

 def testExtraWhitespace(self):
     v1 = sip.parseViaHeader('SIP/2.0/UDP 192.168.1.1:5060')
     v2 = sip.parseViaHeader('SIP/2.0/UDP     192.168.1.1:5060')
     self.assertEqual(v1.transport, v2.transport)
     self.assertEqual(v1.host, v2.host)
     self.assertEqual(v1.port, v2.port)

 def checkRoundtrip(self, v):
     s = v.toString()
     self.assertEqual(s, sip.parseViaHeader(s).toString())