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

from __future__ import annotations

import asyncio
import logging
import os
import socket
import threading
import unittest
from collections.abc import Iterable
from ipaddress import ip_address
from threading import Event
from unittest.mock import patch

import pytest

import zeroconf as r
from zeroconf import DNSAddress, RecordUpdate, const
from zeroconf._protocol.outgoing import DNSOutgoing
from zeroconf._services import info
from zeroconf._services.info import ServiceInfo
from zeroconf._utils.net import IPVersion
from zeroconf.asyncio import AsyncZeroconf

from .. import _inject_response, 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 TestServiceInfo(unittest.TestCase):
    def test_get_name(self):
        """Verify the name accessor can strip the type."""
        desc = {"path": "/~paulsm/"}
        service_name = "name._type._tcp.local."
        service_type = "_type._tcp.local."
        service_server = "ash-1.local."
        service_address = socket.inet_aton("10.0.1.2")
        info = ServiceInfo(
            service_type,
            service_name,
            22,
            0,
            0,
            desc,
            service_server,
            addresses=[service_address],
        )
        assert info.get_name() == "name"

    def test_service_info_rejects_non_matching_updates(self):
        """Verify records with the wrong name are rejected."""

        zc = r.Zeroconf(interfaces=["127.0.0.1"])
        desc = {"path": "/~paulsm/"}
        service_name = "name._type._tcp.local."
        service_type = "_type._tcp.local."
        service_server = "ash-1.local."
        service_address = socket.inet_aton("10.0.1.2")
        ttl = 120
        now = r.current_time_millis()
        info = ServiceInfo(
            service_type,
            service_name,
            22,
            0,
            0,
            desc,
            service_server,
            addresses=[service_address],
        )
        # Verify backwards compatibility with calling with None
        info.async_update_records(zc, now, [])
        # Matching updates
        info.async_update_records(
            zc,
            now,
            [
                RecordUpdate(
                    r.DNSText(
                        service_name,
                        const._TYPE_TXT,
                        const._CLASS_IN | const._CLASS_UNIQUE,
                        ttl,
                        b"\x04ff=0\x04ci=2\x04sf=0\x0bsh=6fLM5A==",
                    ),
                    None,
                )
            ],
        )
        assert info.properties[b"ci"] == b"2"
        info.async_update_records(
            zc,
            now,
            [
                RecordUpdate(
                    r.DNSService(
                        service_name,
                        const._TYPE_SRV,
                        const._CLASS_IN | const._CLASS_UNIQUE,
                        ttl,
                        0,
                        0,
                        80,
                        "ASH-2.local.",
                    ),
                    None,
                )
            ],
        )
        assert info.server_key == "ash-2.local."
        assert info.server == "ASH-2.local."
        new_address = socket.inet_aton("10.0.1.3")
        info.async_update_records(
            zc,
            now,
            [
                RecordUpdate(
                    r.DNSAddress(
                        "ASH-2.local.",
                        const._TYPE_A,
                        const._CLASS_IN | const._CLASS_UNIQUE,
                        ttl,
                        new_address,
                    ),
                    None,
                )
            ],
        )
        assert new_address in info.addresses
        # Non-matching updates
        info.async_update_records(
            zc,
            now,
            [
                RecordUpdate(
                    r.DNSText(
                        "incorrect.name.",
                        const._TYPE_TXT,
                        const._CLASS_IN | const._CLASS_UNIQUE,
                        ttl,
                        b"\x04ff=0\x04ci=3\x04sf=0\x0bsh=6fLM5A==",
                    ),
                    None,
                )
            ],
        )
        assert info.properties[b"ci"] == b"2"
        info.async_update_records(
            zc,
            now,
            [
                RecordUpdate(
                    r.DNSService(
                        "incorrect.name.",
                        const._TYPE_SRV,
                        const._CLASS_IN | const._CLASS_UNIQUE,
                        ttl,
                        0,
                        0,
                        80,
                        "ASH-2.local.",
                    ),
                    None,
                )
            ],
        )
        assert info.server_key == "ash-2.local."
        assert info.server == "ASH-2.local."
        new_address = socket.inet_aton("10.0.1.4")
        info.async_update_records(
            zc,
            now,
            [
                RecordUpdate(
                    r.DNSAddress(
                        "incorrect.name.",
                        const._TYPE_A,
                        const._CLASS_IN | const._CLASS_UNIQUE,
                        ttl,
                        new_address,
                    ),
                    None,
                )
            ],
        )
        assert new_address not in info.addresses
        zc.close()

    def test_service_info_rejects_expired_records(self):
        """Verify records that are expired are rejected."""
        zc = r.Zeroconf(interfaces=["127.0.0.1"])
        desc = {"path": "/~paulsm/"}
        service_name = "name._type._tcp.local."
        service_type = "_type._tcp.local."
        service_server = "ash-1.local."
        service_address = socket.inet_aton("10.0.1.2")
        ttl = 120
        now = r.current_time_millis()
        info = ServiceInfo(
            service_type,
            service_name,
            22,
            0,
            0,
            desc,
            service_server,
            addresses=[service_address],
        )
        # Matching updates
        info.async_update_records(
            zc,
            now,
            [
                RecordUpdate(
                    r.DNSText(
                        service_name,
                        const._TYPE_TXT,
                        const._CLASS_IN | const._CLASS_UNIQUE,
                        ttl,
                        b"\x04ff=0\x04ci=2\x04sf=0\x0bsh=6fLM5A==",
                    ),
                    None,
                )
            ],
        )
        assert info.properties[b"ci"] == b"2"
        # Expired record
        expired_record = r.DNSText(
            service_name,
            const._TYPE_TXT,
            const._CLASS_IN | const._CLASS_UNIQUE,
            ttl,
            b"\x04ff=0\x04ci=3\x04sf=0\x0bsh=6fLM5A==",
        )
        zc.cache._async_set_created_ttl(expired_record, 1000, 1)
        info.async_update_records(zc, now, [RecordUpdate(expired_record, None)])
        assert info.properties[b"ci"] == b"2"
        zc.close()

    @unittest.skipIf(not has_working_ipv6(), "Requires IPv6")
    @unittest.skipIf(os.environ.get("SKIP_IPV6"), "IPv6 tests disabled")
    def test_get_info_partial(self):
        zc = r.Zeroconf(interfaces=["127.0.0.1"])

        service_name = "name._type._tcp.local."
        service_type = "_type._tcp.local."
        service_server = "ash-1.local."
        service_text = b"path=/~matt1/"
        service_address = "10.0.1.2"
        service_address_v6_ll = "fe80::52e:c2f2:bc5f:e9c6"
        service_scope_id = 12

        service_info = None
        send_event = Event()
        service_info_event = Event()

        last_sent: r.DNSOutgoing | None = None

        def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT, v6_flow_scope=()):
            """Sends an outgoing packet."""
            nonlocal last_sent

            last_sent = out
            send_event.set()

        # patch the zeroconf send
        with patch.object(zc, "async_send", send):

            def mock_incoming_msg(records: Iterable[r.DNSRecord]) -> r.DNSIncoming:
                generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)

                for record in records:
                    generated.add_answer_at_time(record, 0)

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

            def get_service_info_helper(zc, type, name):
                nonlocal service_info
                service_info = zc.get_service_info(type, name)
                service_info_event.set()

            try:
                ttl = 120
                helper_thread = threading.Thread(
                    target=get_service_info_helper,
                    args=(zc, service_type, service_name),
                )
                helper_thread.start()
                wait_time = 1

                # Expect query for SRV, TXT, A, AAAA
                send_event.wait(wait_time)
                assert last_sent is not None
                assert len(last_sent.questions) == 4
                assert r.DNSQuestion(service_name, const._TYPE_SRV, const._CLASS_IN) in last_sent.questions
                assert r.DNSQuestion(service_name, const._TYPE_TXT, const._CLASS_IN) in last_sent.questions
                assert r.DNSQuestion(service_name, const._TYPE_A, const._CLASS_IN) in last_sent.questions
                assert r.DNSQuestion(service_name, const._TYPE_AAAA, const._CLASS_IN) in last_sent.questions
                assert service_info is None

                # Expect query for SRV, A, AAAA
                last_sent = None
                send_event.clear()
                _inject_response(
                    zc,
                    mock_incoming_msg(
                        [
                            r.DNSText(
                                service_name,
                                const._TYPE_TXT,
                                const._CLASS_IN | const._CLASS_UNIQUE,
                                ttl,
                                service_text,
                            )
                        ]
                    ),
                )
                send_event.wait(wait_time)
                assert last_sent is not None
                assert len(last_sent.questions) == 3  # type: ignore[unreachable]
                assert r.DNSQuestion(service_name, const._TYPE_SRV, const._CLASS_IN) in last_sent.questions
                assert r.DNSQuestion(service_name, const._TYPE_A, const._CLASS_IN) in last_sent.questions
                assert r.DNSQuestion(service_name, const._TYPE_AAAA, const._CLASS_IN) in last_sent.questions
                assert service_info is None

                # Expect query for A, AAAA
                last_sent = None
                send_event.clear()
                _inject_response(
                    zc,
                    mock_incoming_msg(
                        [
                            r.DNSService(
                                service_name,
                                const._TYPE_SRV,
                                const._CLASS_IN | const._CLASS_UNIQUE,
                                ttl,
                                0,
                                0,
                                80,
                                service_server,
                            )
                        ]
                    ),
                )
                send_event.wait(wait_time)
                assert last_sent is not None
                assert len(last_sent.questions) == 2
                assert r.DNSQuestion(service_server, const._TYPE_A, const._CLASS_IN) in last_sent.questions
                assert r.DNSQuestion(service_server, const._TYPE_AAAA, const._CLASS_IN) in last_sent.questions
                last_sent = None
                assert service_info is None

                # Expect no further queries
                last_sent = None
                send_event.clear()
                _inject_response(
                    zc,
                    mock_incoming_msg(
                        [
                            r.DNSAddress(
                                service_server,
                                const._TYPE_A,
                                const._CLASS_IN | const._CLASS_UNIQUE,
                                ttl,
                                socket.inet_pton(socket.AF_INET, service_address),
                            ),
                            r.DNSAddress(
                                service_server,
                                const._TYPE_AAAA,
                                const._CLASS_IN | const._CLASS_UNIQUE,
                                ttl,
                                socket.inet_pton(socket.AF_INET6, service_address_v6_ll),
                                scope_id=service_scope_id,
                            ),
                        ]
                    ),
                )
                send_event.wait(wait_time)
                assert last_sent is None
                assert service_info is not None

            finally:
                helper_thread.join()
                zc.remove_all_service_listeners()
                zc.close()

    @unittest.skipIf(not has_working_ipv6(), "Requires IPv6")
    @unittest.skipIf(os.environ.get("SKIP_IPV6"), "IPv6 tests disabled")
    def test_get_info_suppressed_by_question_history(self):
        zc = r.Zeroconf(interfaces=["127.0.0.1"])

        service_name = "name._type._tcp.local."
        service_type = "_type._tcp.local."

        service_info = None
        send_event = Event()
        service_info_event = Event()

        last_sent: r.DNSOutgoing | None = None

        def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT, v6_flow_scope=()):
            """Sends an outgoing packet."""
            nonlocal last_sent

            last_sent = out
            send_event.set()

        # patch the zeroconf send
        with patch.object(zc, "async_send", send):

            def mock_incoming_msg(records: Iterable[r.DNSRecord]) -> r.DNSIncoming:
                generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)

                for record in records:
                    generated.add_answer_at_time(record, 0)

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

            def get_service_info_helper(zc, type, name):
                nonlocal service_info
                service_info = zc.get_service_info(type, name)
                service_info_event.set()

            try:
                helper_thread = threading.Thread(
                    target=get_service_info_helper,
                    args=(zc, service_type, service_name),
                )
                helper_thread.start()
                wait_time = (const._LISTENER_TIME + info._AVOID_SYNC_DELAY_RANDOM_INTERVAL[1] + 5) / 1000

                # Expect query for SRV, TXT, A, AAAA
                send_event.wait(wait_time)
                assert last_sent is not None
                assert len(last_sent.questions) == 4
                assert r.DNSQuestion(service_name, const._TYPE_SRV, const._CLASS_IN) in last_sent.questions
                assert r.DNSQuestion(service_name, const._TYPE_TXT, const._CLASS_IN) in last_sent.questions
                assert r.DNSQuestion(service_name, const._TYPE_A, const._CLASS_IN) in last_sent.questions
                assert r.DNSQuestion(service_name, const._TYPE_AAAA, const._CLASS_IN) in last_sent.questions
                assert service_info is None

                # Expect query for SRV only as A, AAAA, and TXT are suppressed
                # by the question history
                last_sent = None
                send_event.clear()
                for _ in range(3):
                    send_event.wait(
                        wait_time * 0.25
                    )  # Wait long enough to be inside the question history window
                    now = r.current_time_millis()
                    zc.question_history.add_question_at_time(
                        r.DNSQuestion(service_name, const._TYPE_A, const._CLASS_IN),
                        now,
                        set(),
                    )
                    zc.question_history.add_question_at_time(
                        r.DNSQuestion(service_name, const._TYPE_AAAA, const._CLASS_IN),
                        now,
                        set(),
                    )
                    zc.question_history.add_question_at_time(
                        r.DNSQuestion(service_name, const._TYPE_TXT, const._CLASS_IN),
                        now,
                        set(),
                    )
                send_event.wait(wait_time * 0.25)
                assert last_sent is not None
                assert len(last_sent.questions) == 1  # type: ignore[unreachable]
                assert r.DNSQuestion(service_name, const._TYPE_SRV, const._CLASS_IN) in last_sent.questions
                assert service_info is None

                wait_time = (
                    const._DUPLICATE_QUESTION_INTERVAL + info._AVOID_SYNC_DELAY_RANDOM_INTERVAL[1] + 5
                ) / 1000
                # Expect no queries as all are suppressed by the question history
                last_sent = None
                send_event.clear()
                for _ in range(3):
                    send_event.wait(
                        wait_time * 0.25
                    )  # Wait long enough to be inside the question history window
                    now = r.current_time_millis()
                    zc.question_history.add_question_at_time(
                        r.DNSQuestion(service_name, const._TYPE_A, const._CLASS_IN),
                        now,
                        set(),
                    )
                    zc.question_history.add_question_at_time(
                        r.DNSQuestion(service_name, const._TYPE_AAAA, const._CLASS_IN),
                        now,
                        set(),
                    )
                    zc.question_history.add_question_at_time(
                        r.DNSQuestion(service_name, const._TYPE_TXT, const._CLASS_IN),
                        now,
                        set(),
                    )
                    zc.question_history.add_question_at_time(
                        r.DNSQuestion(service_name, const._TYPE_SRV, const._CLASS_IN),
                        now,
                        set(),
                    )
                send_event.wait(wait_time * 0.25)
                # All questions are suppressed so no query should be sent
                assert last_sent is None
                assert service_info is None

            finally:
                helper_thread.join()
                zc.remove_all_service_listeners()
                zc.close()

    def test_get_info_single(self):
        zc = r.Zeroconf(interfaces=["127.0.0.1"])

        service_name = "name._type._tcp.local."
        service_type = "_type._tcp.local."
        service_server = "ash-1.local."
        service_text = b"path=/~matt1/"
        service_address = "10.0.1.2"

        service_info = None
        send_event = Event()
        service_info_event = Event()

        last_sent: r.DNSOutgoing | None = None

        def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT, v6_flow_scope=()):
            """Sends an outgoing packet."""
            nonlocal last_sent

            last_sent = out
            send_event.set()

        # patch the zeroconf send
        with patch.object(zc, "async_send", send):

            def mock_incoming_msg(records: Iterable[r.DNSRecord]) -> r.DNSIncoming:
                generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)

                for record in records:
                    generated.add_answer_at_time(record, 0)

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

            def get_service_info_helper(zc, type, name):
                nonlocal service_info
                service_info = zc.get_service_info(type, name)
                service_info_event.set()

            try:
                ttl = 120
                helper_thread = threading.Thread(
                    target=get_service_info_helper,
                    args=(zc, service_type, service_name),
                )
                helper_thread.start()
                wait_time = 1

                # Expect query for SRV, TXT, A, AAAA
                send_event.wait(wait_time)
                assert last_sent is not None
                assert len(last_sent.questions) == 4
                assert r.DNSQuestion(service_name, const._TYPE_SRV, const._CLASS_IN) in last_sent.questions
                assert r.DNSQuestion(service_name, const._TYPE_TXT, const._CLASS_IN) in last_sent.questions
                assert r.DNSQuestion(service_name, const._TYPE_A, const._CLASS_IN) in last_sent.questions
                assert r.DNSQuestion(service_name, const._TYPE_AAAA, const._CLASS_IN) in last_sent.questions
                assert service_info is None

                # Expect no further queries
                last_sent = None
                send_event.clear()
                _inject_response(
                    zc,
                    mock_incoming_msg(
                        [
                            r.DNSText(
                                service_name,
                                const._TYPE_TXT,
                                const._CLASS_IN | const._CLASS_UNIQUE,
                                ttl,
                                service_text,
                            ),
                            r.DNSService(
                                service_name,
                                const._TYPE_SRV,
                                const._CLASS_IN | const._CLASS_UNIQUE,
                                ttl,
                                0,
                                0,
                                80,
                                service_server,
                            ),
                            r.DNSAddress(
                                service_server,
                                const._TYPE_A,
                                const._CLASS_IN | const._CLASS_UNIQUE,
                                ttl,
                                socket.inet_pton(socket.AF_INET, service_address),
                            ),
                        ]
                    ),
                )
                send_event.wait(wait_time)
                assert last_sent is None
                assert service_info is not None

            finally:
                helper_thread.join()
                zc.remove_all_service_listeners()
                zc.close()

    def test_service_info_duplicate_properties_txt_records(self):
        """Verify the first property is always used when there are duplicates in a txt record."""

        zc = r.Zeroconf(interfaces=["127.0.0.1"])
        desc = {"path": "/~paulsm/"}
        service_name = "name._type._tcp.local."
        service_type = "_type._tcp.local."
        service_server = "ash-1.local."
        service_address = socket.inet_aton("10.0.1.2")
        ttl = 120
        now = r.current_time_millis()
        info = ServiceInfo(
            service_type,
            service_name,
            22,
            0,
            0,
            desc,
            service_server,
            addresses=[service_address],
        )
        info.async_update_records(
            zc,
            now,
            [
                r.RecordUpdate(
                    r.DNSText(
                        service_name,
                        const._TYPE_TXT,
                        const._CLASS_IN | const._CLASS_UNIQUE,
                        ttl,
                        b"\x04ff=0\x04ci=2\x04sf=0\x0bsh=6fLM5A==\x04dd=0\x04jl=2\x04qq=0\x0brr=6fLM5A==\x04ci=3",
                    ),
                    None,
                )
            ],
        )
        assert info.properties[b"dd"] == b"0"
        assert info.properties[b"jl"] == b"2"
        assert info.properties[b"ci"] == b"2"
        zc.close()


