mirror of
https://github.com/hanzoai/agent.git
synced 2026-09-11 21:08:51 +00:00
zeekay
ff7ef7114f
Consolidate all Hanzo agent infrastructure into one repo: - control-plane/: Go control plane server (was agents-new) - sdk/: Go, Python, TypeScript SDKs - deployments/: Docker, Kubernetes, Helm configs - Dockerfile.sdk: Python SDK image (ghcr.io/hanzoai/agent:latest) - CI builds both SDK + control-plane images No more agentfield — everything is hanzoai/agent.
55 lines
1.3 KiB
Python
55 lines
1.3 KiB
Python
import socket
|
|
import pytest
|
|
|
|
from hanzo_agents.utils import get_free_port
|
|
|
|
|
|
def test_get_free_port_iterates_until_success(monkeypatch):
|
|
attempts = [OSError, None]
|
|
|
|
class DummySocket:
|
|
def __init__(self, outcome):
|
|
self._outcome = outcome
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def bind(self, addr):
|
|
if self._outcome is OSError:
|
|
raise OSError("port in use")
|
|
|
|
real_socket = socket.socket
|
|
|
|
def fake_socket(*args, **kwargs):
|
|
if attempts:
|
|
outcome = attempts.pop(0)
|
|
return DummySocket(outcome)
|
|
return real_socket(*args, **kwargs)
|
|
|
|
monkeypatch.setattr("hanzo_agents.utils.socket.socket", fake_socket)
|
|
|
|
port = get_free_port(start_port=1000, end_port=1001)
|
|
assert port == 1001
|
|
|
|
|
|
def test_get_free_port_raises_when_exhausted(monkeypatch):
|
|
class DummySocket:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def bind(self, addr):
|
|
raise OSError("busy")
|
|
|
|
monkeypatch.setattr(
|
|
"hanzo_agents.utils.socket.socket", lambda *args, **kwargs: DummySocket()
|
|
)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
get_free_port(start_port=5, end_port=6)
|