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

from __future__ import annotations

import logging

import os

import socket

import time

import unittest

from threading import Event

from typing import Any

import pytest

import zeroconf as r

from zeroconf import Zeroconf

from zeroconf._services.info import ServiceInfo

from . import _clear_cache, has_working_ipv6

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

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 = {

"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: dict[str, Any] = {"path": "/~paulsm/"}

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)

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.decoded_properties["prop_none"] is None

assert info.decoded_properties["prop_string"] == b"a_prop".decode("utf-8")

assert info.decoded_properties["prop_float"] == "1.0"

assert info.decoded_properties["prop_blank"] == b"a blanked string".decode("utf-8")

assert info.decoded_properties["prop_true"] == "1"

assert info.decoded_properties["prop_false"] == "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"]

assert info.decoded_properties["prop_blank"] == b"an updated string".decode("utf-8")

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"]

assert cached_info.decoded_properties["prop_blank"] == b"an updated string".decode("utf-8")

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)