def test_multiple_addresses():
    type_ = "_http._tcp.local."
    registration_name = f"xxxyyy.{type_}"
    desc = {"path": "/~paulsm/"}
    address_parsed = "10.0.1.2"
    address = socket.inet_aton(address_parsed)

    # New kwarg way
    info = ServiceInfo(
        type_,
        registration_name,
        80,
        0,
        0,
        desc,
        "ash-2.local.",
        addresses=[address, address],
    )

    assert info.addresses == [address, address]
    assert info.parsed_addresses() == [address_parsed, address_parsed]
    assert info.parsed_scoped_addresses() == [address_parsed, address_parsed]

    info = ServiceInfo(
        type_,
        registration_name,
        80,
        0,
        0,
        desc,
        "ash-2.local.",
        parsed_addresses=[address_parsed, address_parsed],
    )
    assert info.addresses == [address, address]
    assert info.parsed_addresses() == [address_parsed, address_parsed]
    assert info.parsed_scoped_addresses() == [address_parsed, address_parsed]

    if has_working_ipv6() and not os.environ.get("SKIP_IPV6"):
        address_v6_parsed = "2001:db8::1"
        address_v6 = socket.inet_pton(socket.AF_INET6, address_v6_parsed)
        address_v6_ll_parsed = "fe80::52e:c2f2:bc5f:e9c6"
        address_v6_ll_scoped_parsed = "fe80::52e:c2f2:bc5f:e9c6%12"
        address_v6_ll = socket.inet_pton(socket.AF_INET6, address_v6_ll_parsed)
        interface_index = 12
        infos = [
            ServiceInfo(
                type_,
                registration_name,
                80,
                0,
                0,
                desc,
                "ash-2.local.",
                addresses=[address, address_v6, address_v6_ll],
                interface_index=interface_index,
            ),
            ServiceInfo(
                type_,
                registration_name,
                80,
                0,
                0,
                desc,
                "ash-2.local.",
                parsed_addresses=[
                    address_parsed,
                    address_v6_parsed,
                    address_v6_ll_parsed,
                ],
                interface_index=interface_index,
            ),
        ]
        for info in infos:
            assert info.addresses == [address]
            assert info.addresses_by_version(r.IPVersion.All) == [
                address,
                address_v6,
                address_v6_ll,
            ]
            assert info.ip_addresses_by_version(r.IPVersion.All) == [
                ip_address(address),
                ip_address(address_v6),
                ip_address(address_v6_ll_scoped_parsed),
            ]
            assert info.addresses_by_version(r.IPVersion.V4Only) == [address]
            assert info.ip_addresses_by_version(r.IPVersion.V4Only) == [ip_address(address)]
            assert info.addresses_by_version(r.IPVersion.V6Only) == [
                address_v6,
                address_v6_ll,
            ]
            assert info.ip_addresses_by_version(r.IPVersion.V6Only) == [
                ip_address(address_v6),
                ip_address(address_v6_ll_scoped_parsed),
            ]
            assert info.parsed_addresses() == [
                address_parsed,
                address_v6_parsed,
                address_v6_ll_parsed,
            ]
            assert info.parsed_addresses(r.IPVersion.V4Only) == [address_parsed]
            assert info.parsed_addresses(r.IPVersion.V6Only) == [
                address_v6_parsed,
                address_v6_ll_parsed,
            ]
            assert info.parsed_scoped_addresses() == [
                address_parsed,
                address_v6_parsed,
                address_v6_ll_scoped_parsed,
            ]
            assert info.parsed_scoped_addresses(r.IPVersion.V4Only) == [address_parsed]
            assert info.parsed_scoped_addresses(r.IPVersion.V6Only) == [
                address_v6_parsed,
                address_v6_ll_scoped_parsed,
            ]


