#!/usr/bin/env python

# -*- coding: utf-8 -*-

""" Unit tests for zeroconf.py """

import logging

import socket

import struct

import time

import unittest

from threading import Event

from six import indexbytes

from six.moves import xrange

import zeroconf as r

from zeroconf import (

DNSHinfo,

DNSText,

ServiceBrowser,

ServiceInfo,

ServiceStateChange,

Zeroconf,

ZeroconfServiceTypes,

)

log = logging.getLogger('zeroconf')

original_logging_level = [None]

def setup_module():

original_logging_level[0] = log.level

log.setLevel(logging.DEBUG)

def teardown_module():

log.setLevel(original_logging_level[0])

class PacketGeneration(unittest.TestCase):

def test_parse_own_packet_simple(self):

generated = r.DNSOutgoing(0)

r.DNSIncoming(generated.packet())

def test_parse_own_packet_simple_unicast(self):

generated = r.DNSOutgoing(0, 0)

r.DNSIncoming(generated.packet())

def test_parse_own_packet_flags(self):

generated = r.DNSOutgoing(r._FLAGS_QR_QUERY)

r.DNSIncoming(generated.packet())

def test_parse_own_packet_question(self):

generated = r.DNSOutgoing(r._FLAGS_QR_QUERY)

generated.add_question(r.DNSQuestion("testname.local.", r._TYPE_SRV,

r._CLASS_IN))

r.DNSIncoming(generated.packet())

def test_match_question(self):

generated = r.DNSOutgoing(r._FLAGS_QR_QUERY)

question = r.DNSQuestion("testname.local.", r._TYPE_SRV, r._CLASS_IN)

generated.add_question(question)

parsed = r.DNSIncoming(generated.packet())

self.assertEqual(len(generated.questions), 1)

self.assertEqual(len(generated.questions), len(parsed.questions))

self.assertEqual(question, parsed.questions[0])

def test_dns_hinfo(self):

generated = r.DNSOutgoing(0)

generated.add_additional_answer(

DNSHinfo('irrelevant', r._TYPE_HINFO, 0, 0, 'cpu', 'os'))

parsed = r.DNSIncoming(generated.packet())

self.assertEqual(parsed.answers[0].cpu, u'cpu')

self.assertEqual(parsed.answers[0].os, u'os')

generated = r.DNSOutgoing(0)

generated.add_additional_answer(

DNSHinfo('irrelevant', r._TYPE_HINFO, 0, 0, 'cpu', 'x' * 257))

self.assertRaises(r.NamePartTooLongException, generated.packet)

class PacketForm(unittest.TestCase):

def test_transaction_id(self):

"""ID must be zero in a DNS-SD packet"""

generated = r.DNSOutgoing(r._FLAGS_QR_QUERY)

bytes = generated.packet()

id = indexbytes(bytes, 0) << 8 | indexbytes(bytes, 1)

self.assertEqual(id, 0)

def test_query_header_bits(self):

generated = r.DNSOutgoing(r._FLAGS_QR_QUERY)

bytes = generated.packet()

flags = indexbytes(bytes, 2) << 8 | indexbytes(bytes, 3)

self.assertEqual(flags, 0x0)

def test_response_header_bits(self):

generated = r.DNSOutgoing(r._FLAGS_QR_RESPONSE)

bytes = generated.packet()

flags = indexbytes(bytes, 2) << 8 | indexbytes(bytes, 3)

self.assertEqual(flags, 0x8000)

def test_numbers(self):

generated = r.DNSOutgoing(r._FLAGS_QR_RESPONSE)

bytes = generated.packet()

(numQuestions, numAnswers, numAuthorities,

numAdditionals) = struct.unpack('!4H', bytes[4:12])

self.assertEqual(numQuestions, 0)

self.assertEqual(numAnswers, 0)

self.assertEqual(numAuthorities, 0)

self.assertEqual(numAdditionals, 0)

def test_numbers_questions(self):

generated = r.DNSOutgoing(r._FLAGS_QR_RESPONSE)

question = r.DNSQuestion("testname.local.", r._TYPE_SRV, r._CLASS_IN)

for i in xrange(10):

generated.add_question(question)

bytes = generated.packet()

(numQuestions, numAnswers, numAuthorities,

numAdditionals) = struct.unpack('!4H', bytes[4:12])

