ive harnessed the harness
Something went wrong. Try again.
2.0 kB · 51 lines
Python
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152from __future__ import annotations
import jsonimport unittest
from peer import Peer, output
class InstructionsTests(unittest.IsolatedAsyncioTestCase): """The exact instruction text is resolved by the host, never guessed by the worker."""
async def test_instructions_returns_the_host_resolved_text(self): host_answer = { "digest": "a" * 64, "text": "you are a participant\n\ncontract v2\n", "sources": [{"name": "soul", "digest": "b" * 64, "text": "you are a participant\n"}], "resolution": { "base_digest": "c" * 64, "form": "diff", "applied": [{"source": "contract", "start_line": 1, "old_lines": ["contract v1\n"], "new_lines": ["contract v2\n"]}], }, }
async def service(message): self.assertEqual(message["request"], {"method": "runtime.instructions"}) return {"kind": "ok", "value": host_answer}
async with Peer() as peer: result = await peer.execute( "from klbr import runtime\nimport json\nprint(json.dumps(await runtime.instructions(), sort_keys=True))", service, )
self.assertEqual(result["status"], "ok") self.assertEqual(json.loads(output(result)), host_answer)
async def test_instructions_does_not_invent_a_text_when_the_host_declines(self): async def service(message): self.assertEqual(message["request"]["method"], "runtime.instructions") return {"kind": "ok", "value": {"unavailable": {"instructions": "no pinned header"}}}
async with Peer() as peer: result = await peer.execute( "from klbr import runtime\nimport json\nprint(json.dumps(await runtime.instructions()))", service, )
self.assertEqual(result["status"], "ok") answered = json.loads(output(result)) self.assertNotIn("text", answered) self.assertIn("unavailable", answered)