def test_scoped_addresses_from_cache():
    type_ = "_http._tcp.local."
    registration_name = f"scoped.{type_}"
    zeroconf = r.Zeroconf(interfaces=["127.0.0.1"])
    host = "scoped.local."

    zeroconf.cache.async_add_records(
        [
            r.DNSPointer(
                type_,
                const._TYPE_PTR,
                const._CLASS_IN | const._CLASS_UNIQUE,
                120,
                registration_name,
            ),
            r.DNSService(
                registration_name,
                const._TYPE_SRV,
                const._CLASS_IN | const._CLASS_UNIQUE,
                120,
                0,
                0,
                80,
                host,
            ),
            r.DNSAddress(
                host,
                const._TYPE_AAAA,
                const._CLASS_IN | const._CLASS_UNIQUE,
                120,
                socket.inet_pton(socket.AF_INET6, "fe80::52e:c2f2:bc5f:e9c6"),
                scope_id=12,
            ),
        ]
    )

    # New kwarg way
    info = ServiceInfo(type_, registration_name)
    info.load_from_cache(zeroconf)
    assert info.parsed_scoped_addresses() == ["fe80::52e:c2f2:bc5f:e9c6%12"]
    assert info.ip_addresses_by_version(r.IPVersion.V6Only) == [ip_address("fe80::52e:c2f2:bc5f:e9c6%12")]
    zeroconf.close()


