"""Tests for the shared skill surface: `Settled`, the generated digest, the remedies."""
from __future__ import annotations
import asyncio
import inspect
import os
import sys
import types
import unittest
SRC = os.path.join(os.path.dirname(__file__), "..", "src")
sys.path.insert(0, SRC)
from rlm import skill # noqa: E402
def fixture_module(source: str, name: str = "fixture_skill") -> types.ModuleType:
"""A skill-shaped module: functions defined in it, wrapped like the host wraps skills."""
module = types.ModuleType(name)
exec(source, module.__dict__) # noqa: S102 - the fixture's source is the test's own
module.__class__ = type("_Wrapped", (types.ModuleType,), {})
return module
class SettledTest(unittest.TestCase):
def test_settled_primitives_await_to_themselves(self):
async def probe():
return [await skill.settled("text"), await skill.settled([1, 2]), await skill.settled({"a": 1})]
values = asyncio.run(probe())
self.assertEqual(values, ["text", [1, 2], {"a": 1}])
self.assertIsInstance(values[0], str)
self.assertIsInstance(values[1], list)
self.assertIsInstance(values[2], dict)
def test_settled_twins_keep_value_semantics(self):
twin = skill.settled("hello")
self.assertIs(type(twin) is str, False)
self.assertEqual(len(twin), 5)
self.assertEqual(f"{twin}!", "hello!")
self.assertEqual(twin.upper(), "HELLO")
self.assertEqual(skill.settled([1, 2]) + [3], [1, 2, 3])
def test_double_await_returns_the_same_value(self):
async def probe():
once = await skill.settled("x")
return await once
self.assertEqual(asyncio.run(probe()), "x")
def test_result_subclass_keeps_its_type_and_repr(self):
class CommitList(list):
def __repr__(self) -> str:
return "oneline"
rows = CommitList([{"hash": "abc"}])
self.assertIs(skill.settled(rows), rows)
self.assertEqual(repr(skill.settled(rows)), "oneline")
def test_scalar_results_pass_through(self):
self.assertEqual(skill.settled(7), 7)
self.assertEqual(skill.settled((1, 2)), (1, 2))
def test_none_is_not_a_result(self):
with self.assertRaises(TypeError) as caught:
skill.settled(None)
self.assertIn("must not return None", str(caught.exception))
class SettleModuleTest(unittest.TestCase):
def settle(self, source: str) -> dict:
namespace: dict = {}
exec(source, namespace) # noqa: S102 - the fixture's source is the test's own
skill.settle_module(namespace)
return namespace
def test_every_published_call_is_settled(self):
namespace = self.settle(
"""
__all__ = ["claim", "rows", "lookup", "enrich", "Rows", "hidden"]
def claim(rev="HEAD") -> str:
return "claim"
def rows() -> list:
return [1]
def lookup() -> dict:
return {}
async def enrich(rev) -> str:
return f"enriched {rev}"
class Rows(list):
pass
def hidden() -> str:
return "not published"
"""
)
for name in ("claim", "rows", "lookup"):
self.assertTrue(inspect.isawaitable(namespace[name]()), name)
self.assertIsInstance(namespace["claim"](), str)
self.assertIsInstance(namespace["rows"](), list)
self.assertIsInstance(namespace["lookup"](), dict)
# A coroutine call stays pending: awaiting it runs the call, and the digest still marks it.
self.assertTrue(inspect.iscoroutinefunction(namespace["enrich"]))
pending = namespace["enrich"]("HEAD")
self.assertTrue(inspect.iscoroutine(pending))
pending.close()
self.assertEqual(asyncio.run(self._probe(namespace)), ["claim", [1], {}, "enriched HEAD"])
self.assertEqual(namespace["hidden"](), "not published")
self.assertTrue(isinstance(namespace["Rows"], type)) # a class is not wrapped
async def _probe(self, namespace: dict) -> list:
return [
await namespace["claim"](),
await namespace["rows"](),
await namespace["lookup"](),
await namespace["enrich"]("HEAD"),
]
def test_awaiting_a_call_equals_calling_it(self):
namespace = self.settle(
"""
__all__ = ["claim"]
def claim(rev="HEAD") -> str:
return f"claim {rev}"
"""
)
claimed = namespace["claim"]
async def probe():
return await claimed("HEAD")
self.assertEqual(asyncio.run(probe()), claimed("HEAD"))
def test_published_results_keep_the_skill_signature(self):
namespace = self.settle(
"""
__all__ = ["log"]
def log(rev="HEAD", *, limit=None, paths=None) -> str:
'''List commits.'''
return "log"
"""
)
signature = inspect.signature(namespace["log"])
self.assertEqual(list(signature.parameters), ["rev", "limit", "paths"])
self.assertEqual(namespace["log"].__name__, "log")
self.assertEqual(namespace["log"].__doc__, "List commits.")
def test_aliases_share_one_wrapper(self):
namespace = self.settle(
"""
__all__ = ["find", "search"]
def find(pattern) -> list:
return [pattern]
search = find
"""
)
self.assertIs(namespace["find"], namespace["search"])
def test_none_is_not_settled_and_stays_loud(self):
namespace = self.settle(
"""
__all__ = ["lookup"]
def lookup() -> str | None:
return None
"""
)
self.assertIsNone(namespace["lookup"]())
self.assertFalse(inspect.isawaitable(namespace["lookup"]()))
async def probe():
return await namespace["lookup"]()
with self.assertRaises(TypeError):
asyncio.run(probe())
class DigestTest(unittest.TestCase):
SOURCE = """
__all__ = ["log", "show", "Rows"]
def log(rev="HEAD", limit=None, *paths, path=None) -> str:
\"\"\"List commits: the cap is limit.\"\"\"
return "log"
async def show(rev, path) -> str:
\"\"\"Show one file at a revision.\"\"\"
return "show"
class Rows(list):
def first(self) -> str:
return "row"
def helper() -> str:
return "not declared"
"""
def surface(self, source: str = SOURCE, name: str = "fixture_skill") -> skill.Surface:
return skill.Surface([(name, fixture_module(source, name))])
def test_digest_lines_are_generated_from_the_live_signatures(self):
lines = self.surface().digest().splitlines()
self.assertTrue(lines[0].startswith("# skill surface (generated from the live modules"))
self.assertIn("fixture_skill.log(rev?='HEAD', limit?=None, *paths, path?=None) -> str", lines)
self.assertIn("fixture_skill.show(rev, path) -> str [pending]", lines)
self.assertIn("Rows: .first()", lines)
def test_digest_publishes_only_the_declared_names(self):
digest = self.surface().digest()
self.assertNotIn("fixture_skill.helper", digest)
source = self.SOURCE.replace("def helper", "def helper_renamed")
self.assertEqual(
[call.name for call in skill.Surface([("search", fixture_module(source, "search"))]).calls],
["search.log", "search.show"],
)
def test_digest_follows_a_changed_signature(self):
before = self.surface().digest()
after = self.surface(self.SOURCE.replace("limit=None", "max_count=None")).digest()
self.assertIn("limit?=None", before)
self.assertIn("max_count?=None", after)
self.assertNotIn("limit?=None", after)
def test_digest_drops_whole_modules_by_name_past_the_cap(self):
many = "\n".join(f"def call_{index}(value=None) -> str:\n return 'x'" for index in range(40))
surface = skill.Surface(
[("fixture_skill", fixture_module(f"{many}\n__all__ = {[f'call_{i}' for i in range(40)]!r}"))]
)
full = surface.digest()
self.assertIn("fixture_skill.call_39(", full)
cap = skill._MAX_DIGEST_BYTES
self.addCleanup(setattr, skill, "_MAX_DIGEST_BYTES", cap)
# Room for the named overflow line, but not for the calls themselves.
skill._MAX_DIGEST_BYTES = len(skill._DIGEST_HEADER.encode()) + 200
capped = surface.digest()
self.assertLessEqual(len(capped.encode()), skill._MAX_DIGEST_BYTES)
self.assertIn("inspect.signature(fixture_skill.<fn>)", capped)
self.assertNotIn("log(rev?=", capped)
def test_runtime_rows_publish_bash_and_its_result_shape(self):
import importlib
surface = skill.Surface(
[("bash", importlib.import_module("rlm.bash")), ("rlm", importlib.import_module("rlm").rlm)]
)
digest = surface.digest()
self.assertIn("bash(command,", digest)
self.assertIn("BashResult: ", digest)
self.assertIn("BashHandle: ", digest)
self.assertIn("rlm.spawn(prompt, *, name, model, thinking?=None, host?=None, workspace?=None, allow_push?=False, read_only?=False, deliverable?=None, evidence_bar?=None, falsifier?=None)", digest)
class RemedyTest(unittest.TestCase):
SOURCE = DigestTest.SOURCE
def setUp(self) -> None:
self.surface = skill.Surface([("fixture_skill", fixture_module(self.SOURCE))])
def test_unexpected_keyword_names_the_signature_and_the_nearest_legal_name(self):
remedy = self.surface.remedy(TypeError("log() got an unexpected keyword argument 'max_count'"))
self.assertIn("signature: fixture_skill.log(rev?='HEAD', limit?=None, *paths, path?=None) -> str", remedy)
self.assertIn("nearest: 'max_count' -> limit (fixture_skill.log); use limit=", remedy)
def test_missing_positional_names_the_required_arguments_and_the_doc_line(self):
remedy = self.surface.remedy(TypeError("show() missing 2 required positional arguments: 'rev', 'path'"))
self.assertIn("signature: fixture_skill.show(rev, path) -> str", remedy)
self.assertIn("required: rev, path must be passed and has no default", remedy)
self.assertIn("doc: Show one file at a revision.", remedy)
def test_multiple_values_names_the_once_only_binding(self):
remedy = self.surface.remedy(TypeError("log() got multiple values for argument 'rev'"))
self.assertIn("multiple values: 'rev' is bound twice; pass it once, positionally or as rev=", remedy)
def test_await_on_a_settled_value_names_the_fix(self):
remedy = self.surface.remedy(TypeError("object list can't be used in 'await' expression"))
self.assertIn("`list` is settled", remedy)
self.assertIn("drop the `await`", remedy)
def test_called_as_a_method_names_the_fix(self):
remedy = self.surface.remedy(TypeError("'str' object is not callable"))
self.assertIn("is a value, not a call: drop the `()`", remedy)
def test_wrong_module_attribute_names_the_nearest_published_call(self):
remedy = self.surface.remedy(AttributeError("module 'fixture_skill' has no attribute 'logg'"))
self.assertIn("attribute: fixture_skill has no 'logg'; nearest: fixture_skill.log", remedy)
def test_wrong_attribute_on_a_result_names_the_nearest_member_and_its_kind(self):
remedy = self.surface.remedy(AttributeError("'Rows' object has no attribute 'first_row'"))
self.assertIn("nearest: .first is a method (call it)", remedy)
def test_unknown_shape_appends_nothing(self):
self.assertEqual(self.surface.remedy(ValueError("nothing here")), "")
self.assertEqual(self.surface.remedy(TypeError("something else entirely")), "")
self.assertEqual(self.surface.remedy(AttributeError("unrelated")), "")
def test_ambiguous_names_are_bounded(self):
names = "\n".join(f"def name_{index}(value=None):\n return {index}" for index in range(6))
surface = skill.Surface(
[("fixture_skill", fixture_module(f"{names}\n__all__ = ['name_0', 'name_1', 'name_2', 'name_3']"))]
)
self.assertLessEqual(len(surface.calls), 4)
self.assertEqual(len(surface.named("name_0")), 1)
remedy = surface.remedy(TypeError("name_0() got an unexpected keyword argument 'values'"))
self.assertEqual(remedy.count("signature: "), 1)
if __name__ == "__main__":
unittest.main()