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.
99 lines
3.0 KiB
Python
99 lines
3.0 KiB
Python
import datetime
|
|
|
|
from hanzo_agents.did_manager import DIDManager, DIDIdentityPackage
|
|
|
|
|
|
def make_package():
|
|
return {
|
|
"agent_did": {
|
|
"did": "did:agent:123",
|
|
"private_key_jwk": "priv",
|
|
"public_key_jwk": "pub",
|
|
"derivation_path": "m/0",
|
|
"component_type": "agent",
|
|
},
|
|
"reasoner_dids": {
|
|
"reasoner_a": {
|
|
"did": "did:reasoner:a",
|
|
"private_key_jwk": "priv_a",
|
|
"public_key_jwk": "pub_a",
|
|
"derivation_path": "m/1",
|
|
"component_type": "reasoner",
|
|
}
|
|
},
|
|
"skill_dids": {
|
|
"skill_b": {
|
|
"did": "did:skill:b",
|
|
"private_key_jwk": "priv_b",
|
|
"public_key_jwk": "pub_b",
|
|
"derivation_path": "m/2",
|
|
"component_type": "skill",
|
|
}
|
|
},
|
|
"hanzo_agents_server_id": "hanzo_agents-1",
|
|
}
|
|
|
|
|
|
def test_register_agent_success(monkeypatch):
|
|
manager = DIDManager("http://hanzo_agents", "node")
|
|
|
|
class DummyResponse:
|
|
status_code = 200
|
|
|
|
@staticmethod
|
|
def json():
|
|
return {"success": True, "identity_package": make_package()}
|
|
|
|
monkeypatch.setattr("requests.post", lambda *a, **k: DummyResponse())
|
|
|
|
ok = manager.register_agent([], [])
|
|
assert ok is True
|
|
assert manager.is_enabled() is True
|
|
assert manager.get_agent_did() == "did:agent:123"
|
|
assert manager.get_function_did("reasoner_a") == "did:reasoner:a"
|
|
assert manager.get_function_did("unknown") == "did:agent:123"
|
|
|
|
|
|
def test_register_agent_failure_status(monkeypatch):
|
|
manager = DIDManager("http://hanzo_agents", "node")
|
|
|
|
class DummyResponse:
|
|
status_code = 500
|
|
text = "boom"
|
|
|
|
monkeypatch.setattr("requests.post", lambda *a, **k: DummyResponse())
|
|
ok = manager.register_agent([], [])
|
|
assert ok is False
|
|
assert manager.is_enabled() is False
|
|
|
|
|
|
def test_create_execution_context(monkeypatch):
|
|
manager = DIDManager("http://hanzo_agents", "node")
|
|
package = manager._parse_identity_package(make_package())
|
|
assert isinstance(package, DIDIdentityPackage)
|
|
manager.identity_package = package
|
|
manager.enabled = True
|
|
|
|
ctx = manager.create_execution_context(
|
|
execution_id="exec1",
|
|
workflow_id="wf1",
|
|
session_id="sess",
|
|
caller_function="reasoner_a",
|
|
target_function="skill_b",
|
|
)
|
|
assert ctx is not None
|
|
assert ctx.caller_did == "did:reasoner:a"
|
|
assert ctx.target_did == "did:skill:b"
|
|
assert isinstance(ctx.timestamp, datetime.datetime)
|
|
|
|
|
|
def test_create_execution_context_missing_identity():
|
|
manager = DIDManager("http://hanzo_agents", "node")
|
|
assert manager.create_execution_context("e", "w", "s", "a", "b") is None
|
|
|
|
|
|
def test_get_identity_summary_disabled():
|
|
manager = DIDManager("http://hanzo_agents", "node")
|
|
summary = manager.get_identity_summary()
|
|
assert summary["enabled"] is False
|