# This test uses asyncio because it needs to access the cache directly
# which is not threadsafe
@pytest.mark.asyncio
async def test_multiple_a_addresses_newest_address_first():
    """Test that info.addresses returns the newest seen address first."""
    type_ = "_http._tcp.local."
    registration_name = f"multiarec.{type_}"
    desc = {"path": "/~paulsm/"}
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
    cache = aiozc.zeroconf.cache
    host = "multahost.local."
    record1 = r.DNSAddress(host, const._TYPE_A, const._CLASS_IN, 1000, b"\x7f\x00\x00\x01")
    record2 = r.DNSAddress(host, const._TYPE_A, const._CLASS_IN, 1000, b"\x7f\x00\x00\x02")
    cache.async_add_records([record1, record2])

    # New kwarg way
    info = ServiceInfo(type_, registration_name, 80, 0, 0, desc, host)
    info.load_from_cache(aiozc.zeroconf)
    assert info.addresses == [b"\x7f\x00\x00\x02", b"\x7f\x00\x00\x01"]
    await aiozc.async_close()


@pytest.mark.asyncio
async def test_invalid_a_addresses(caplog):
    type_ = "_http._tcp.local."
    registration_name = f"multiarec.{type_}"
    desc = {"path": "/~paulsm/"}
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
    cache = aiozc.zeroconf.cache
    host = "multahost.local."
    record1 = r.DNSAddress(host, const._TYPE_A, const._CLASS_IN, 1000, b"a")
    record2 = r.DNSAddress(host, const._TYPE_A, const._CLASS_IN, 1000, b"b")
    cache.async_add_records([record1, record2])

    # New kwarg way
    info = ServiceInfo(type_, registration_name, 80, 0, 0, desc, host)
    info.load_from_cache(aiozc.zeroconf)
    assert not info.addresses
    assert "Encountered invalid address while processing record" in caplog.text

    await aiozc.async_close()