self.assertEqual(numQuestions, 10)

self.assertEqual(numAnswers, 0)

self.assertEqual(numAuthorities, 0)

self.assertEqual(numAdditionals, 0)

class Names(unittest.TestCase):

def test_long_name(self):

generated = r.DNSOutgoing(r._FLAGS_QR_RESPONSE)

question = r.DNSQuestion("this.is.a.very.long.name.with.lots.of.parts.in.it.local.",

r._TYPE_SRV, r._CLASS_IN)

generated.add_question(question)

r.DNSIncoming(generated.packet())

def test_exceedingly_long_name(self):

generated = r.DNSOutgoing(r._FLAGS_QR_RESPONSE)

name = "%slocal." % ("part." * 1000)

question = r.DNSQuestion(name, r._TYPE_SRV, r._CLASS_IN)

generated.add_question(question)

r.DNSIncoming(generated.packet())

def test_exceedingly_long_name_part(self):

name = "%s.local." % ("a" * 1000)

generated = r.DNSOutgoing(r._FLAGS_QR_RESPONSE)

question = r.DNSQuestion(name, r._TYPE_SRV, r._CLASS_IN)

generated.add_question(question)

self.assertRaises(r.NamePartTooLongException, generated.packet)

def test_same_name(self):

name = "paired.local."

generated = r.DNSOutgoing(r._FLAGS_QR_RESPONSE)

question = r.DNSQuestion(name, r._TYPE_SRV, r._CLASS_IN)

generated.add_question(question)

generated.add_question(question)

r.DNSIncoming(generated.packet())

class Framework(unittest.TestCase):

def test_launch_and_close(self):

rv = r.Zeroconf()

rv.close()

class Exceptions(unittest.TestCase):

browser = None

@classmethod

def setUpClass(cls):

cls.browser = Zeroconf()

@classmethod

def tearDownClass(cls):

cls.browser.close()

cls.browser = None

def test_bad_service_info_name(self):

self.assertRaises(

r.BadTypeInNameException,

self.browser.get_service_info, "type", "type_not")

def test_bad_service_names(self):

bad_names_to_try = (

'',

'local',

'_tcp.local.',

'_udp.local.',

'._udp.local.',

'_@._tcp.local.',

'_A@._tcp.local.',

'_x--x._tcp.local.',

'_-x._udp.local.',

'_x-._tcp.local.',

'_22._udp.local.',

'_2-2._tcp.local.',

'_1234567890-abcde._udp.local.',

'._x._udp.local.',

)

for name in bad_names_to_try:

self.assertRaises(

r.BadTypeInNameException,

self.browser.get_service_info, name, 'x.' + name)

def test_bad_sub_types(self):

bad_names_to_try = (

'_sub._http._tcp.local.',

'x.sub._http._tcp.local.',

'a' * 64 + '._sub._http._tcp.local.',

'a' * 62 + u'â._sub._http._tcp.local.',

)

for name in bad_names_to_try:

self.assertRaises(

r.BadTypeInNameException, r.service_type_name, name)

def test_good_service_names(self):

good_names_to_try = (

'_x._tcp.local.',

'_x._udp.local.',

'_12345-67890-abc._udp.local.',

'x._sub._http._tcp.local.',

'a' * 63 + '._sub._http._tcp.local.',

'a' * 61 + u'â._sub._http._tcp.local.',

)

for name in good_names_to_try:

r.service_type_name(name)

class ServiceTypesQuery(unittest.TestCase):

def test_integration_with_listener(self):

type_ = "_test-srvc-type._tcp.local."

name = "xxxyyy"

registration_name = "%s.%s" % (name, type_)

zeroconf_registrar = Zeroconf(interfaces=['127.0.0.1'])

desc = {'path': '/~paulsm/'}

info = ServiceInfo(

type_, registration_name,

socket.inet_aton("10.0.1.2"), 80, 0, 0,

desc, "ash-2.local.")

zeroconf_registrar.register_service(info)

try:

service_types = ZeroconfServiceTypes.find(timeout=0.5)

assert type_ in service_types

service_types = ZeroconfServiceTypes.find(

zc=zeroconf_registrar, timeout=0.5)

assert type_ in service_types

finally:

zeroconf_registrar.close()

