mock up socket registration, event selection & socket message response
(monkeypatch, probe_response)
| 7 | |
| 8 | |
| 9 | def test_probing(monkeypatch, probe_response): |
| 10 | "mock up socket registration, event selection & socket message response" |
| 11 | |
| 12 | sck = None |
| 13 | |
| 14 | def mock_register(selector, rsock, evtmask): |
| 15 | """get hold of the multicast socket that will send the Probe message, |
| 16 | when the socket is registered with the selector""" |
| 17 | global sck |
| 18 | # The Probe is sent multicast, we use that to identify the right socket. |
| 19 | # Identification could be done better, but this is enough for now. |
| 20 | if (rsock.proto == socket.IPPROTO_UDP and |
| 21 | rsock.getsockname()[1] == MULTICAST_PORT): |
| 22 | sck = rsock |
| 23 | |
| 24 | def mock_select(*args): |
| 25 | "set a mock Probe response event in motion for the same socket" |
| 26 | global sck |
| 27 | if sck and sck.getsockname()[1] == MULTICAST_PORT: |
| 28 | key = selectors.SelectorKey(sck, sck.fileno(), [], "") |
| 29 | # to mock just one response we just nullify the sock |
| 30 | sck = None |
| 31 | return [(key, selectors.EVENT_READ)] |
| 32 | else: |
| 33 | return [] |
| 34 | |
| 35 | def mock_recvfrom(*args): |
| 36 | return probe_response |
| 37 | |
| 38 | monkeypatch.setattr(selectors.DefaultSelector, "register", mock_register) |
| 39 | monkeypatch.setattr(selectors.DefaultSelector, "select", mock_select) |
| 40 | monkeypatch.setattr(socket.socket, "recvfrom", mock_recvfrom) |
| 41 | |
| 42 | # we cannot use a fixture that'd start discovery for us, since the socket |
| 43 | # selector registration happens at startup time |
| 44 | wsd = WSDiscovery() |
| 45 | wsd.start() |
| 46 | found = wsd.searchServices() |
| 47 | |
| 48 | assert len(found) == 1 |
| 49 | assert probe_response[1] in found[0].getXAddrs()[0] |
| 50 | assert len(found[0].getScopes()) == 4 |
nothing calls this directly
no test coverage detected