@unittest.skipIf(not has_working_ipv6(), "Requires IPv6")
@unittest.skipIf(os.environ.get("SKIP_IPV6"), "IPv6 tests disabled")
def test_filter_address_by_type_from_service_info():
    """Verify dns_addresses can filter by ipversion."""
    desc = {"path": "/~paulsm/"}
    type_ = "_homeassistant._tcp.local."
    name = "MyTestHome"
    registration_name = f"{name}.{type_}"
    ipv4 = socket.inet_aton("10.0.1.2")
    ipv6 = socket.inet_pton(socket.AF_INET6, "2001:db8::1")
    info = ServiceInfo(type_, registration_name, 80, 0, 0, desc, "ash-2.local.", addresses=[ipv4, ipv6])

    def dns_addresses_to_addresses(dns_address: list[DNSAddress]) -> list[bytes]:
        return [address.address for address in dns_address]

    assert dns_addresses_to_addresses(info.dns_addresses()) == [ipv4, ipv6]
    assert dns_addresses_to_addresses(info.dns_addresses(version=r.IPVersion.All)) == [
        ipv4,
        ipv6,
    ]
    assert dns_addresses_to_addresses(info.dns_addresses(version=r.IPVersion.V4Only)) == [ipv4]
    assert dns_addresses_to_addresses(info.dns_addresses(version=r.IPVersion.V6Only)) == [ipv6]


def test_changing_name_updates_serviceinfo_key():
    """Verify a name change will adjust the underlying key value."""
    type_ = "_homeassistant._tcp.local."
    name = "MyTestHome"
    info_service = ServiceInfo(
        type_,
        f"{name}.{type_}",
        80,
        0,
        0,
        {"path": "/~paulsm/"},
        "ash-2.local.",
        addresses=[socket.inet_aton("10.0.1.2")],
    )
    assert info_service.key == "mytesthome._homeassistant._tcp.local."
    info_service.name = "YourTestHome._homeassistant._tcp.local."
    assert info_service.key == "yourtesthome._homeassistant._tcp.local."


def test_serviceinfo_address_updates():
    """Verify adding/removing/setting addresses on ServiceInfo."""
    type_ = "_homeassistant._tcp.local."
    name = "MyTestHome"

    # Verify addresses and parsed_addresses are mutually exclusive
    with pytest.raises(TypeError):
        info_service = ServiceInfo(
            type_,
            f"{name}.{type_}",
            80,
            0,
            0,
            {"path": "/~paulsm/"},
            "ash-2.local.",
            addresses=[socket.inet_aton("10.0.1.2")],
            parsed_addresses=["10.0.1.2"],
        )

    info_service = ServiceInfo(
        type_,
        f"{name}.{type_}",
        80,
        0,
        0,
        {"path": "/~paulsm/"},
        "ash-2.local.",
        addresses=[socket.inet_aton("10.0.1.2")],
    )
    info_service.addresses = [socket.inet_aton("10.0.1.3")]
    assert info_service.addresses == [socket.inet_aton("10.0.1.3")]


def test_serviceinfo_accepts_bytes_or_string_dict():
    """Verify a bytes or string dict can be passed to ServiceInfo."""
    type_ = "_homeassistant._tcp.local."
    name = "MyTestHome"
    addresses = [socket.inet_aton("10.0.1.2")]
    server_name = "ash-2.local."
    info_service = ServiceInfo(
        type_,
        f"{name}.{type_}",
        80,
        0,
        0,
        {b"path": b"/~paulsm/"},
        server_name,
        addresses=addresses,
    )
    assert info_service.dns_text().text == b"\x0epath=/~paulsm/"
    info_service = ServiceInfo(
        type_,
        f"{name}.{type_}",
        80,
        0,
        0,
        {"path": "/~paulsm/"},
        server_name,
        addresses=addresses,
    )
    assert info_service.dns_text().text == b"\x0epath=/~paulsm/"
    info_service = ServiceInfo(
        type_,
        f"{name}.{type_}",
        80,
        0,
        0,
        {b"path": "/~paulsm/"},
        server_name,
        addresses=addresses,
    )
    assert info_service.dns_text().text == b"\x0epath=/~paulsm/"
    info_service = ServiceInfo(
        type_,
        f"{name}.{type_}",
        80,
        0,
        0,
        {"path": b"/~paulsm/"},
        server_name,
        addresses=addresses,
    )
    assert info_service.dns_text().text == b"\x0epath=/~paulsm/"


def test_asking_qu_questions():
    """Verify explicitly asking QU questions."""
    type_ = "_quservice._tcp.local."
    zeroconf = r.Zeroconf(interfaces=["127.0.0.1"])

    # we are going to patch the zeroconf send to check query transmission
    old_send = zeroconf.async_send

    first_outgoing = None

    def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT):
        """Sends an outgoing packet."""
        nonlocal first_outgoing
        if first_outgoing is None:
            first_outgoing = out
        old_send(out, addr=addr, port=port)

    # patch the zeroconf send
    with patch.object(zeroconf, "async_send", send):
        zeroconf.get_service_info(f"name.{type_}", type_, 500, question_type=r.DNSQuestionType.QU)
        assert first_outgoing.questions[0].unicast is True  # type: ignore[union-attr]
        zeroconf.close()


def test_asking_qm_questions():
    """Verify explicitly asking QM questions."""
    type_ = "_quservice._tcp.local."
    zeroconf = r.Zeroconf(interfaces=["127.0.0.1"])

    # we are going to patch the zeroconf send to check query transmission
    old_send = zeroconf.async_send

    first_outgoing = None

    def send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT):
        """Sends an outgoing packet."""
        nonlocal first_outgoing
        if first_outgoing is None:
            first_outgoing = out
        old_send(out, addr=addr, port=port)

    # patch the zeroconf send
    with patch.object(zeroconf, "async_send", send):
        zeroconf.get_service_info(f"name.{type_}", type_, 500, question_type=r.DNSQuestionType.QM)
        assert first_outgoing.questions[0].unicast is False  # type: ignore[union-attr]
        zeroconf.close()


def test_request_timeout():
    """Test that the timeout does not throw an exception and finishes close to the actual timeout."""
    zeroconf = r.Zeroconf(interfaces=["127.0.0.1"])
    start_time = r.current_time_millis()
    assert zeroconf.get_service_info("_notfound.local.", "notthere._notfound.local.") is None
    end_time = r.current_time_millis()
    zeroconf.close()
    # 3000ms for the default timeout
    # 1000ms for loaded systems + schedule overhead
    assert (end_time - start_time) < 3000 + 1000