def test_integration_with_subtype_and_listener(self):

subtype_ = "_subtype._sub"

type_ = "_type._tcp.local."

name = "xxxyyy"

# Note: discovery returns only DNS-SD type not subtype

discovery_type = "%s.%s" % (subtype_, type_)

registration_name = "%s.%s" % (name, type_)

zeroconf_registrar = Zeroconf(interfaces=['127.0.0.1'])

desc = {'path': '/~paulsm/'}

info = ServiceInfo(

discovery_type, registration_name,

socket.inet_aton("10.0.1.2"), 80, 0, 0,

desc, "ash-2.local.")

zeroconf_registrar.register_service(info)

try:

service_types = ZeroconfServiceTypes.find(timeout=0.5)

print(service_types)

assert discovery_type in service_types

service_types = ZeroconfServiceTypes.find(

zc=zeroconf_registrar, timeout=0.5)

assert discovery_type in service_types

finally:

zeroconf_registrar.close()

class ListenerTest(unittest.TestCase):

def test_integration_with_listener_class(self):

service_added = Event()

service_removed = Event()

subtype_name = "My special Subtype"

type_ = "_http._tcp.local."

subtype = subtype_name + "._sub." + type_

name = "xxxyyy"

registration_name = "%s.%s" % (name, type_)

class MyListener(object):

def add_service(self, zeroconf, type, name):

zeroconf.get_service_info(type, name)

service_added.set()

def remove_service(self, zeroconf, type, name):

service_removed.set()

zeroconf_browser = Zeroconf()

zeroconf_browser.add_service_listener(subtype, MyListener())

properties = dict(

prop_none=None,

prop_string=b'a_prop',

prop_float=1.0,

prop_blank=b'a blanked string',

prop_true=1,

prop_false=0,

)

zeroconf_registrar = Zeroconf()

desc = {'path': '/~paulsm/'}

desc.update(properties)

info_service = ServiceInfo(

subtype, registration_name,

socket.inet_aton("10.0.1.2"), 80, 0, 0,

desc, "ash-2.local.")

zeroconf_registrar.register_service(info_service)

try:

service_added.wait(1)

assert service_added.is_set()

# short pause to allow multicast timers to expire

time.sleep(2)

# clear the answer cache to force query

for record in zeroconf_browser.cache.entries():

zeroconf_browser.cache.remove(record)

# get service info without answer cache

info = zeroconf_browser.get_service_info(type_, registration_name)

assert info.properties[b'prop_none'] is False

assert info.properties[b'prop_string'] == properties['prop_string']

assert info.properties[b'prop_float'] is False

assert info.properties[b'prop_blank'] == properties['prop_blank']

assert info.properties[b'prop_true'] is True

assert info.properties[b'prop_false'] is False

info = zeroconf_browser.get_service_info(subtype, registration_name)

assert info.properties[b'prop_none'] is False

zeroconf_registrar.unregister_service(info_service)

service_removed.wait(1)

assert service_removed.is_set()

finally:

zeroconf_registrar.close()

zeroconf_browser.close()

def test_integration():

service_added = Event()

service_removed = Event()

type_ = "_http._tcp.local."

registration_name = "xxxyyy.%s" % type_

def on_service_state_change(zeroconf, service_type, state_change, name):

if name == registration_name:

if state_change is ServiceStateChange.Added:

service_added.set()

elif state_change is ServiceStateChange.Removed:

service_removed.set()

zeroconf_browser = Zeroconf()

browser = ServiceBrowser(zeroconf_browser, type_, [on_service_state_change])

zeroconf_registrar = Zeroconf()

desc = {'path': '/~paulsm/'}

info = ServiceInfo(

type_, registration_name,

socket.inet_aton("10.0.1.2"), 80, 0, 0,

desc, "ash-2.local.")

zeroconf_registrar.register_service(info)

try:

service_added.wait(1)

assert service_added.is_set()

# Don't remove service, allow close() to cleanup

finally:

zeroconf_registrar.close()

browser.cancel()

zeroconf_browser.close()

def test_dnstext_repr_works():

# There was an issue on Python 3 that prevented DNSText's repr

# from working when the text was longer than 10 bytes

text = DNSText('irrelevant', None, 0, 0, b'12345678901')

repr(text)