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
|
import socket
import contextlib
import pytest
from tempora import timing
import portend
def socket_infos():
"""
Generate addr infos for connections to localhost
"""
host = None # all available interfaces
port = portend.find_available_local_port()
family = socket.AF_UNSPEC
socktype = socket.SOCK_STREAM
proto = 0
flags = socket.AI_PASSIVE
return socket.getaddrinfo(host, port, family, socktype, proto, flags)
def id_for_info(info):
af, = info[:1]
return str(af)
def build_addr_infos():
params = list(socket_infos())
ids = list(map(id_for_info, params))
return locals()
@pytest.fixture(**build_addr_infos())
def listening_addr(request):
af, socktype, proto, canonname, sa = request.param
sock = socket.socket(af, socktype, proto)
sock.bind(sa)
sock.listen(5)
with contextlib.closing(sock):
yield sa
@pytest.fixture(**build_addr_infos())
def nonlistening_addr(request):
af, socktype, proto, canonname, sa = request.param
return sa
@pytest.fixture
def immediate_timeout(monkeypatch):
monkeypatch.setattr(timing.Timer, 'expired', lambda: True)
class TestChecker:
def test_check_port_listening(self, listening_addr):
with pytest.raises(portend.PortNotFree):
portend.Checker().assert_free(listening_addr)
def test_check_port_nonlistening(self, nonlistening_addr):
portend.Checker().assert_free(nonlistening_addr)
def test_free_with_immediate_timeout(self, nonlistening_addr, immediate_timeout):
host, port = nonlistening_addr[:2]
portend.free(host, port, timeout=1.0)
def test_occupied_with_immediate_timeout(self, listening_addr, immediate_timeout):
host, port = listening_addr[:2]
portend.occupied(host, port, timeout=1.0)
|
