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
|
import socket
from contextlib import closing
import pytest
from xprocess import ProcessStarter
pytest_plugins = "pytester"
@pytest.fixture
def example(xprocess):
"""fixture for testing ResourceWarnings
in test_resource_cleanup.py module"""
class Starter(ProcessStarter):
pattern = "foo"
args = ("sh", "-c", "echo foo; sleep 10; echo bar")
xprocess.ensure("example", Starter)
yield
xprocess.getinfo("example").terminate()
@pytest.fixture
def tcp_port():
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
s.bind(("", 0))
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
return s.getsockname()[1]
|
