1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
"""Unit tests for zeroconf._services.registry."""
from __future__ import annotations
import socket
import unittest
import zeroconf as r
from zeroconf import ServiceInfo
class TestServiceRegistry(unittest.TestCase):
def test_only_register_once(self):
type_ = "_test-srvc-type._tcp.local."
name = "xxxyyy"
registration_name = f"{name}.{type_}"
desc = {"path": "/~paulsm/"}
info = ServiceInfo(
type_,
registration_name,
80,
0,
0,
desc,
"ash-2.local.",
addresses=[socket.inet_aton("10.0.1.2")],
)
registry = r.ServiceRegistry()
registry.async_add(info)
self.assertRaises(r.ServiceNameAlreadyRegistered, registry.async_add, info)
registry.async_remove(info)
registry.async_add(info)
def test_register_same_server(self):
type_ = "_test-srvc-type._tcp.local."
name = "xxxyyy"
name2 = "xxxyyy2"
registration_name = f"{name}.{type_}"
registration_name2 = f"{name2}.{type_}"
desc = {"path": "/~paulsm/"}
info = ServiceInfo(
type_,
registration_name,
80,
0,
0,
desc,
"same.local.",
addresses=[socket.inet_aton("10.0.1.2")],
)
info2 = ServiceInfo(
type_,
registration_name2,
80,
0,
0,
desc,
"same.local.",
addresses=[socket.inet_aton("10.0.1.2")],
)
registry = r.ServiceRegistry()
registry.async_add(info)
registry.async_add(info2)
assert registry.async_get_infos_server("same.local.") == [info, info2]
registry.async_remove(info)
assert registry.async_get_infos_server("same.local.") == [info2]
registry.async_remove(info2)
assert registry.async_get_infos_server("same.local.") == []
def test_unregister_multiple_times(self):
"""Verify we can unregister a service multiple times.
In production unregister_service and unregister_all_services
may happen at the same time during shutdown. We want to treat
this as non-fatal since its expected to happen and it is unlikely
that the callers know about each other.
"""
type_ = "_test-srvc-type._tcp.local."
name = "xxxyyy"
registration_name = f"{name}.{type_}"
desc = {"path": "/~paulsm/"}
info = ServiceInfo(
type_,
registration_name,
80,
0,
0,
desc,
"ash-2.local.",
addresses=[socket.inet_aton("10.0.1.2")],
)
registry = r.ServiceRegistry()
registry.async_add(info)
self.assertRaises(r.ServiceNameAlreadyRegistered, registry.async_add, info)
registry.async_remove(info)
registry.async_remove(info)
def test_lookups(self):
type_ = "_test-srvc-type._tcp.local."
name = "xxxyyy"
registration_name = f"{name}.{type_}"
desc = {"path": "/~paulsm/"}
info = ServiceInfo(
type_,
registration_name,
80,
0,
0,
desc,
"ash-2.local.",
addresses=[socket.inet_aton("10.0.1.2")],
)
registry = r.ServiceRegistry()
registry.async_add(info)
assert registry.async_get_service_infos() == [info]
assert registry.async_get_info_name(registration_name) == info
assert registry.async_get_infos_type(type_) == [info]
assert registry.async_get_infos_server("ash-2.local.") == [info]
assert registry.async_get_types() == [type_]
def test_lookups_upper_case_by_lower_case(self):
type_ = "_test-SRVC-type._tcp.local."
name = "Xxxyyy"
registration_name = f"{name}.{type_}"
desc = {"path": "/~paulsm/"}
info = ServiceInfo(
type_,
registration_name,
80,
0,
0,
desc,
"ASH-2.local.",
addresses=[socket.inet_aton("10.0.1.2")],
)
registry = r.ServiceRegistry()
registry.async_add(info)
assert registry.async_get_service_infos() == [info]
assert registry.async_get_info_name(registration_name.lower()) == info
assert registry.async_get_infos_type(type_.lower()) == [info]
assert registry.async_get_infos_server("ash-2.local.") == [info]
assert registry.async_get_types() == [type_.lower()]
|
