Refactor opsqueue_service to use wait_for_server for port assignment and add utility function for server wait logic #34
Conversation
…nment and add utility function for server wait logic
There was a problem hiding this comment.
Pull request overview
This PR refactors the test infrastructure to dynamically detect server ports rather than pre-assigning them. The main change introduces a wait_for_server utility function that monitors a process until it starts listening on a port, eliminating the need for the previous random_free_port() approach.
Key Changes:
- Added a new
wait_for_server()utility function that polls a process usingpsutilto detect when it starts listening on a port - Modified
opsqueue_service()to usepsutil.Popeninstead ofsubprocess.Popenand dynamically discover the server port viawait_for_server() - Changed the default port parameter from
Noneto0, which allows the OS to assign an available port that is then discovered
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 7 comments.
| File | Description |
|---|---|
| libs/opsqueue_python/tests/util.py | New utility module containing wait_for_server() function with retry logic to wait for a process to start listening on a port |
| libs/opsqueue_python/tests/conftest.py | Updated opsqueue_service() to use psutil.Popen and call wait_for_server() for dynamic port discovery instead of pre-assigning ports |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
|
||
|
|
||
| @retry( | ||
| retry_on_exceptions=ValueError, |
There was a problem hiding this comment.
The RuntimeError exceptions raised for unexpected process states (lines 43, 45-47) and for multiple listening ports (line 57) are not configured to be retried by the @retry decorator, which only retries on ValueError. If a process temporarily has access issues before it's fully started, or if there's a race condition where multiple ports are briefly detected, these will immediately fail without retry. Consider whether these error conditions should also be retried or if the retry logic should be adjusted.
| retry_on_exceptions=ValueError, | |
| retry_on_exceptions=(ValueError, RuntimeError), |
| with psutil.Popen(command, cwd=PROJECT_ROOT, env=env) as process: | ||
| _host, port = wait_for_server(process) | ||
| try: | ||
| wrapper = OpsqueueProcess(port=port, process=process) |
There was a problem hiding this comment.
The process type has been changed from subprocess.Popen to psutil.Popen, but the OpsqueueProcess dataclass still expects subprocess.Popen[bytes]. This type mismatch should be addressed by updating the dataclass type annotation to match. Note: psutil.Popen is a subclass of subprocess.Popen, so the code will work at runtime, but type checkers may report errors.
| import subprocess | ||
| import uuid | ||
| import os | ||
| from libs.opsqueue_python.tests.util import wait_for_server |
There was a problem hiding this comment.
The import statement uses an absolute path starting with libs.opsqueue_python.tests.util which might not work correctly depending on how the Python path is configured. Consider using a relative import like from .util import wait_for_server or from tests.util import wait_for_server instead, which would be more robust to different project structures and import contexts.
| from libs.opsqueue_python.tests.util import wait_for_server | |
| from .util import wait_for_server |
| Wait for a process to be listening on a single port. | ||
| If the process is listening on no ports, a ValueError is thrown and this is retried. | ||
| If multiple ports are listening, a RuntimeError is thrown. |
There was a problem hiding this comment.
The docstring states "If multiple ports are listening, a RuntimeError is thrown" but doesn't explain why this is considered an error condition. Since the function is designed to wait for servers that should listen on a single port, it would be helpful to document this assumption more clearly, such as: "This function expects the process to listen on exactly one port. If multiple ports are listening, a RuntimeError is raised as this indicates an unexpected server configuration."
| Wait for a process to be listening on a single port. | |
| If the process is listening on no ports, a ValueError is thrown and this is retried. | |
| If multiple ports are listening, a RuntimeError is thrown. | |
| Wait for a process to be listening on exactly one port and return that address. | |
| This function expects the given process to listen on a single port only. If the | |
| process is listening on no ports, a ValueError is raised and the check is retried. | |
| If multiple ports are listening, a RuntimeError is raised as this indicates an | |
| unexpected server configuration. |
|
|
||
| import psutil | ||
| from opnieuw import retry | ||
| from psutil._common import pconn |
There was a problem hiding this comment.
Importing from psutil._common accesses a private/internal module (indicated by the underscore prefix). This could break in future versions of psutil as internal APIs are not guaranteed to be stable. Consider using public psutil APIs instead, or be aware that this might require changes when psutil is updated.
| *, port: int | None = None | ||
| *, port: int = 0, | ||
| ) -> Generator[OpsqueueProcess, None, None]: | ||
| global test_opsqueue_port_offset |
There was a problem hiding this comment.
The global variable test_opsqueue_port_offset is declared but never actually used in this function. This appears to be leftover from the previous implementation. Consider removing this line to avoid confusion.
| global test_opsqueue_port_offset |
| def wait_for_server(proc: psutil.Popen) -> tuple[str, int]: | ||
| """ | ||
| Wait for a process to be listening on a single port. | ||
| If the process is listening on no ports, a ValueError is thrown and this is retried. | ||
| If multiple ports are listening, a RuntimeError is thrown. | ||
| """ | ||
| if not proc.is_running(): | ||
| raise ValueError(f"Process {proc} is not running") | ||
|
|
||
| try: | ||
| # Try to get the connections of the main process first, if that fails try the children. | ||
| # Processes wrapped with `timeout` do not have connections themselves. | ||
| connections: list[pconn] = ( | ||
| proc.net_connections() | ||
| or [ | ||
| child_conn | ||
| for child in proc.children(recursive=False) | ||
| for child_conn in child.net_connections() | ||
| ] | ||
| or [ | ||
| child_conn | ||
| for child in proc.children(recursive=True) | ||
| for child_conn in child.net_connections() | ||
| ] | ||
| ) | ||
| except psutil.AccessDenied as e: | ||
| match proc.status(): | ||
| case psutil.STATUS_ZOMBIE | psutil.STATUS_DEAD | psutil.STATUS_STOPPED: | ||
| raise RuntimeError(f"Process {proc} has exited unexpectedly") from e | ||
| case _: | ||
| raise RuntimeError( | ||
| f"Could not get `net_connections` for process {proc}, access denied " | ||
| ) from e | ||
|
|
||
| ports = [x for x in connections if x.status == psutil.CONN_LISTEN] | ||
| listen_count = len(ports) | ||
|
|
||
| if listen_count == 0: | ||
| raise ValueError(f"Process {proc} is not listening on any ports") | ||
| if listen_count == 1: | ||
| return ports[0].laddr | ||
|
|
||
| raise RuntimeError(f"Process {proc} is listening on multiple ports") |
There was a problem hiding this comment.
The new wait_for_server utility function lacks test coverage. Given that it contains complex logic with multiple error paths (process not running, access denied, zombie/dead status, no ports listening, multiple ports), this function should have dedicated unit tests to ensure reliability. The repository appears to use comprehensive testing (test files exist in the tests directory), so this new utility should follow the same pattern.
1 participant
Refactor opsqueue_service to use wait_for_server for port assignment and add utility function for server wait logic.