mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
1bb193c327
Test files received fixtures by importing them from a sibling test module (`from tests.x.test_y import daemon, owner # noqa: F401`). Every test that then takes the fixture as a parameter is a ruff F811 redefinition, which main's code-health ratchet blocks (it ignores noqa), and it couples test modules to each other. Move each shared fixture to where pytest resolves it by name, scopes unchanged (all function-scoped): - model_peer -> tests/_fixtures/model_peer.py, registered in tests/conftest.py (used by gateway, acp_adapter and hermes_cli tests) - daemon, server_spec -> tests/acp_adapter/conftest.py - owner, api -> tests/gateway/conftest.py - worker -> new tests/hermes_state/conftest.py - test_session_hosted_rpc's `owner` (a different object: hosted-room authority on its own loop) -> tests/gateway/conftest.py as `hosted_owner`, so it does not collide with the API `owner`; its three user files rename the parameter. Conftest fixtures import lazily so the gateway conftest still installs its telegram/discord mocks before any gateway module loads. Plain helpers (viewer, editor, control, policy) stay imported from their modules. Also drop an unused HostedRoomAttachmentStore local in test_hosted_attachment_bridge (F841) while that file is touched. Same tests collected and passing in every touched file. (cherry picked from commit 463102382f8cd9bbc9e0c605be7e1345165efda4)
24 lines
805 B
Python
24 lines
805 B
Python
"""Loopback OpenAI-compatible model peer shared by the ordinary-daemon wire tests.
|
|
|
|
Registered in ``tests/conftest.py`` (see the note there on why fixtures are imported rather than
|
|
listed in ``pytest_plugins``) so gateway, ACP and CLI tests receive ``model_peer`` by name.
|
|
"""
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def model_peer():
|
|
from http.server import ThreadingHTTPServer
|
|
import threading
|
|
from tests.gateway.fixtures.shared_authority_peer import ModelPeer
|
|
server = ThreadingHTTPServer(('127.0.0.1', 0), ModelPeer)
|
|
server.requests, server.metadata_requests = [], []
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
yield server
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=5)
|