wirenboard / wb-nm-helper / 132

Committed 01 Jun 2026 06:37AM UTC coverage: 80.268% (+0.8%) from 79.446%

Build # 132

Build Type

push

github

Committed by

web-flow

Commit Message

Migrate from pycares to dnspython (#132)

Coverage Stats

93 of 95 new or added lines in 2 files covered. (97.89%)

3531 of 4399 relevant lines covered (80.27%)

0.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.01

/tests/test_dns_resolver.py

from unittest.mock import MagicMock, call, patch

import pytest

from wb.nm_helper import dns_resolver


def test_resolve_domain_name_uses_provided_servers_and_domains():
    resolver = MagicMock()
    resolver.resolve.return_value = [MagicMock(to_text=MagicMock(return_value="1.1.1.1"))]
    original_factory = object()

    with patch.object(dns_resolver.dns.query, "socket_factory", original_factory), patch.object(
        dns_resolver.dns.resolver, "Resolver", return_value=resolver
    ), patch.object(
        dns_resolver.dns.name, "from_text", side_effect=lambda domain: f"name:{domain}"
    ) as from_text:
        result = dns_resolver.resolve_domain_name("host", "eth0", ["8.8.8.8"], ["lan", " corp "])

    assert result == ["1.1.1.1"]
    assert resolver.nameservers == ["8.8.8.8"]
    assert resolver.search == ["name:lan", "name:corp"]
    assert resolver.use_search_by_default is True
    resolver.resolve.assert_called_once_with("host", "A")
    assert from_text.mock_calls == [call("lan"), call("corp")]


def test_resolve_domain_name_keeps_default_resolver_settings_for_empty_servers_and_domains():
    resolver = MagicMock()
    resolver.resolve.return_value = [MagicMock(to_text=MagicMock(return_value="1.1.1.1"))]
    resolver.nameservers = ["system"]
    resolver.search = ["system-domain"]
    resolver.use_search_by_default = False
    original_factory = object()

    with patch.object(dns_resolver.dns.query, "socket_factory", original_factory), patch.object(
        dns_resolver.dns.resolver, "Resolver", return_value=resolver
    ), patch.object(dns_resolver.dns.name, "from_text") as from_text:
        result = dns_resolver.resolve_domain_name("host", "eth0", [], [])

    assert result == ["1.1.1.1"]
    assert resolver.nameservers == ["system"]
    assert resolver.search == ["system-domain"]
    assert resolver.use_search_by_default is False
    resolver.resolve.assert_called_once_with("host", "A")
    from_text.assert_not_called()


def test_resolve_domain_name_binds_socket_to_interface_and_restores_factory():
    fake_socket = MagicMock()

    class ResolverStub:  # pylint: disable=too-few-public-methods
        def __init__(self):
            self.timeout = None
            self.lifetime = None
            self.nameservers = []

        def resolve(self, *_args, **_kwargs):
            dns_resolver.dns.query.socket_factory(
                dns_resolver.socket.AF_INET, dns_resolver.socket.SOCK_DGRAM, 0
            )
            return [MagicMock(to_text=MagicMock(return_value="1.1.1.1"))]

    resolver = ResolverStub()

    def original_factory(*_args, **_kwargs):
        return None

    with patch.object(dns_resolver.dns.query, "socket_factory", original_factory), patch.object(
        dns_resolver.dns.resolver, "Resolver", return_value=resolver
    ), patch.object(dns_resolver.socket, "socket", return_value=fake_socket) as socket_ctor:
        result = dns_resolver.resolve_domain_name("host", "eth0", ["8.8.8.8"], [])
        assert dns_resolver.dns.query.socket_factory is original_factory

    assert result == ["1.1.1.1"]
    socket_ctor.assert_called_once_with(dns_resolver.socket.AF_INET, dns_resolver.socket.SOCK_DGRAM, 0)
    fake_socket.setsockopt.assert_called_once_with(
        dns_resolver.socket.SOL_SOCKET,
        dns_resolver.socket.SO_BINDTODEVICE,
        b"eth0\0",
    )


def test_resolve_domain_name_restores_factory_when_resolve_raises():
    fake_socket = MagicMock()

    class ResolverStub:  # pylint: disable=too-few-public-methods
        def __init__(self):
            self.timeout = None
            self.lifetime = None
            self.nameservers = []

        def resolve(self, *_args, **_kwargs):
            dns_resolver.dns.query.socket_factory(
                dns_resolver.socket.AF_INET, dns_resolver.socket.SOCK_DGRAM, 0
            )
            raise RuntimeError("boom")

    resolver = ResolverStub()

    def original_factory(*_args, **_kwargs):
        return None

    with patch.object(dns_resolver.dns.query, "socket_factory", original_factory), patch.object(
        dns_resolver.dns.resolver, "Resolver", return_value=resolver
    ), patch.object(dns_resolver.socket, "socket", return_value=fake_socket):
        with pytest.raises(dns_resolver.DomainNameResolveException, match="Error during host resolving"):
            dns_resolver.resolve_domain_name("host", "eth0", [], [])
        assert dns_resolver.dns.query.socket_factory is original_factory

    fake_socket.setsockopt.assert_called_once_with(
        dns_resolver.socket.SOL_SOCKET,
        dns_resolver.socket.SO_BINDTODEVICE,
        b"eth0\0",
    )

1	from unittest.mock import MagicMock, call, patch	1✔
2
3	import pytest	1✔
4
5	from wb.nm_helper import dns_resolver	1✔
6
7
8	def test_resolve_domain_name_uses_provided_servers_and_domains():	1✔
9	resolver = MagicMock()	1✔
10	resolver.resolve.return_value = [MagicMock(to_text=MagicMock(return_value="1.1.1.1"))]	1✔
11	original_factory = object()	1✔
12
13	with patch.object(dns_resolver.dns.query, "socket_factory", original_factory), patch.object(	1✔
14	dns_resolver.dns.resolver, "Resolver", return_value=resolver
15	), patch.object(
16	dns_resolver.dns.name, "from_text", side_effect=lambda domain: f"name:{domain}"
17	) as from_text:
18	result = dns_resolver.resolve_domain_name("host", "eth0", ["8.8.8.8"], ["lan", " corp "])	1✔
19
20	assert result == ["1.1.1.1"]	1✔
21	assert resolver.nameservers == ["8.8.8.8"]	1✔
22	assert resolver.search == ["name:lan", "name:corp"]	1✔
23	assert resolver.use_search_by_default is True	1✔
24	resolver.resolve.assert_called_once_with("host", "A")	1✔
25	assert from_text.mock_calls == [call("lan"), call("corp")]	1✔
26
27
28	def test_resolve_domain_name_keeps_default_resolver_settings_for_empty_servers_and_domains():	1✔
29	resolver = MagicMock()	1✔
30	resolver.resolve.return_value = [MagicMock(to_text=MagicMock(return_value="1.1.1.1"))]	1✔
31	resolver.nameservers = ["system"]	1✔
32	resolver.search = ["system-domain"]	1✔
33	resolver.use_search_by_default = False	1✔
34	original_factory = object()	1✔
35
36	with patch.object(dns_resolver.dns.query, "socket_factory", original_factory), patch.object(	1✔
37	dns_resolver.dns.resolver, "Resolver", return_value=resolver
38	), patch.object(dns_resolver.dns.name, "from_text") as from_text:
39	result = dns_resolver.resolve_domain_name("host", "eth0", [], [])	1✔
40
41	assert result == ["1.1.1.1"]	1✔
42	assert resolver.nameservers == ["system"]	1✔
43	assert resolver.search == ["system-domain"]	1✔
44	assert resolver.use_search_by_default is False	1✔
45	resolver.resolve.assert_called_once_with("host", "A")	1✔
46	from_text.assert_not_called()	1✔
47
48
49	def test_resolve_domain_name_binds_socket_to_interface_and_restores_factory():	1✔
50	fake_socket = MagicMock()	1✔
51
52	class ResolverStub: # pylint: disable=too-few-public-methods	1✔
53	def __init__(self):	1✔
54	self.timeout = None	1✔
55	self.lifetime = None	1✔
56	self.nameservers = []	1✔
57
58	def resolve(self, _args, *_kwargs):	1✔
59	dns_resolver.dns.query.socket_factory(	1✔
60	dns_resolver.socket.AF_INET, dns_resolver.socket.SOCK_DGRAM, 0
61	)
62	return [MagicMock(to_text=MagicMock(return_value="1.1.1.1"))]	1✔
63
64	resolver = ResolverStub()	1✔
65
66	def original_factory(_args, *_kwargs):	1✔
NEW 67	return None	×
68
69	with patch.object(dns_resolver.dns.query, "socket_factory", original_factory), patch.object(	1✔
70	dns_resolver.dns.resolver, "Resolver", return_value=resolver
71	), patch.object(dns_resolver.socket, "socket", return_value=fake_socket) as socket_ctor:
72	result = dns_resolver.resolve_domain_name("host", "eth0", ["8.8.8.8"], [])	1✔
73	assert dns_resolver.dns.query.socket_factory is original_factory	1✔
74
75	assert result == ["1.1.1.1"]	1✔
76	socket_ctor.assert_called_once_with(dns_resolver.socket.AF_INET, dns_resolver.socket.SOCK_DGRAM, 0)	1✔
77	fake_socket.setsockopt.assert_called_once_with(	1✔
78	dns_resolver.socket.SOL_SOCKET,
79	dns_resolver.socket.SO_BINDTODEVICE,
80	b"eth0\0",
81	)
82
83
84	def test_resolve_domain_name_restores_factory_when_resolve_raises():	1✔
85	fake_socket = MagicMock()	1✔
86
87	class ResolverStub: # pylint: disable=too-few-public-methods	1✔
88	def __init__(self):	1✔
89	self.timeout = None	1✔
90	self.lifetime = None	1✔
91	self.nameservers = []	1✔
92
93	def resolve(self, _args, *_kwargs):	1✔
94	dns_resolver.dns.query.socket_factory(	1✔
95	dns_resolver.socket.AF_INET, dns_resolver.socket.SOCK_DGRAM, 0
96	)
97	raise RuntimeError("boom")	1✔
98
99	resolver = ResolverStub()	1✔
100
101	def original_factory(_args, *_kwargs):	1✔
NEW 102	return None	×
103
104	with patch.object(dns_resolver.dns.query, "socket_factory", original_factory), patch.object(	1✔
105	dns_resolver.dns.resolver, "Resolver", return_value=resolver
106	), patch.object(dns_resolver.socket, "socket", return_value=fake_socket):
107	with pytest.raises(dns_resolver.DomainNameResolveException, match="Error during host resolving"):	1✔
108	dns_resolver.resolve_domain_name("host", "eth0", [], [])	1✔
109	assert dns_resolver.dns.query.socket_factory is original_factory	1✔
110
111	fake_socket.setsockopt.assert_called_once_with(	1✔
112	dns_resolver.socket.SOL_SOCKET,
113	dns_resolver.socket.SO_BINDTODEVICE,
114	b"eth0\0",
115	)

wirenboard / wb-nm-helper / 132

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous