#!/usr/bin/env python

""" Unit tests for zeroconf._services. """

import logging

import socket

import time

import os

import unittest

from threading import Event

from unittest.mock import patch

import pytest

import zeroconf as r

from zeroconf import Zeroconf

from zeroconf._services.info import ServiceInfo

from . import has_working_ipv6, _clear_cache

log = logging.getLogger('zeroconf')

original_logging_level = logging.NOTSET

def setup_module():

global original_logging_level

original_logging_level = log.level

log.setLevel(logging.DEBUG)

def teardown_module():

if original_logging_level != logging.NOTSET:

log.setLevel(original_logging_level)

class ListenerTest(unittest.TestCase):

def test_integration_with_listener_class(self):

sub_service_added = Event()

service_added = Event()

service_removed = Event()

sub_service_updated = Event()

duplicate_service_added = Event()

subtype_name = "_printer"

type_ = "_http._tcp.local."

subtype = subtype_name + "._sub." + type_

name = "UPPERxxxyyyæøå"

registration_name = f"{name}.{subtype}"

class MyListener(r.ServiceListener):

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()

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

pass

class DuplicateListener(r.ServiceListener):

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

duplicate_service_added.set()

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

pass

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

pass

class MySubListener(r.ServiceListener):

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

sub_service_added.set()

pass

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

pass

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

sub_service_updated.set()

listener = MyListener()

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

zeroconf_browser.add_service_listener(type_, listener)

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(interfaces=['127.0.0.1'])

desc = {'path': '/~paulsm/'} # type: Dict

desc.update(properties)

addresses = [socket.inet_aton("10.0.1.2")]

if has_working_ipv6() and not os.environ.get('SKIP_IPV6'):

addresses.append(socket.inet_pton(socket.AF_INET6, "6001:db8::1"))

addresses.append(socket.inet_pton(socket.AF_INET6, "2001:db8::1"))

info_service = ServiceInfo(

subtype, registration_name, port=80, properties=desc, server="ash-2.local.", addresses=addresses

)

zeroconf_registrar.register_service(info_service)

with patch.object(

zeroconf_registrar.engine.protocols[0], "suppress_duplicate_packet", return_value=False

), patch.object(

zeroconf_registrar.engine.protocols[1], "suppress_duplicate_packet", return_value=False

):

try:

service_added.wait(1)

assert service_added.is_set()

# short pause to allow multicast timers to expire

time.sleep(3)

zeroconf_browser.add_service_listener(type_, DuplicateListener())

duplicate_service_added.wait(

1

) # Ensure a listener for the same type calls back right away from cache

# clear the answer cache to force query

_clear_cache(zeroconf_browser)

cached_info = ServiceInfo(type_, registration_name)

cached_info.load_from_cache(zeroconf_browser)

assert cached_info.properties == {}

# get service info without answer cache

info = zeroconf_browser.get_service_info(type_, registration_name)

assert info is not None

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

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

assert info.properties[b'prop_float'] == b'1.0'

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

assert info.properties[b'prop_true'] == b'1'

assert info.properties[b'prop_false'] == b'0'

assert info.addresses == addresses[:1] # no V6 by default

assert set(info.addresses_by_version(r.IPVersion.All)) == set(addresses)

cached_info = ServiceInfo(type_, registration_name)

cached_info.load_from_cache(zeroconf_browser)

assert cached_info.properties is not None

# Populate the cache

zeroconf_browser.get_service_info(subtype, registration_name)

# get service info with only the cache

cached_info = ServiceInfo(subtype, registration_name)

cached_info.load_from_cache(zeroconf_browser)

assert cached_info.properties is not None

assert cached_info.properties[b'prop_float'] == b'1.0'

# get service info with only the cache with the lowercase name

cached_info = ServiceInfo(subtype, registration_name.lower())

cached_info.load_from_cache(zeroconf_browser)

# Ensure uppercase output is preserved

assert cached_info.name == registration_name

assert cached_info.key == registration_name.lower()

assert cached_info.properties is not None

assert cached_info.properties[b'prop_float'] == b'1.0'

info = zeroconf_browser.get_service_info(subtype, registration_name)

assert info is not None

assert info.properties is not None

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

cached_info = ServiceInfo(subtype, registration_name.lower())

cached_info.load_from_cache(zeroconf_browser)

assert cached_info.properties is not None

assert cached_info.properties[b'prop_none'] is None

# test TXT record update

sublistener = MySubListener()

zeroconf_browser.add_service_listener(subtype, sublistener)

properties['prop_blank'] = b'an updated string'

desc.update(properties)

info_service = ServiceInfo(

subtype,

registration_name,

80,

0,

0,

desc,

"ash-2.local.",

addresses=[socket.inet_aton("10.0.1.2")],

)

zeroconf_registrar.update_service(info_service)

sub_service_added.wait(1) # we cleared the cache above

assert sub_service_added.is_set()

info = zeroconf_browser.get_service_info(type_, registration_name)

assert info is not None

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

cached_info = ServiceInfo(subtype, registration_name)

cached_info.load_from_cache(zeroconf_browser)

assert cached_info.properties is not None

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

zeroconf_registrar.unregister_service(info_service)

service_removed.wait(1)

assert service_removed.is_set()

finally:

zeroconf_registrar.close()

zeroconf_browser.remove_service_listener(listener)

zeroconf_browser.close()

def test_servicelisteners_raise_not_implemented():

"""Verify service listeners raise when one of the methods is not implemented."""

class MyPartialListener(r.ServiceListener):

"""A listener that does not implement anything."""

zc = r.Zeroconf(interfaces=['127.0.0.1'])

with pytest.raises(NotImplementedError):

MyPartialListener().add_service(

zc, "_tivo-videostream._tcp.local.", "Tivo1._tivo-videostream._tcp.local."

)

with pytest.raises(NotImplementedError):

MyPartialListener().remove_service(

zc, "_tivo-videostream._tcp.local.", "Tivo1._tivo-videostream._tcp.local."

)

with pytest.raises(NotImplementedError):

MyPartialListener().update_service(

zc, "_tivo-videostream._tcp.local.", "Tivo1._tivo-videostream._tcp.local."

)

zc.close()

def test_signal_registration_interface():

"""Test adding and removing from the SignalRegistrationInterface."""

interface = r.SignalRegistrationInterface([])

def dummy():

pass

interface.register_handler(dummy)

interface.unregister_handler(dummy)

with pytest.raises(ValueError):

interface.unregister_handler(dummy)