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

from __future__ import annotations

import logging

import socket

import time

import unittest.mock

from unittest.mock import patch

import zeroconf as r

from zeroconf import ServiceInfo, Zeroconf, const

from . import _inject_responses

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 Names(unittest.TestCase):

def test_long_name(self):

generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)

question = r.DNSQuestion(

"this.is.a.very.long.name.with.lots.of.parts.in.it.local.",

const._TYPE_SRV,

const._CLASS_IN,

)

generated.add_question(question)

r.DNSIncoming(generated.packets()[0])

def test_exceedingly_long_name(self):

generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)

name = f"{'part.' * 1000}local."

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

generated.add_question(question)

r.DNSIncoming(generated.packets()[0])

def test_extra_exceedingly_long_name(self):

generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)

name = f"{'part.' * 4000}local."

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

generated.add_question(question)

r.DNSIncoming(generated.packets()[0])

def test_exceedingly_long_name_part(self):

name = f"{'a' * 1000}.local."

generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)

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

generated.add_question(question)

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

def test_same_name(self):

name = "paired.local."

generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)

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

generated.add_question(question)

generated.add_question(question)

r.DNSIncoming(generated.packets()[0])

def test_verify_name_change_with_lots_of_names(self):

# instantiate a zeroconf instance

zc = Zeroconf(interfaces=["127.0.0.1"])

# create a bunch of servers

type_ = "_my-service._tcp.local."

name = "a wonderful service"

server_count = 300

self.generate_many_hosts(zc, type_, name, server_count)

# verify that name changing works

self.verify_name_change(zc, type_, name, server_count)

zc.close()

def test_large_packet_exception_log_handling(self):

"""Verify we downgrade debug after warning."""

# instantiate a zeroconf instance

zc = Zeroconf(interfaces=["127.0.0.1"])

with (

patch("zeroconf._logger.log.warning") as mocked_log_warn,

patch("zeroconf._logger.log.debug") as mocked_log_debug,

):

# now that we have a long packet in our possession, let's verify the

# exception handling.

out = r.DNSOutgoing(const._FLAGS_QR_RESPONSE | const._FLAGS_AA)

out.data.append(b"\0" * 10000)

# mock the zeroconf logger and check for the correct logging backoff

call_counts = mocked_log_warn.call_count, mocked_log_debug.call_count

# try to send an oversized packet

zc.send(out)

assert mocked_log_warn.call_count == call_counts[0]

zc.send(out)

assert mocked_log_warn.call_count == call_counts[0]

# mock the zeroconf logger and check for the correct logging backoff

call_counts = mocked_log_warn.call_count, mocked_log_debug.call_count

# force receive on oversized packet

zc.send(out, const._MDNS_ADDR, const._MDNS_PORT)

zc.send(out, const._MDNS_ADDR, const._MDNS_PORT)

time.sleep(0.3)

r.log.debug(

"warn %d debug %d was %s",

mocked_log_warn.call_count,

mocked_log_debug.call_count,

call_counts,

)

assert mocked_log_debug.call_count > call_counts[0]

# close our zeroconf which will close the sockets

zc.close()

def verify_name_change(self, zc, type_, name, number_hosts):

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

info_service = ServiceInfo(

type_,

f"{name}.{type_}",

80,

0,

0,

desc,

"ash-2.local.",

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

)

# verify name conflict

self.assertRaises(r.NonUniqueNameException, zc.register_service, info_service)

# verify no name conflict https://tools.ietf.org/html/rfc6762#section-6.6

zc.register_service(info_service, cooperating_responders=True)

# Create a new object since allow_name_change will mutate the

# original object and then we will have the wrong service

# in the registry

info_service2 = ServiceInfo(

type_,

f"{name}.{type_}",

80,

0,

0,

desc,

"ash-2.local.",

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

)

zc.register_service(info_service2, allow_name_change=True)

assert info_service2.name.split(".")[0] == f"{name}-{number_hosts + 1}"

def generate_many_hosts(self, zc, type_, name, number_hosts):

block_size = 25

number_hosts = int((number_hosts - 1) / block_size + 1) * block_size

out = r.DNSOutgoing(const._FLAGS_QR_RESPONSE | const._FLAGS_AA)

for i in range(1, number_hosts + 1):

next_name = name if i == 1 else f"{name}-{i}"

self.generate_host(out, next_name, type_)

_inject_responses(zc, [r.DNSIncoming(packet) for packet in out.packets()])

@staticmethod

def generate_host(out, host_name, type_):

name = ".".join((host_name, type_))

out.add_answer_at_time(

r.DNSPointer(type_, const._TYPE_PTR, const._CLASS_IN, const._DNS_OTHER_TTL, name),

0,

)

out.add_answer_at_time(

r.DNSService(

type_,

const._TYPE_SRV,

const._CLASS_IN | const._CLASS_UNIQUE,

const._DNS_HOST_TTL,

0,

0,

80,

name,

),

0,

)