Skip to content

fix: CI test suite is very WET #1757

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
bluetoothbot wants to merge 1 commit into from
+43 −3
Closed

fix: CI test suite is very WET #1757

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion tests/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,12 +26,13 @@
import platform
import socket
import time
from collections.abc import Iterable
from functools import cache
from unittest import mock

import ifaddr

from zeroconf import DNSIncoming, DNSQuestion, DNSRecord, Zeroconf
from zeroconf import DNSIncoming, DNSOutgoing, DNSQuestion, DNSRecord, Zeroconf, const
from zeroconf._history import QuestionHistory

_MONOTONIC_RESOLUTION = time.get_clock_info("monotonic").resolution
Expand Down Expand Up @@ -70,6 +71,14 @@ def suppresses(self, question: DNSQuestion, now: float, known_answers: set[DNSRe
return False


def mock_incoming_msg(records: Iterable[DNSRecord]) -> DNSIncoming:
"""Build a `DNSIncoming` response message from a list of `DNSRecord`s."""
generated = DNSOutgoing(const._FLAGS_QR_RESPONSE)
for record in records:
generated.add_answer_at_time(record, 0)
return DNSIncoming(generated.packets()[0])


def _inject_responses(zc: Zeroconf, msgs: list[DNSIncoming]) -> None:
"""Inject a DNSIncoming response."""
assert zc.loop is not None
Expand Down
35 changes: 33 additions & 2 deletions tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,15 +3,16 @@
from __future__ import annotations

import threading
from collections.abc import Generator
from collections.abc import AsyncGenerator, Generator
from unittest.mock import patch

import pytest

from zeroconf import _core, const
from zeroconf import Zeroconf, _core, const
from zeroconf._handlers import query_handler
from zeroconf._services import browser as service_browser
from zeroconf._services import info as service_info
from zeroconf.asyncio import AsyncZeroconf


@pytest.fixture(autouse=True)
Expand All @@ -23,6 +24,36 @@ def verify_threads_ended():
assert not threads


@pytest.fixture
def zc_loopback() -> Generator[Zeroconf]:
"""Yield a loopback `Zeroconf` and close it on teardown.

Replaces the inline `zc = Zeroconf(interfaces=["127.0.0.1"])` +
explicit `zc.close()` pattern duplicated across the suite. Calling
`zc.close()` inside a test is still safe — `close()` is idempotent.
"""
zc = Zeroconf(interfaces=["127.0.0.1"])
try:
yield zc
finally:
zc.close()


@pytest.fixture
async def aiozc_loopback() -> AsyncGenerator[AsyncZeroconf]:
"""Yield a loopback `AsyncZeroconf` and close it on teardown.

Replaces the inline `aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])`
+ explicit `await aiozc.async_close()` pattern duplicated across the
suite. Calling `async_close()` inside a test is still safe.
"""
aiozc = AsyncZeroconf(interfaces=["127.0.0.1"])
try:
yield aiozc
finally:
await aiozc.async_close()


@pytest.fixture
def run_isolated():
"""Change the mDNS port to run the test in isolation."""
Expand Down
Loading