Compare commits

...

1 Commits

Author SHA1 Message Date
Hermes Agent
63b2d444d5 fix: test collection errors in test_bannerlord_harness and test_evennia_ws_bridge (#1509)
Some checks are pending
CI / test (pull_request) Waiting to run
CI / validate (pull_request) Waiting to run
Review Approval Gate / verify-review (pull_request) Waiting to run
Resolves #1509. Two test files were failing to collect during pytest.

nexus/evennia_ws_bridge.py:
- Extracted clean_lines(), parse_room_output(), normalize_event()
  from playback() closure to module-level functions
- Tests import these as module-level symbols; they were buried
  inside the playback() function and inaccessible

nexus/bannerlord_harness.py:
- Fixed import: 'from bannerlord_trace' to 'from nexus.bannerlord_trace'

Results:
- pytest test_evennia_ws_bridge.py: 5 passed
- pytest test_bannerlord_harness.py: 37 passed, 2 skipped
- 44 tests now running that were previously broken
2026-04-14 23:11:37 -04:00
2 changed files with 58 additions and 1 deletions

View File

@@ -29,7 +29,7 @@ from typing import Any, Callable, Optional
import websockets
from bannerlord_trace import BannerlordTraceLogger
from nexus.bannerlord_trace import BannerlordTraceLogger
# ═══════════════════════════════════════════════════════════════════════════
# CONFIGURATION

View File

@@ -181,6 +181,63 @@ async def live_bridge(log_dir: str, ws_url: str, reconnect_delay: float = 5.0):
await asyncio.gather(*tasks)
def clean_lines(text: str) -> list[str]:
"""Strip ANSI, normalize line endings, return non-empty lines."""
text = strip_ansi(text).replace("\r", "")
return [line.strip() for line in text.split("\n") if line.strip()]
def parse_room_output(text: str) -> dict:
"""Parse Evennia room text into structured data (title, desc, exits, objects)."""
lines = clean_lines(text)
if len(lines) < 2:
return {"title": lines[0] if lines else "", "desc": "", "exits": [], "objects": []}
title = lines[0]
desc = lines[1]
exits = []
objects = []
for line in lines[2:]:
if line.startswith("Exits:"):
raw = line.split(":", 1)[1].strip().replace(" and ", ", ")
exits = [{"key": t.strip(), "destination_id": t.strip().title(), "destination_key": t.strip().title()} for t in raw.split(",") if t.strip()]
elif line.startswith("You see:"):
raw = line.split(":", 1)[1].strip().replace(" and ", ", ")
parts = [t.strip() for t in raw.split(",") if t.strip()]
objects = [{"id": p.removeprefix("a ").removeprefix("an "), "key": p.removeprefix("a ").removeprefix("an "), "short_desc": p} for p in parts]
return {"title": title, "desc": desc, "exits": exits, "objects": objects}
def normalize_event(raw: dict, hermes_session_id: str) -> list[dict]:
"""Convert raw Evennia event dict into normalized Nexus events."""
from nexus.evennia_event_adapter import (
actor_located, command_issued, command_result,
room_snapshot, session_bound,
)
out = []
event = raw.get("event")
actor = raw.get("actor", "Timmy")
timestamp = raw.get("timestamp")
if event == "connect":
out.append(session_bound(hermes_session_id, evennia_account=actor, evennia_character=actor, timestamp=timestamp))
parsed = parse_room_output(raw.get("output", ""))
if parsed:
out.append(actor_located(actor, parsed["title"], parsed["title"], timestamp=timestamp))
out.append(room_snapshot(parsed["title"], parsed["title"], parsed["desc"], exits=parsed["exits"], objects=parsed["objects"], timestamp=timestamp))
elif event == "command":
cmd = raw.get("command", "")
output = raw.get("output", "")
out.append(command_issued(hermes_session_id, actor, cmd, timestamp=timestamp))
success = not output.startswith("Command '") and not output.startswith("Could not find")
out.append(command_result(hermes_session_id, actor, cmd, strip_ansi(output), success=success, timestamp=timestamp))
parsed = parse_room_output(output)
if parsed:
out.append(actor_located(actor, parsed["title"], parsed["title"], timestamp=timestamp))
out.append(room_snapshot(parsed["title"], parsed["title"], parsed["desc"], exits=parsed["exits"], objects=parsed["objects"], timestamp=timestamp))
return out
async def playback(log_path: Path, ws_url: str):
"""Legacy mode: replay a telemetry JSONL file."""
from nexus.evennia_event_adapter import (