scp_cli_check.py (15008B)
1 #!/usr/bin/env python3 2 """Cheap native CLI/protocol tests, separate from SCP compiler integration.""" 3 import json 4 import shutil 5 import os 6 from pathlib import Path 7 import subprocess 8 import sys 9 import tempfile 10 import unittest 11 12 REPO = Path(__file__).resolve().parent.parent 13 TRICU = Path(sys.argv[1] if len(sys.argv) > 1 else REPO / "result/bin/tricu").resolve() 14 PIN = "0123456789abcdef" * 4 15 16 class CommandTests(unittest.TestCase): 17 def setUp(self): 18 self.temp = tempfile.TemporaryDirectory(prefix="scp-cli-") 19 self.root = Path(self.temp.name) 20 self.runner = self.root / "runner" 21 self.runner.write_text(f"#!{sys.executable}\n" + 22 "import json, os, pathlib, sys, time\n" 23 "if os.getenv('SCP_TEST_SLEEP'): time.sleep(10)\n" 24 "args=sys.argv[1:]\n" 25 "if args[0] == 'arboricx':\n" 26 " pathlib.Path(os.environ['SCP_TEST_CAPTURE']).with_suffix('.args').write_text(json.dumps(args))\n" 27 " print('pinned-run-ok'); sys.exit(0)\n" 28 "pathlib.Path(os.environ['SCP_TEST_CAPTURE']).write_text(json.dumps(args))\n" 29 "for field in ('--input-file', '--input-tree-file', '--state-file', '--state-tree-file'):\n" 30 " if field in args: pathlib.Path(os.environ['SCP_TEST_CAPTURE']).with_suffix('.data').write_bytes(pathlib.Path(args[args.index(field)+1]).read_bytes())\n" 31 "if os.getenv('SCP_TEST_FAIL') and os.getenv('SCP_TEST_FAIL_PHASE', '') in args[1]: print('configuration-error'); sys.exit(0)\n" 32 f"pin={PIN!r}\n" 33 "if os.getenv('SCP_TEST_NO_FIELDS'): print('file-build-ok'); sys.exit(0)\n" 34 "if args[1] == 'build':\n" 35 " if os.getenv('SCP_TEST_DUPLICATE'): print('code-manifest='+pin)\n" 36 " print('code-manifest='+pin); print('code-receipt='+pin); print('code-build-ok')\n" 37 "elif args[1] in ('configure', 'configure-entry'):\n" 38 " print('workers='+pin); print('invocations='+pin); print('configuration-receipt='+pin); print('configuration-ok')\n" 39 "elif args[1] == 'select': print('invocations='+pin); print('invocation-ready')\n" 40 "else: print('file-manifest='+pin); print('file-build-ok')\n") 41 self.runner.chmod(0o755) 42 self.capture = self.root / "source" 43 self.bootstrap = self.root / "bootstrap" 44 self.bootstrap.mkdir() 45 for source, target in [('compiler-library.arboricx','compiler-bootstrap.arboricx'), 46 ('support-library.arboricx','scp-bootstrap.arboricx')]: 47 shutil.copyfile(REPO / 'test/fixtures' / source, self.bootstrap / target) 48 49 def tearDown(self): 50 self.temp.cleanup() 51 52 def call(self, command, *options, **environment): 53 env = dict(os.environ, SCP_TEST_CAPTURE=str(self.capture), **environment) 54 return subprocess.run([str(TRICU), "scp", command, "--bootstrap", str(self.bootstrap), 55 "--store", str(self.root / 'store with "quotes"'), "--seconds", "1", 56 "--tricu", str(self.runner), *options], cwd=self.root, 57 env=env, text=True, capture_output=True) 58 59 def test_build_and_quoting(self): 60 result = self.call("build", "--directory", str(REPO), "--workspace", 61 "demos.consumer-prefix-workspace", "--compiler", PIN) 62 self.assertEqual(result.returncode, 0, result.stderr) 63 self.assertIn(str(self.root / 'store with "quotes"'), json.loads(self.capture.read_text())) 64 self.assertEqual(result.stdout, f"code-manifest={PIN}\ncode-receipt={PIN}\n") 65 66 def test_explicit_configuration_fields(self): 67 result = self.call("configure", "--code", PIN, "--entry", "configurePrefix", "--input-leaf") 68 self.assertNotEqual(result.returncode, 0) 69 self.assertFalse(self.capture.exists()) 70 result = self.call("configure", "--code", "mutable-alias", "--entry", "configurePrefix", 71 "--input-leaf", "--state-leaf") 72 self.assertNotEqual(result.returncode, 0) 73 74 def test_bytes_are_not_source(self): 75 prefix = self.root / 'prefix "quotes"' 76 prefix.write_bytes(b'ERROR " \\ !source untrusted\n') 77 result = self.call("configure", "--code", PIN, "--entry", "configurePrefix", 78 "--input-file", str(prefix), "--state-leaf") 79 self.assertEqual(result.returncode, 0, result.stderr) 80 source = self.capture.read_text() 81 self.assertNotIn("!source untrusted", source) 82 self.assertNotIn(str(prefix), source) 83 self.assertIn("--state-leaf", source) 84 self.assertEqual(self.capture.with_suffix('.data').read_bytes(), prefix.read_bytes()) 85 86 def test_missing_marker_is_failure(self): 87 result = self.call("configure", "--code", PIN, "--entry", "configurePrefix", 88 "--input-leaf", "--state-leaf", SCP_TEST_FAIL="1") 89 self.assertEqual(result.returncode, 1) 90 self.assertIn("configuration-error", result.stderr) 91 92 def test_marker_without_pin_is_failure(self): 93 result = self.call("file", "--invocations", PIN, "--entry", "prefixLines", 94 "--label", "matches", SCP_TEST_NO_FIELDS="1") 95 self.assertEqual(result.returncode, 1) 96 self.assertIn("invalid file-manifest output", result.stderr) 97 98 def test_duplicate_pin_and_exclusive_inputs(self): 99 result = self.call("build", "--directory", str(REPO), "--workspace", 100 "demos.consumer-prefix-workspace", "--compiler", PIN, 101 SCP_TEST_DUPLICATE="1") 102 self.assertEqual(result.returncode, 1) 103 self.assertIn("duplicate or invalid code-manifest", result.stderr) 104 result = self.call("configure", "--code", PIN, "--entry", "configurePrefix", 105 "--input-leaf", "--input-tree-pin", PIN, "--state-leaf") 106 self.assertNotEqual(result.returncode, 0) 107 108 def test_configuration_tree_transports(self): 109 payload = self.root / "argument.tree" 110 payload.write_bytes(bytes([0])) 111 result = self.call("configure", "--code", PIN, "--entry", "configure", 112 "--input-tree-file", str(payload), "--state-tree-pin", PIN) 113 self.assertEqual(result.returncode, 0, result.stderr) 114 source = self.capture.read_text() 115 self.assertIn("--input-tree-file", source) 116 self.assertIn("--state-tree-pin", source) 117 118 def test_entry_binding(self): 119 result = self.call("configure-entry", "--code", PIN, "--entry", "fizzbuzz", 120 "--name", "chosen", "--state-tree-pin", PIN) 121 self.assertEqual(result.returncode, 0, result.stderr) 122 self.assertIn("configure-entry", self.capture.read_text()) 123 self.assertIn('"chosen"', self.capture.read_text()) 124 result = self.call("configure-entry", "--code", PIN, "--entry", "fizzbuzz", "--name", "chosen") 125 self.assertNotEqual(result.returncode, 0) 126 127 def test_direct_tree_display_option(self): 128 result = self.call("run-tree", "--manifest", PIN, "--entry", "fizzbuzz", 129 "--input-tree-pin", PIN, "--format", "tree") 130 self.assertEqual(result.returncode, 0, result.stderr) 131 arguments = json.loads(self.capture.with_suffix('.args').read_text()) 132 self.assertEqual(arguments[arguments.index("--format") + 1], "tree") 133 134 def test_deadline(self): 135 result = self.call("file", "--invocations", PIN, "--entry", "prefixLines", 136 "--label", "matches", SCP_TEST_SLEEP="1") 137 self.assertEqual(result.returncode, 124) 138 self.assertIn("deadline 1s", result.stderr) 139 140 def test_run_never_generates_source(self): 141 result = self.call("run-file", "--manifest", PIN, "--input-file", "absent input") 142 self.assertEqual(result.returncode, 0, result.stderr) 143 self.assertEqual(result.stdout, "pinned-run-ok\n") 144 self.assertFalse(self.capture.exists()) 145 146 class ProjectTests(unittest.TestCase): 147 setUp = CommandTests.setUp 148 tearDown = CommandTests.tearDown 149 150 def declaration(self): 151 self.project = self.root / "program.scp-project" 152 value = { 153 "version": "scp-project.v1", "name": "example", 154 "workspace": "demos.consumer-prefix-workspace", "directory": ".", 155 "configuration": {"entry": "configurePrefix", "input": {"text": "ERROR "}, "state": {"leaf": True}}, 156 "presentation": {"profile": "scp.file-lines.result.v1", "entry": "prefixLines", "label": "matches"}} 157 self.project.write_text(json.dumps(value)) 158 return value 159 160 def project_call(self, command, *options, **environment): 161 env = dict(os.environ, SCP_TEST_CAPTURE=str(self.capture), **environment) 162 return subprocess.run([str(TRICU), "scp", command, 163 "--seconds", "1", "--tricu", str(self.runner), *options], 164 cwd=self.root, env=env, text=True, capture_output=True) 165 166 def prepare(self, **environment): 167 return self.project_call("prepare", str(self.project), "--bootstrap", str(self.bootstrap), "--store", str(self.root / "cas"), 168 "--output", str(self.root / "prepared"), **environment) 169 170 def test_project_prepare_and_source_free_run(self): 171 self.declaration() 172 result = self.prepare() 173 self.assertEqual(result.returncode, 0, result.stderr) 174 prepared = json.loads((self.root / "prepared").read_text()) 175 self.assertEqual(prepared["store"], "cas") 176 self.assertEqual(prepared["file-manifest"], PIN) 177 self.assertRegex(prepared["compiler"], r"^[0-9a-f]{64}$") 178 self.project.unlink() 179 self.capture.unlink() 180 result = self.project_call("run", str(self.root / "prepared"), "--input-file", "missing input") 181 self.assertEqual(result.returncode, 0, result.stderr) 182 self.assertEqual(result.stdout, "pinned-run-ok\n") 183 self.assertFalse(self.capture.exists()) 184 185 def test_direct_profile_and_wrong_transport(self): 186 value = self.declaration() 187 value["presentation"] = {"profile": "scp-invocation.result.v1", "entry": "fizzbuzz"} 188 payload = self.root / "phases.tree" 189 payload.write_bytes(bytes([0])) 190 value["configuration"] = {"bind": "fizzbuzz", "state": {"tree-file": "phases.tree"}} 191 self.project.write_text(json.dumps(value)) 192 result = self.prepare() 193 self.assertEqual(result.returncode, 0, result.stderr) 194 prepared = json.loads((self.root / "prepared").read_text()) 195 self.assertNotIn("file-manifest", prepared) 196 self.assertEqual(prepared["entry"], "fizzbuzz") 197 wrong = self.project_call("run", str(self.root / "prepared"), "--input-file", "raw bytes") 198 self.assertEqual(wrong.returncode, 1) 199 self.assertIn("direct invocation requires", wrong.stderr) 200 result = self.project_call("run", str(self.root / "prepared"), "--input-tree-pin", PIN) 201 self.assertEqual(result.returncode, 0, result.stderr) 202 self.assertEqual(result.stdout, "pinned-run-ok\n") 203 self.assertIn("--result", json.loads(self.capture.with_suffix('.args').read_text())) 204 result = self.project_call("run", str(self.root / "prepared"), "--input-tree-pin", PIN, 205 "--format", "tree") 206 self.assertEqual(result.returncode, 0, result.stderr) 207 arguments = json.loads(self.capture.with_suffix('.args').read_text()) 208 self.assertEqual(arguments[arguments.index("--format") + 1], "tree") 209 210 def test_failed_reprepare_preserves_previous_record(self): 211 self.declaration() 212 self.assertEqual(self.prepare().returncode, 0) 213 old = (self.root / "prepared").read_bytes() 214 result = self.prepare(SCP_TEST_FAIL="1", SCP_TEST_FAIL_PHASE="configure") 215 self.assertEqual(result.returncode, 1) 216 self.assertIn("example: preparation stopped at configure", result.stderr) 217 self.assertEqual((self.root / "prepared").read_bytes(), old) 218 219 def test_reject_duplicate_unknown_and_missing_fields(self): 220 value = self.declaration() 221 for invalid in (dict(value, surprise=True), {key: val for key, val in value.items() if key != "configuration"}): 222 self.project.write_text(json.dumps(invalid)) 223 result = self.prepare() 224 self.assertEqual(result.returncode, 1) 225 self.assertNotIn("Traceback", result.stderr) 226 self.assertFalse(self.capture.exists()) 227 self.project.write_text('{"version":"scp-project.v1","version":"scp-project.v1"}') 228 self.assertIn("duplicate declaration key", self.prepare().stderr) 229 230 def test_nested_duplicate_and_unicode_snapshot(self): 231 value = self.declaration() 232 value["configuration"]["input"] = {"text": 'préfixe "\\\"\n\t'} 233 self.project.write_text(json.dumps(value)) 234 result = self.prepare() 235 self.assertEqual(result.returncode, 0, result.stderr) 236 self.assertEqual(self.capture.with_suffix('.data').read_bytes(), 237 value["configuration"]["input"]["text"].encode("utf-8")) 238 declaration = json.dumps(value).replace('"leaf": true', '"leaf": true, "leaf": true') 239 self.project.write_text(declaration) 240 self.assertIn("duplicate declaration key", self.prepare().stderr) 241 242 def test_sibling_store_is_relative(self): 243 self.declaration() 244 output = self.root / "records/program.prepared" 245 result = self.project_call("prepare", str(self.project), "--bootstrap", str(self.bootstrap), 246 "--store", str(self.root / "cas"), "--output", str(output)) 247 self.assertEqual(result.returncode, 0, result.stderr) 248 self.assertEqual(json.loads(output.read_text())["store"], "../cas") 249 result = self.project_call("run", str(output), "--input-file", "input") 250 self.assertEqual(result.returncode, 0, result.stderr) 251 arguments = json.loads(self.capture.with_suffix('.args').read_text()) 252 self.assertEqual(arguments[arguments.index("--store") + 1], str(self.root / "cas")) 253 254 def test_missing_data_fails_before_build(self): 255 value = self.declaration() 256 value["configuration"]["input"] = {"file": "missing"} 257 self.project.write_text(json.dumps(value)) 258 self.assertEqual(self.prepare().returncode, 1) 259 self.assertFalse(self.capture.exists()) 260 261 def test_invalid_prepared_pin_is_friendly_error(self): 262 self.declaration() 263 self.assertEqual(self.prepare().returncode, 0) 264 value = json.loads((self.root / "prepared").read_text()) 265 value["file-manifest"] = {"not": "a pin"} 266 (self.root / "prepared").write_text(json.dumps(value)) 267 result = self.project_call("run", str(self.root / "prepared"), "--input-file", "input") 268 self.assertEqual(result.returncode, 1) 269 self.assertNotIn("Traceback", result.stderr) 270 271 272 if __name__ == "__main__": 273 unittest.main(argv=[sys.argv[0]])