ive harnessed the harness
Something went wrong. Try again.
3.0 kB · 69 lines
Python
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970from __future__ import annotations
import jsonimport unittest
from peer import Peer, output
class EnvironmentDiscoverabilityTests(unittest.IsolatedAsyncioTestCase): async def test_inspection_uses_host_manifest_not_worker_cwd(self): """The normal SDK must expose the host response, not infer identity locally.""" host_manifest = { "source_root": {"path": "/truthful/workspace", "exists": True, "writable": False}, "draft_root": {"path": "/truthful/workspace/.klbr/drafts", "target": {"id": "target-remote", "workspace": "/truthful/workspace"}, "exists": False}, "target": {"id": "target-remote", "workspace": "/truthful/workspace"}, "selected_behavior_revision": "release-selected", "selected_behavior_epoch": 7, "applied_behavior_revision": "release-applied", "purpose": "interactive", "cell_budget": {"limit_ms": 1700, "remaining_ms": 1337}, }
async def service(message): self.assertEqual(message["request"], {"method": "runtime.inspect"}) return {"kind": "ok", "value": host_manifest}
async with Peer() as peer: result = await peer.execute( "from klbr import runtime\nimport json\nprint(json.dumps(await runtime.inspect(), sort_keys=True))", service, )
self.assertEqual(result["status"], "ok") inspected = json.loads(output(result)) self.assertEqual(inspected, host_manifest) self.assertNotEqual(inspected["source_root"]["path"], str(peer.directory)) self.assertEqual(inspected["draft_root"]["path"], "/truthful/workspace/.klbr/drafts") self.assertEqual(inspected["selected_behavior_revision"], "release-selected") self.assertEqual(inspected["applied_behavior_revision"], "release-applied") self.assertNotIn("api_key", inspected) secret = "credential-value-must-not-appear" self.assertNotIn(secret, json.dumps(inspected))
async def test_inspection_does_not_turn_unavailable_fields_into_guesses(self): async def service(message): self.assertEqual(message["request"]["method"], "runtime.inspect") return { "kind": "ok", "value": { "draft_root": {"path": "/truthful/workspace/.klbr/drafts", "exists": False}, "unavailable": {"cumulative_model_budget": "not supplied"}, }, }
async with Peer() as peer: result = await peer.execute( "from klbr import runtime\nimport json\nprint(json.dumps(await runtime.inspect()))", service, )
inspected = json.loads(output(result)) self.assertEqual(inspected["draft_root"]["exists"], False) self.assertNotIn("draft_root", inspected["unavailable"]) self.assertNotIn("cwd", inspected) self.assertNotIn("ps", inspected)
if __name__ == "__main__": unittest.main()