@pytest.mark.asyncio
async def test_we_try_four_times_with_random_delay():
    """Verify we try four times even with the random delay."""
    type_ = "_typethatisnothere._tcp.local."
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])

    # we are going to patch the zeroconf send to check query transmission
    request_count = 0

    def async_send(out, addr=const._MDNS_ADDR, port=const._MDNS_PORT):
        """Sends an outgoing packet."""
        nonlocal request_count
        request_count += 1

    # patch the zeroconf send
    with patch.object(aiozc.zeroconf, "async_send", async_send):
        await aiozc.async_get_service_info(f"willnotbefound.{type_}", type_)

    await aiozc.async_close()

    assert request_count == 4


@pytest.mark.asyncio
async def test_release_wait_when_new_recorded_added():
    """Test that async_request returns as soon as new matching records are added to the cache."""
    type_ = "_http._tcp.local."
    registration_name = f"multiarec.{type_}"
    desc = {"path": "/~paulsm/"}
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
    host = "multahost.local."

    # New kwarg way
    info = ServiceInfo(type_, registration_name, 80, 0, 0, desc, host)
    task = asyncio.create_task(info.async_request(aiozc.zeroconf, timeout=200))
    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSNsec(
            registration_name,
            const._TYPE_NSEC,
            const._CLASS_IN | const._CLASS_UNIQUE,
            const._DNS_OTHER_TTL,
            registration_name,
            [const._TYPE_AAAA],
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSService(
            registration_name,
            const._TYPE_SRV,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            0,
            0,
            80,
            host,
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSAddress(
            host,
            const._TYPE_A,
            const._CLASS_IN,
            10000,
            b"\x7f\x00\x00\x01",
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSText(
            registration_name,
            const._TYPE_TXT,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            b"\x04ff=0\x04ci=2\x04sf=0\x0bsh=6fLM5A==",
        ),
        0,
    )
    await aiozc.zeroconf.async_wait_for_start()
    await asyncio.sleep(0)
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))
    assert await asyncio.wait_for(task, timeout=2)
    assert info.addresses == [b"\x7f\x00\x00\x01"]
    await aiozc.async_close()


@pytest.mark.asyncio
async def test_port_changes_are_seen():
    """Test that port changes are seen by async_request."""
    type_ = "_http._tcp.local."
    registration_name = f"multiarec.{type_}"
    desc = {"path": "/~paulsm/"}
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
    host = "multahost.local."

    # New kwarg way
    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSNsec(
            registration_name,
            const._TYPE_NSEC,
            const._CLASS_IN | const._CLASS_UNIQUE,
            const._DNS_OTHER_TTL,
            registration_name,
            [const._TYPE_AAAA],
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSService(
            registration_name,
            const._TYPE_SRV,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            0,
            0,
            80,
            host,
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSAddress(
            host,
            const._TYPE_A,
            const._CLASS_IN,
            10000,
            b"\x7f\x00\x00\x01",
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSText(
            registration_name,
            const._TYPE_TXT,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            b"\x04ff=0\x04ci=2\x04sf=0\x0bsh=6fLM5A==",
        ),
        0,
    )
    await aiozc.zeroconf.async_wait_for_start()
    await asyncio.sleep(0)
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))

    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSService(
            registration_name,
            const._TYPE_SRV,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            90,
            90,
            81,
            host,
        ),
        0,
    )
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))

    info = ServiceInfo(type_, registration_name, 80, 10, 10, desc, host)
    await info.async_request(aiozc.zeroconf, timeout=200)
    assert info.port == 81
    assert info.priority == 90
    assert info.weight == 90
    await aiozc.async_close()


@pytest.mark.asyncio
async def test_port_changes_are_seen_with_directed_request():
    """Test that port changes are seen by async_request with a directed request."""
    type_ = "_http._tcp.local."
    registration_name = f"multiarec.{type_}"
    desc = {"path": "/~paulsm/"}
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
    host = "multahost.local."

    # New kwarg way
    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSNsec(
            registration_name,
            const._TYPE_NSEC,
            const._CLASS_IN | const._CLASS_UNIQUE,
            const._DNS_OTHER_TTL,
            registration_name,
            [const._TYPE_AAAA],
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSService(
            registration_name,
            const._TYPE_SRV,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            0,
            0,
            80,
            host,
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSAddress(
            host,
            const._TYPE_A,
            const._CLASS_IN,
            10000,
            b"\x7f\x00\x00\x01",
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSText(
            registration_name,
            const._TYPE_TXT,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            b"\x04ff=0\x04ci=2\x04sf=0\x0bsh=6fLM5A==",
        ),
        0,
    )
    await aiozc.zeroconf.async_wait_for_start()
    await asyncio.sleep(0)
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))

    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSService(
            registration_name,
            const._TYPE_SRV,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            90,
            90,
            81,
            host,
        ),
        0,
    )
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))

    info = ServiceInfo(type_, registration_name, 80, 10, 10, desc, host)
    await info.async_request(aiozc.zeroconf, timeout=200, addr="127.0.0.1", port=5353)
    assert info.port == 81
    assert info.priority == 90
    assert info.weight == 90
    await aiozc.async_close()


@pytest.mark.asyncio
async def test_ipv4_changes_are_seen():
    """Test that ipv4 changes are seen by async_request."""
    type_ = "_http._tcp.local."
    registration_name = f"multiaipv4rec.{type_}"
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
    host = "multahost.local."

    # New kwarg way
    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSNsec(
            registration_name,
            const._TYPE_NSEC,
            const._CLASS_IN | const._CLASS_UNIQUE,
            const._DNS_OTHER_TTL,
            registration_name,
            [const._TYPE_AAAA],
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSService(
            registration_name,
            const._TYPE_SRV,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            0,
            0,
            80,
            host,
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSAddress(
            host,
            const._TYPE_A,
            const._CLASS_IN,
            10000,
            b"\x7f\x00\x00\x01",
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSText(
            registration_name,
            const._TYPE_TXT,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            b"\x04ff=0\x04ci=2\x04sf=0\x0bsh=6fLM5A==",
        ),
        0,
    )
    await aiozc.zeroconf.async_wait_for_start()
    await asyncio.sleep(0)
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))
    info = ServiceInfo(type_, registration_name)
    info.load_from_cache(aiozc.zeroconf)
    assert info.addresses_by_version(IPVersion.V4Only) == [b"\x7f\x00\x00\x01"]

    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSAddress(
            host,
            const._TYPE_A,
            const._CLASS_IN,
            10000,
            b"\x7f\x00\x00\x02",
        ),
        0,
    )
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))

    info = ServiceInfo(type_, registration_name)
    info.load_from_cache(aiozc.zeroconf)
    assert info.addresses_by_version(IPVersion.V4Only) == [
        b"\x7f\x00\x00\x02",
        b"\x7f\x00\x00\x01",
    ]
    await info.async_request(aiozc.zeroconf, timeout=200)
    assert info.addresses_by_version(IPVersion.V4Only) == [
        b"\x7f\x00\x00\x02",
        b"\x7f\x00\x00\x01",
    ]
    await aiozc.async_close()


