307 lines
10 KiB
Python
307 lines
10 KiB
Python
# ruff: noqa: INP001
|
|
"""Unit tests for runtime_activity service helpers.
|
|
|
|
All tests are pure-Python — no gateway connection required.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from app.services.openclaw.runtime_activity import (
|
|
RuntimeMessageEvent,
|
|
_collect_tool_names,
|
|
_correlate_session,
|
|
_extract_text,
|
|
_parse_timestamp,
|
|
normalize_message,
|
|
redact_tool_args,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _extract_text
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestExtractText:
|
|
|
|
def test_string_content_short(self):
|
|
text, truncated = _extract_text("Hello world")
|
|
assert text == "Hello world"
|
|
assert not truncated
|
|
|
|
def test_string_content_truncated(self):
|
|
long_text = "x" * 400
|
|
text, truncated = _extract_text(long_text)
|
|
assert len(text) == 300
|
|
assert truncated
|
|
|
|
def test_none_content(self):
|
|
text, truncated = _extract_text(None)
|
|
assert text == ""
|
|
assert not truncated
|
|
|
|
def test_list_text_block(self):
|
|
content = [{"type": "text", "text": "The answer is 42."}]
|
|
text, truncated = _extract_text(content)
|
|
assert "42" in text
|
|
assert not truncated
|
|
|
|
def test_list_tool_use_block(self):
|
|
content = [
|
|
{"type": "text", "text": "I'll run a command."},
|
|
{"type": "tool_use", "name": "bash", "input": {"command": "ls -la"}},
|
|
]
|
|
text, _ = _extract_text(content)
|
|
assert "run a command" in text
|
|
assert "[tool: bash]" in text
|
|
|
|
def test_list_tool_result_string(self):
|
|
content = [
|
|
{"type": "tool_result", "tool_use_id": "x", "content": "file1.txt\nfile2.txt"},
|
|
]
|
|
text, _ = _extract_text(content)
|
|
assert "[result:" in text
|
|
|
|
def test_list_multiple_text_blocks(self):
|
|
content = [
|
|
{"type": "text", "text": "First."},
|
|
{"type": "text", "text": "Second."},
|
|
]
|
|
text, _ = _extract_text(content)
|
|
assert "First" in text
|
|
assert "Second" in text
|
|
|
|
def test_empty_list(self):
|
|
text, truncated = _extract_text([])
|
|
assert text == ""
|
|
assert not truncated
|
|
|
|
def test_non_dict_blocks_skipped(self):
|
|
content = ["plain string", {"type": "text", "text": "valid"}]
|
|
text, _ = _extract_text(content)
|
|
assert "valid" in text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# redact_tool_args
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRedactToolArgs:
|
|
|
|
def test_no_sensitive_keys(self):
|
|
args = {"command": "ls -la", "cwd": "/home"}
|
|
result = redact_tool_args(args)
|
|
assert result["command"] == "ls -la"
|
|
assert result["cwd"] == "/home"
|
|
|
|
def test_password_redacted(self):
|
|
result = redact_tool_args({"password": "secret123", "user": "admin"})
|
|
assert result["password"] == "[REDACTED]"
|
|
assert result["user"] == "admin"
|
|
|
|
def test_token_redacted(self):
|
|
result = redact_tool_args({"token": "sk-abc123", "action": "read"})
|
|
assert result["token"] == "[REDACTED]"
|
|
|
|
def test_api_key_redacted(self):
|
|
result = redact_tool_args({"api_key": "abc", "model": "gpt-4"})
|
|
assert result["api_key"] == "[REDACTED]"
|
|
|
|
def test_case_insensitive(self):
|
|
result = redact_tool_args({"PASSWORD": "x", "Secret": "y"})
|
|
assert result["PASSWORD"] == "[REDACTED]"
|
|
assert result["Secret"] == "[REDACTED]"
|
|
|
|
def test_long_string_truncated(self):
|
|
result = redact_tool_args({"content": "A" * 1000})
|
|
assert len(result["content"]) < 1000
|
|
assert "truncated" in result["content"]
|
|
|
|
def test_empty_dict(self):
|
|
assert redact_tool_args({}) == {}
|
|
|
|
def test_non_dict_returns_empty(self):
|
|
assert redact_tool_args("bad") == {} # type: ignore[arg-type]
|
|
|
|
def test_credentials_key_redacted(self):
|
|
result = redact_tool_args({"credentials": "AKIA...", "region": "us-east-1"})
|
|
assert result["credentials"] == "[REDACTED]"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _collect_tool_names
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCollectToolNames:
|
|
|
|
def test_no_tool_use(self):
|
|
assert _collect_tool_names("plain text") == []
|
|
|
|
def test_single_tool_use(self):
|
|
content = [{"type": "tool_use", "name": "bash", "input": {}}]
|
|
assert _collect_tool_names(content) == ["bash"]
|
|
|
|
def test_multiple_tool_uses(self):
|
|
content = [
|
|
{"type": "text", "text": "hi"},
|
|
{"type": "tool_use", "name": "read_file", "input": {}},
|
|
{"type": "tool_use", "name": "bash", "input": {}},
|
|
]
|
|
names = _collect_tool_names(content)
|
|
assert names == ["read_file", "bash"]
|
|
|
|
def test_no_name_fallback(self):
|
|
content = [{"type": "tool_use"}]
|
|
assert _collect_tool_names(content) == ["unknown"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _parse_timestamp
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestParseTimestamp:
|
|
|
|
def test_iso_zulu(self):
|
|
msg = {"timestamp": "2026-05-21T10:00:00Z"}
|
|
ts = _parse_timestamp(msg)
|
|
assert isinstance(ts, datetime)
|
|
assert ts.year == 2026
|
|
assert ts.hour == 10
|
|
|
|
def test_iso_with_offset(self):
|
|
msg = {"timestamp": "2026-05-21T12:00:00+02:00"}
|
|
ts = _parse_timestamp(msg)
|
|
assert ts is not None
|
|
assert ts.hour == 10 # converted to UTC
|
|
|
|
def test_created_at_fallback(self):
|
|
msg = {"created_at": "2026-05-21T09:30:00Z"}
|
|
ts = _parse_timestamp(msg)
|
|
assert ts is not None
|
|
|
|
def test_no_timestamp_returns_none(self):
|
|
assert _parse_timestamp({}) is None
|
|
|
|
def test_malformed_returns_none(self):
|
|
assert _parse_timestamp({"timestamp": "not-a-date"}) is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _correlate_session
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCorrelateSession:
|
|
|
|
def test_lead_session_extracts_board_id(self):
|
|
board_id = "d8ec2aa9-fa86-4ab2-9f17-6518ccd600df"
|
|
agent_slug, bid = _correlate_session(f"agent:lead-{board_id}:main")
|
|
assert agent_slug is None
|
|
assert bid == board_id
|
|
|
|
def test_agent_session_extracts_slug(self):
|
|
slug, bid = _correlate_session("agent:my-agent:main")
|
|
assert slug == "my-agent"
|
|
assert bid is None
|
|
|
|
def test_unknown_format_returns_nones(self):
|
|
slug, bid = _correlate_session("some-random-key")
|
|
assert slug is None
|
|
assert bid is None
|
|
|
|
def test_empty_returns_nones(self):
|
|
slug, bid = _correlate_session("")
|
|
assert slug is None
|
|
assert bid is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# normalize_message
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestNormalizeMessage:
|
|
|
|
def _make_msg(self, **kwargs) -> dict:
|
|
return {
|
|
"role": "assistant",
|
|
"content": "Hello!",
|
|
"model": "claude-sonnet-4-6",
|
|
**kwargs,
|
|
}
|
|
|
|
def test_basic_assistant_message(self):
|
|
msg = self._make_msg()
|
|
event = normalize_message("agent:test:main", "Test", msg, 0)
|
|
assert isinstance(event, RuntimeMessageEvent)
|
|
assert event.role == "assistant"
|
|
assert event.model == "claude-sonnet-4-6"
|
|
assert event.content_preview == "Hello!"
|
|
assert not event.content_truncated
|
|
assert not event.has_tool_use
|
|
assert event.session_key == "agent:test:main"
|
|
assert event.session_label == "Test"
|
|
assert event.message_index == 0
|
|
|
|
def test_user_message(self):
|
|
msg = {"role": "user", "content": "What time is it?"}
|
|
event = normalize_message("session:1", None, msg, 1)
|
|
assert event.role == "user"
|
|
assert event.model is None
|
|
assert event.session_label is None
|
|
|
|
def test_tool_use_detected(self):
|
|
msg = {
|
|
"role": "assistant",
|
|
"content": [
|
|
{"type": "text", "text": "Running command."},
|
|
{"type": "tool_use", "name": "bash", "input": {"command": "ls"}},
|
|
],
|
|
}
|
|
event = normalize_message("s", None, msg, 0)
|
|
assert event.has_tool_use
|
|
assert "bash" in event.tool_names
|
|
|
|
def test_long_content_truncated(self):
|
|
long = "w" * 400
|
|
msg = self._make_msg(content=long)
|
|
event = normalize_message("s", None, msg, 0)
|
|
assert event.content_truncated
|
|
assert len(event.content_preview) == 300
|
|
|
|
def test_event_id_is_stable(self):
|
|
msg = self._make_msg()
|
|
e1 = normalize_message("s", None, msg, 0)
|
|
e2 = normalize_message("s", None, msg, 0)
|
|
assert e1.event_id == e2.event_id
|
|
|
|
def test_event_id_differs_by_index(self):
|
|
msg = self._make_msg()
|
|
e1 = normalize_message("s", None, msg, 0)
|
|
e2 = normalize_message("s", None, msg, 1)
|
|
assert e1.event_id != e2.event_id
|
|
|
|
def test_board_id_from_lead_session(self):
|
|
board_id = "d8ec2aa9-fa86-4ab2-9f17-6518ccd600df"
|
|
msg = self._make_msg()
|
|
event = normalize_message(f"agent:lead-{board_id}:main", None, msg, 0)
|
|
assert event.board_id == board_id
|
|
assert event.agent_id is None
|
|
|
|
def test_to_dict_serialisable(self):
|
|
msg = self._make_msg()
|
|
event = normalize_message("s", "Label", msg, 0)
|
|
d = event.to_dict()
|
|
assert d["role"] == "assistant"
|
|
assert d["session_label"] == "Label"
|
|
assert isinstance(d["tool_names"], list)
|
|
# timestamp is None here since no timestamp in msg
|
|
assert d["timestamp"] is None
|
|
|
|
def test_timestamp_parsed(self):
|
|
msg = self._make_msg(timestamp="2026-05-21T10:00:00Z")
|
|
event = normalize_message("s", None, msg, 0)
|
|
assert event.timestamp is not None
|
|
assert event.timestamp.hour == 10
|