| # Copyright (c) 2014 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import unittest |
| |
| import dpkt |
| import fake_host |
| import socket |
| import zeroconf |
| |
| |
| FAKE_HOSTNAME = 'fakehost1' |
| |
| FAKE_IPADDR = '192.168.11.22' |
| |
| |
| class TestZeroconfDaemon(unittest.TestCase): |
| """Test class for ZeroconfDaemon.""" |
| |
| def setUp(self): |
| self._host = fake_host.FakeHost(FAKE_IPADDR) |
| self._zero = zeroconf.ZeroconfDaemon(self._host, FAKE_HOSTNAME) |
| |
| |
| def _query_A(self, name): |
| """Returns the list of A records matching the given name. |
| |
| @param name: A domain name. |
| @return a list of dpkt.dns.DNS.RR objects, one for each matching record. |
| """ |
| q = dpkt.dns.DNS.Q(name=name, type=dpkt.dns.DNS_A) |
| return self._zero._process_A(q) |
| |
| |
| def testRegisterService(self): |
| """Tests that we get appropriate records after registering a service.""" |
| SERVICE_PORT = 9 |
| SERVICE_TXT_LIST = ['lies=lies'] |
| self._zero.register_service('unique_prefix', '_service_type', |
| '_tcp', SERVICE_PORT, SERVICE_TXT_LIST) |
| name = '_service_type._tcp.local' |
| fq_name = 'unique_prefix.' + name |
| # Issue SRV, PTR, and TXT queries |
| q_srv = dpkt.dns.DNS.Q(name=fq_name, type=dpkt.dns.DNS_SRV) |
| q_txt = dpkt.dns.DNS.Q(name=fq_name, type=dpkt.dns.DNS_TXT) |
| q_ptr = dpkt.dns.DNS.Q(name=name, type=dpkt.dns.DNS_PTR) |
| ptr_responses = self._zero._process_PTR(q_ptr) |
| srv_responses = self._zero._process_SRV(q_srv) |
| txt_responses = self._zero._process_TXT(q_txt) |
| self.assertTrue(ptr_responses) |
| self.assertTrue(srv_responses) |
| self.assertTrue(txt_responses) |
| ptr_resp = ptr_responses[0] |
| srv_resp = [resp for resp in srv_responses |
| if resp.type == dpkt.dns.DNS_SRV][0] |
| txt_resp = txt_responses[0] |
| # Check that basic things are right. |
| self.assertEqual(fq_name, ptr_resp.ptrname) |
| self.assertEqual(FAKE_HOSTNAME + '.' + self._zero.domain, |
| srv_resp.srvname) |
| self.assertEqual(SERVICE_PORT, srv_resp.port) |
| self.assertEqual(SERVICE_TXT_LIST, txt_resp.text) |
| |
| |
| def testProperties(self): |
| """Test the initial properties set by the constructor.""" |
| self.assertEqual(self._zero.host, self._host) |
| self.assertEqual(self._zero.hostname, FAKE_HOSTNAME) |
| self.assertEqual(self._zero.domain, 'local') # Default domain |
| self.assertEqual(self._zero.full_hostname, FAKE_HOSTNAME + '.local') |
| |
| |
| def testSocketInit(self): |
| """Test that the constructor listens for mDNS traffic.""" |
| |
| # Should create an UDP socket and bind it to the mDNS address and port. |
| self.assertEqual(len(self._host._sockets), 1) |
| sock = self._host._sockets[0] |
| |
| self.assertEqual(sock._family, socket.AF_INET) # IPv4 |
| self.assertEqual(sock._sock_type, socket.SOCK_DGRAM) # UDP |
| |
| # Check it is listening for UDP packets on the mDNS address and port. |
| self.assertTrue(sock._bound) |
| self.assertEqual(sock._bind_ip_addr, '224.0.0.251') # mDNS address |
| self.assertEqual(sock._bind_port, 5353) # mDNS port |
| self.assertTrue(callable(sock._bind_recv_callback)) |
| |
| |
| def testRecordsInit(self): |
| """Test the A record of the host is registered.""" |
| host_A = self._query_A(self._zero.full_hostname) |
| self.assertGreater(len(host_A), 0) |
| |
| record = host_A[0] |
| # Check the hostname and the packed IP address. |
| self.assertEqual(record.name, self._zero.full_hostname) |
| self.assertEqual(record.ip, socket.inet_aton(self._host.ip_addr)) |
| |
| |
| def testDoubleTXTProcessing(self): |
| """Test when more than one TXT record is present in a packet. |
| |
| A mDNS packet can include several answer records for several domains and |
| record type. A corner case found on the field presents a mDNS packet |
| with two TXT records for the same domain name on the same packet on its |
| authoritative answers section while the packet itself is a query. |
| """ |
| # Build the mDNS packet with two TXT records. |
| domain_name = 'other_host.local' |
| answers = [ |
| dpkt.dns.DNS.RR( |
| type = dpkt.dns.DNS_TXT, |
| cls = dpkt.dns.DNS_IN, |
| ttl = 120, |
| name = domain_name, |
| text = ['one', 'two']), |
| dpkt.dns.DNS.RR( |
| type = dpkt.dns.DNS_TXT, |
| cls = dpkt.dns.DNS_IN, |
| ttl = 120, |
| name = domain_name, |
| text = ['two'])] |
| # The packet is a query packet, with extra answers on the autoritative |
| # section. |
| mdns = dpkt.dns.DNS( |
| op = dpkt.dns.DNS_QUERY, # Standard query |
| rcode = dpkt.dns.DNS_RCODE_NOERR, |
| q = [], |
| an = [], |
| ns = answers) |
| |
| # Record the new answers received on the answer_calls list. |
| answer_calls = [] |
| self._zero.add_answer_observer(lambda args: answer_calls.extend(args)) |
| |
| # Send the packet to the registered callback. |
| sock = self._host._sockets[0] |
| cbk = sock._bind_recv_callback |
| cbk(str(mdns), '1234', 5353) |
| |
| # Check that the answers callback is called with all the answers in the |
| # received order. |
| self.assertEqual(len(answer_calls), 2) |
| ans1, ans2 = answer_calls # Each ans is a (rrtype, rrname, data) |
| self.assertEqual(ans1[2], ('one', 'two')) |
| self.assertEqual(ans2[2], ('two',)) |
| |
| # Check that the two records were cached. |
| records = self._zero.cached_results(domain_name, dpkt.dns.DNS_TXT) |
| self.assertEqual(len(records), 2) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