@pytest.mark.asyncio
async def test_ipv6_changes_are_seen():
    """Test that ipv6 changes are seen by async_request."""
    type_ = "_http._tcp.local."
    registration_name = f"multiaipv6rec.{type_}"
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
    host = "multahost.local."

    # New kwarg way
    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSNsec(
            registration_name,
            const._TYPE_NSEC,
            const._CLASS_IN | const._CLASS_UNIQUE,
            const._DNS_OTHER_TTL,
            registration_name,
            [const._TYPE_A],
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSService(
            registration_name,
            const._TYPE_SRV,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            0,
            0,
            80,
            host,
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSAddress(
            host,
            const._TYPE_AAAA,
            const._CLASS_IN,
            10000,
            b"\xde\xad\xbe\xef\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSText(
            registration_name,
            const._TYPE_TXT,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            b"\x04ff=0\x04ci=2\x04sf=0\x0bsh=6fLM5A==",
        ),
        0,
    )
    await aiozc.zeroconf.async_wait_for_start()
    await asyncio.sleep(0)
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))
    info = ServiceInfo(type_, registration_name)
    info.load_from_cache(aiozc.zeroconf)
    assert info.addresses_by_version(IPVersion.V6Only) == [
        b"\xde\xad\xbe\xef\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
    ]
    info.load_from_cache(aiozc.zeroconf)
    assert info.addresses_by_version(IPVersion.V6Only) == [
        b"\xde\xad\xbe\xef\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
    ]

    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSAddress(
            host,
            const._TYPE_AAAA,
            const._CLASS_IN,
            10000,
            b"\x00\xad\xbe\xef\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
        ),
        0,
    )
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))

    info = ServiceInfo(type_, registration_name)
    info.load_from_cache(aiozc.zeroconf)
    assert info.addresses_by_version(IPVersion.V6Only) == [
        b"\x00\xad\xbe\xef\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
        b"\xde\xad\xbe\xef\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
    ]
    await info.async_request(aiozc.zeroconf, timeout=200)
    assert info.addresses_by_version(IPVersion.V6Only) == [
        b"\x00\xad\xbe\xef\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
        b"\xde\xad\xbe\xef\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
    ]

    await aiozc.async_close()


@pytest.mark.asyncio
async def test_bad_ip_addresses_ignored_in_cache():
    """Test that bad ip address in the cache are ignored async_request."""
    type_ = "_http._tcp.local."
    registration_name = f"multiarec.{type_}"
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
    host = "multahost.local."

    # New kwarg way
    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSService(
            registration_name,
            const._TYPE_SRV,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            0,
            0,
            80,
            host,
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSAddress(
            host,
            const._TYPE_A,
            const._CLASS_IN,
            10000,
            b"\x7f\x00\x00\x01",
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSText(
            registration_name,
            const._TYPE_TXT,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            b"\x04ff=0\x04ci=2\x04sf=0\x0bsh=6fLM5A==",
        ),
        0,
    )
    # Manually add a bad record to the cache
    aiozc.zeroconf.cache.async_add_records([DNSAddress(host, const._TYPE_A, const._CLASS_IN, 10000, b"\x00")])

    await aiozc.zeroconf.async_wait_for_start()
    await asyncio.sleep(0)
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))
    info = ServiceInfo(type_, registration_name)
    info.load_from_cache(aiozc.zeroconf)
    assert info.addresses_by_version(IPVersion.V4Only) == [b"\x7f\x00\x00\x01"]


@pytest.mark.asyncio
async def test_service_name_change_as_seen_has_ip_in_cache():
    """Test that service name changes are seen by async_request when the ip is in the cache."""
    type_ = "_http._tcp.local."
    registration_name = f"multiarec.{type_}"
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
    host = "multahost.local."

    # New kwarg way
    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSNsec(
            registration_name,
            const._TYPE_NSEC,
            const._CLASS_IN | const._CLASS_UNIQUE,
            const._DNS_OTHER_TTL,
            registration_name,
            [const._TYPE_AAAA],
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSAddress(
            registration_name,
            const._TYPE_A,
            const._CLASS_IN,
            10000,
            b"\x7f\x00\x00\x01",
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSAddress(
            host,
            const._TYPE_A,
            const._CLASS_IN,
            10000,
            b"\x7f\x00\x00\x02",
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSText(
            registration_name,
            const._TYPE_TXT,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            b"\x04ff=0\x04ci=2\x04sf=0\x0bsh=6fLM5A==",
        ),
        0,
    )
    await aiozc.zeroconf.async_wait_for_start()
    await asyncio.sleep(0)
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))

    info = ServiceInfo(type_, registration_name)
    await info.async_request(aiozc.zeroconf, timeout=200)
    assert info.addresses_by_version(IPVersion.V4Only) == []

    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSService(
            registration_name,
            const._TYPE_SRV,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            0,
            0,
            80,
            host,
        ),
        0,
    )
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))

    info = ServiceInfo(type_, registration_name)
    await info.async_request(aiozc.zeroconf, timeout=200)
    assert info.addresses_by_version(IPVersion.V4Only) == [b"\x7f\x00\x00\x02"]

    await aiozc.async_close()


@pytest.mark.asyncio
async def test_service_name_change_as_seen_ip_not_in_cache():
    """Test that service name changes are seen by async_request when the ip is not in the cache."""
    type_ = "_http._tcp.local."
    registration_name = f"multiarec.{type_}"
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
    host = "multahost.local."

    # New kwarg way
    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSNsec(
            registration_name,
            const._TYPE_NSEC,
            const._CLASS_IN | const._CLASS_UNIQUE,
            const._DNS_OTHER_TTL,
            registration_name,
            [const._TYPE_AAAA],
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSAddress(
            registration_name,
            const._TYPE_A,
            const._CLASS_IN,
            10000,
            b"\x7f\x00\x00\x01",
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSText(
            registration_name,
            const._TYPE_TXT,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            b"\x04ff=0\x04ci=2\x04sf=0\x0bsh=6fLM5A==",
        ),
        0,
    )
    await aiozc.zeroconf.async_wait_for_start()
    await asyncio.sleep(0)
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))

    info = ServiceInfo(type_, registration_name)
    await info.async_request(aiozc.zeroconf, timeout=200)
    assert info.addresses_by_version(IPVersion.V4Only) == []

    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSService(
            registration_name,
            const._TYPE_SRV,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            0,
            0,
            80,
            host,
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSAddress(
            host,
            const._TYPE_A,
            const._CLASS_IN,
            10000,
            b"\x7f\x00\x00\x02",
        ),
        0,
    )
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))

    info = ServiceInfo(type_, registration_name)
    await info.async_request(aiozc.zeroconf, timeout=200)
    assert info.addresses_by_version(IPVersion.V4Only) == [b"\x7f\x00\x00\x02"]

    await aiozc.async_close()


