from __future__ import annotations
import json
import 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)