diff --git a/tests/pylt/service.py b/tests/pylt/service.py index a53668eb..061f0cf7 100644 --- a/tests/pylt/service.py +++ b/tests/pylt/service.py @@ -62,6 +62,15 @@ def __init__(self, *, tests: base.Tests) -> None: self.proc: typing.Optional[subprocess.Popen] = None self.failed = False + def bind_unix_socket(self, *, sockfile: str, backlog: int = 8) -> socket.socket: + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + rel_sockfile = os.path.relpath(sockfile) + if len(rel_sockfile) < len(sockfile): + sockfile = rel_sockfile + sock.bind(sockfile) + sock.listen(8) + return sock + def fork(self, *args: str, inp: typing.Union[tuple[()], FileWithFd, None] = ()) -> None: arguments = list(args) # convert tuple to list stdin: typing.Optional[FileWithFd] @@ -245,9 +254,7 @@ def __init__(self, *, tests: base.Tests) -> None: def prepare_service(self) -> None: assert self.tests self.tests.install_dir(os.path.join("tmp", "sockets")) - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - sock.bind(os.path.relpath(self.sockfile)) - sock.listen(8) + sock = self.bind_unix_socket(sockfile=self.sockfile) self.fork(*self.binary, inp=sock) def cleanup_service(self) -> None: diff --git a/tests/pylt/tests/t-memcached.py b/tests/pylt/tests/t-memcached.py index e34e4b85..08080d81 100644 --- a/tests/pylt/tests/t-memcached.py +++ b/tests/pylt/tests/t-memcached.py @@ -21,9 +21,7 @@ def __init__(self, *, tests: base.Tests) -> None: def prepare_service(self) -> None: assert self.tests self.tests.install_dir(os.path.join("tmp", "sockets")) - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - sock.bind(os.path.relpath(self.sockfile)) - sock.listen(8) + sock = self.bind_unix_socket(sockfile=self.sockfile) self.fork(*self.binary, inp=sock) def cleanup_service(self) -> None: diff --git a/tests/pylt/tests/t-scgi.py b/tests/pylt/tests/t-scgi.py index 68cc7f5e..54d24dd6 100644 --- a/tests/pylt/tests/t-scgi.py +++ b/tests/pylt/tests/t-scgi.py @@ -20,9 +20,7 @@ def __init__(self, *, tests: base.Tests) -> None: def prepare_service(self) -> None: assert self.tests self.tests.install_dir(os.path.join("tmp", "sockets")) - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - sock.bind(os.path.relpath(self.sockfile)) - sock.listen(8) + sock = self.bind_unix_socket(sockfile=self.sockfile) self.fork(*self.binary, inp=sock) def cleanup_service(self) -> None: