mirror of
https://github.com/hanzoai/agents.git
synced 2026-10-30 18:48:36 +00:00
zeekayandGitHub
5b5cf9e9db
Transforms hanzoai/agents from a marketplace config repo into a full agent orchestration platform by merging the AgentField infrastructure: - Go control plane (K8s-like agent orchestration, service mesh, DAG workflows) - Python SDK (FastAPI-based agent framework, LiteLLM, memory, DID/VC) - TypeScript SDK (@hanzo/agents - Vercel AI SDK, MCP, workflows) - Go SDK (idiomatic agent builder + client) - React 19 embedded dashboard (XyFlow DAGs, Recharts, Radix UI) - Docker/K8s/Helm deployment configs - Examples: hello world, deep research, RAG, simulation, benchmarks - Functional test suite across Python/Go/TypeScript All references rebranded: AgentField → Hanzo Agents - Module: github.com/hanzoai/agents/control-plane - Python: hanzo_agents - TypeScript: @hanzo/agents - Go: github.com/hanzoai/agents/sdk/go - Env vars: HANZO_AGENTS_* - Config: hanzo-agents.yaml Previous marketplace configs moved to config/ directory.
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)
|