Use a random port in MockFTPServer. (#7402)

This commit is contained in:
Andrey Rakhmatullin 2026-04-06 17:55:33 +05:00 committed by GitHub
parent f8d103a65a
commit 8a26c3c2a0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 22 additions and 5 deletions

View File

@ -1,5 +1,6 @@
from __future__ import annotations
import re
import sys
from argparse import ArgumentParser
from pathlib import Path
@ -15,20 +16,36 @@ from tests.utils import get_script_run_env
class MockFTPServer:
"""Creates an FTP server on port 2121 with a default passwordless user
"""Creates an FTP server on a random port with a default passwordless user
(anonymous) and a temporary root path that you can read from the
:attr:`path` attribute."""
def __init__(self) -> None:
self.proc: Popen[str] | None = None
self.host: str = "127.0.0.1"
self.port: int | None = None
self.path: Path | None = None
def __enter__(self):
self.path = Path(mkdtemp())
self.proc = Popen(
[sys.executable, "-u", "-m", "tests.mockserver.ftp", "-d", str(self.path)],
stderr=PIPE,
env=get_script_run_env(),
text=True,
)
for line in self.proc.stderr:
if b"starting FTP server" in line:
if "starting FTP server" in line and (
m := re.search(r"starting FTP server on ([^ :]+):(\d+),", line)
):
self.port = int(m.group(2))
break
else:
self.proc.kill()
self.proc.communicate()
raise RuntimeError(
"The FTP server failed to start or the output is unrecognized"
)
return self
def __exit__(self, exc_type, exc_value, traceback):
@ -37,12 +54,12 @@ class MockFTPServer:
self.proc.communicate()
def url(self, path):
return "ftp://127.0.0.1:2121/" + path
return f"ftp://{self.host}:{self.port}/{path}"
def main() -> None:
parser = ArgumentParser()
parser.add_argument("-d", "--directory")
parser.add_argument("-d", "--directory", required=True)
args = parser.parse_args()
authorizer = DummyAuthorizer()
@ -50,7 +67,7 @@ def main() -> None:
authorizer.add_anonymous(args.directory, perm=full_permissions)
handler = FTPHandler
handler.authorizer = authorizer
address = ("127.0.0.1", 2121)
address = ("127.0.0.1", 0)
server = FTPServer(address, handler)
server.serve_forever()