54 lines
3.1 KiB
Python
54 lines
3.1 KiB
Python
"""Value-free diagnostics and refusal boundaries; no cluster access."""
|
|
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
spec = importlib.util.spec_from_file_location('read_only_diagnostic', HERE / 'diagnose.py')
|
|
d = importlib.util.module_from_spec(spec); spec.loader.exec_module(d)
|
|
|
|
|
|
class Tests(unittest.TestCase):
|
|
def test_error_categories_never_echo_raw_diagnostics(self):
|
|
sentinel = b'PRIVATE_SENTINEL_DO_NOT_EMIT'
|
|
for raw, expected in [(b'Forbidden: ' + sentinel, 'forbidden'),
|
|
(b'command terminated with exit code 1 ' + sentinel, 'remote-check-nonzero'),
|
|
(sentinel, 'unclassified-command-failure')]:
|
|
self.assertEqual(d.failure_category(1, raw), expected)
|
|
|
|
def test_only_expected_reads_and_exact_check_scripts_are_labeled(self):
|
|
cfg = {'guard_command': ['sh', '-c', 'EXACT_PUBLIC_GUARD']}
|
|
self.assertEqual(d.step_label(('get', 'nodes', '-o', 'json'), cfg, 'INVENTORY'), 'node-identity-read')
|
|
self.assertEqual(d.step_label(('-n', 'forgejo', 'exec', '-i', 'pod', '--', 'sh', '-c', 'EXACT_PUBLIC_GUARD', 'sh', 'check'), cfg, 'INVENTORY'), 'installed-artifact-guard')
|
|
self.assertEqual(d.step_label(('-n', 'forgejo', 'exec', '-i', 'pod', '--', 'sh', '-c', 'INVENTORY', 'sh'), cfg, 'INVENTORY'), 'existing-repository-hook-inventory')
|
|
for argv in [('create', '-f', '-'), ('patch', 'deployment', 'forge'), ('get', 'secrets'),
|
|
('exec', 'pod', '--', 'sh', '-c', 'OTHER', 'sh'),
|
|
('exec', 'pod', '--', 'sh', '-c', 'EXACT_PUBLIC_GUARD', 'sh', 'apply')]:
|
|
with self.subTest(argv=argv), self.assertRaises(RuntimeError): d.step_label(argv, cfg, 'INVENTORY')
|
|
|
|
def test_failed_read_is_labeled_and_original_native_receipt_is_preserved(self):
|
|
native = Path('/tmp/forge-scanner-rollout-receipt.json')
|
|
before = native.read_bytes() if native.exists() else None
|
|
class Failed:
|
|
stdout = b'PRIVATE_STDOUT_SENTINEL'
|
|
stderr = b'Forbidden: PRIVATE_STDERR_SENTINEL'
|
|
returncode = 1
|
|
with tempfile.TemporaryDirectory() as temp, patch.object(d, 'OUTPUT', Path(temp) / 'receipt.json'), patch.object(d.subprocess, 'run', return_value=Failed()) as command:
|
|
captured = io.StringIO()
|
|
with contextlib.redirect_stdout(captured): result = d.main()
|
|
text = captured.getvalue(); receipt = json.loads(d.OUTPUT.read_bytes())
|
|
self.assertEqual(result, 1)
|
|
self.assertEqual(command.call_count, 1)
|
|
self.assertEqual(receipt['steps'], [{'step': 'node-identity-read', 'returncode': 1, 'category': 'forbidden'}])
|
|
self.assertFalse(receipt['native_mutation'])
|
|
self.assertNotIn('PRIVATE_', text)
|
|
self.assertNotIn('PRIVATE_', d.OUTPUT.read_text())
|
|
self.assertEqual(native.read_bytes() if native.exists() else None, before)
|
|
|
|
|
|
if __name__ == '__main__': unittest.main()
|