@pytest.mark.asyncio
@patch.object(info, "_LISTENER_TIME", 10000000)
async def test_release_wait_when_new_recorded_added_concurrency():
    """Test that concurrent async_request returns as soon as new matching records are added to the cache."""
    type_ = "_http._tcp.local."
    registration_name = f"multiareccon.{type_}"
    desc = {"path": "/~paulsm/"}
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
    host = "multahostcon.local."
    await aiozc.zeroconf.async_wait_for_start()

    # New kwarg way
    info = ServiceInfo(type_, registration_name, 80, 0, 0, desc, host)
    tasks = [asyncio.create_task(info.async_request(aiozc.zeroconf, timeout=200000)) for _ in range(10)]
    await asyncio.sleep(0.1)
    for task in tasks:
        assert not task.done()
    generated = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    generated.add_answer_at_time(
        r.DNSNsec(
            registration_name,
            const._TYPE_NSEC,
            const._CLASS_IN | const._CLASS_UNIQUE,
            const._DNS_OTHER_TTL,
            registration_name,
            [const._TYPE_AAAA],
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSService(
            registration_name,
            const._TYPE_SRV,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            0,
            0,
            80,
            host,
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSAddress(
            host,
            const._TYPE_A,
            const._CLASS_IN,
            10000,
            b"\x7f\x00\x00\x01",
        ),
        0,
    )
    generated.add_answer_at_time(
        r.DNSText(
            registration_name,
            const._TYPE_TXT,
            const._CLASS_IN | const._CLASS_UNIQUE,
            10000,
            b"\x04ff=0\x04ci=2\x04sf=0\x0bsh=6fLM5A==",
        ),
        0,
    )
    await asyncio.sleep(0)
    for task in tasks:
        assert not task.done()
    aiozc.zeroconf.record_manager.async_updates_from_response(r.DNSIncoming(generated.packets()[0]))
    _, pending = await asyncio.wait(tasks, timeout=2)
    assert not pending
    assert info.addresses == [b"\x7f\x00\x00\x01"]
    await aiozc.async_close()


@pytest.mark.asyncio
async def test_service_info_nsec_records():
    """Test we can generate nsec records from ServiceInfo."""
    type_ = "_http._tcp.local."
    registration_name = f"multiareccon.{type_}"
    desc = {"path": "/~paulsm/"}
    host = "multahostcon.local."
    info = ServiceInfo(type_, registration_name, 80, 0, 0, desc, host)
    nsec_record = info.dns_nsec([const._TYPE_A, const._TYPE_AAAA], 50)
    assert nsec_record.name == registration_name
    assert nsec_record.type == const._TYPE_NSEC
    assert nsec_record.ttl == 50
    assert nsec_record.rdtypes == [const._TYPE_A, const._TYPE_AAAA]


@pytest.mark.asyncio
async def test_address_resolver():
    """Test that the address resolver works."""
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
    await aiozc.zeroconf.async_wait_for_start()
    resolver = r.AddressResolver("address_resolver_test.local.")
    resolve_task = asyncio.create_task(resolver.async_request(aiozc.zeroconf, 3000))
    outgoing = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    outgoing.add_answer_at_time(
        r.DNSAddress(
            "address_resolver_test.local.",
            const._TYPE_A,
            const._CLASS_IN,
            10000,
            b"\x7f\x00\x00\x01",
        ),
        0,
    )

    aiozc.zeroconf.async_send(outgoing)
    assert await resolve_task
    assert resolver.addresses == [b"\x7f\x00\x00\x01"]


@pytest.mark.asyncio
async def test_address_resolver_ipv4():
    """Test that the IPv4 address resolver works."""
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
    await aiozc.zeroconf.async_wait_for_start()
    resolver = r.AddressResolverIPv4("address_resolver_test_ipv4.local.")
    resolve_task = asyncio.create_task(resolver.async_request(aiozc.zeroconf, 3000))
    outgoing = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    outgoing.add_answer_at_time(
        r.DNSAddress(
            "address_resolver_test_ipv4.local.",
            const._TYPE_A,
            const._CLASS_IN,
            10000,
            b"\x7f\x00\x00\x01",
        ),
        0,
    )

    aiozc.zeroconf.async_send(outgoing)
    assert await resolve_task
    assert resolver.addresses == [b"\x7f\x00\x00\x01"]


@pytest.mark.asyncio
@unittest.skipIf(not has_working_ipv6(), "Requires IPv6")
@unittest.skipIf(os.environ.get("SKIP_IPV6"), "IPv6 tests disabled")
async def test_address_resolver_ipv6():
    """Test that the IPv6 address resolver works."""
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
    await aiozc.zeroconf.async_wait_for_start()
    resolver = r.AddressResolverIPv6("address_resolver_test_ipv6.local.")
    resolve_task = asyncio.create_task(resolver.async_request(aiozc.zeroconf, 3000))
    outgoing = r.DNSOutgoing(const._FLAGS_QR_RESPONSE)
    outgoing.add_answer_at_time(
        r.DNSAddress(
            "address_resolver_test_ipv6.local.",
            const._TYPE_AAAA,
            const._CLASS_IN,
            10000,
            socket.inet_pton(socket.AF_INET6, "fe80::52e:c2f2:bc5f:e9c6"),
        ),
        0,
    )

    aiozc.zeroconf.async_send(outgoing)
    assert await resolve_task
    assert resolver.ip_addresses_by_version(IPVersion.All) == [ip_address("fe80::52e:c2f2:bc5f:e9c6")]


@pytest.mark.asyncio
async def test_unicast_flag_if_requested() -> None:
    """Verify we try four times even with the random delay."""
    type_ = "_typethatisnothere._tcp.local."
    aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])

    def async_send(out: DNSOutgoing, addr: str | None = None, port: int = const._MDNS_PORT) -> None:
        """Sends an outgoing packet."""
        for question in out.questions:
            assert question.unicast

    # patch the zeroconf send
    with patch.object(aiozc.zeroconf, "async_send", async_send):
        await aiozc.async_get_service_info(
            f"willnotbefound.{type_}", type_, question_type=r.DNSQuestionType.QU
        )

    await aiozc.async_close()