142 lines
6.2 KiB
Python
142 lines
6.2 KiB
Python
import os
|
|
import subprocess
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
from bcachefs_mqtt import Bridge, OPTIONS
|
|
from kent_report import Report
|
|
|
|
UUID = '3d3a0763-4dfe-41e6-93c1-8c791ec98176'
|
|
|
|
|
|
class Client:
|
|
def __init__(self):
|
|
self.messages = []
|
|
self.subscriptions = []
|
|
|
|
def publish(self, topic, payload, **kwargs):
|
|
self.messages.append((topic, payload))
|
|
|
|
def subscribe(self, topic, **kwargs):
|
|
self.subscriptions.append(topic)
|
|
|
|
def will_set(self, topic, payload, **kwargs):
|
|
self.will = (topic, payload, kwargs)
|
|
|
|
|
|
class BridgeTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.options = self.root / UUID / 'options'
|
|
self.options.mkdir(parents=True)
|
|
for option in OPTIONS:
|
|
(self.options / option).write_text('1\n')
|
|
self.client = Client()
|
|
self.bridge = Bridge(self.client, UUID, self.root)
|
|
|
|
def command(self, payload, retain=False, option='reconcile_enabled'):
|
|
self.bridge.on_message(self.client, None, SimpleNamespace(
|
|
topic=f'{self.bridge.base}/{option}/set', payload=payload, retain=retain))
|
|
|
|
def test_connect_discovery_and_state_before_online(self):
|
|
self.bridge.on_connect(self.client, None, None, SimpleNamespace(is_failure=False), None)
|
|
configs = [json.loads(value) for topic, value in self.client.messages if '/switch/' in topic and topic.endswith('/config')]
|
|
self.assertEqual(len(configs), 2)
|
|
self.assertFalse(configs[0]['optimistic'])
|
|
self.assertFalse(configs[0]['retain'])
|
|
self.assertEqual(configs[0]['availability_mode'], 'all')
|
|
self.assertEqual(self.client.messages[-1], (f'{self.bridge.base}/availability', 'online'))
|
|
self.assertEqual(self.client.will[1], 'offline')
|
|
self.assertTrue(self.client.will[2]['retain'])
|
|
|
|
def test_cli_changes_and_unmount_remount(self):
|
|
self.bridge.sample()
|
|
path = self.options / 'reconcile_enabled'
|
|
path.write_text('0\n')
|
|
self.bridge.sample()
|
|
self.assertIn((f'{self.bridge.base}/reconcile_enabled/state', '0'), self.client.messages)
|
|
path.unlink()
|
|
self.bridge.sample()
|
|
self.assertEqual(self.client.messages[-1], (f'{self.bridge.base}/reconcile_enabled/availability', 'offline'))
|
|
self.command(b'1')
|
|
self.assertFalse(path.exists())
|
|
path.write_text('1\n')
|
|
self.bridge.sample()
|
|
self.assertEqual(self.client.messages[-1], (f'{self.bridge.base}/reconcile_enabled/availability', 'online'))
|
|
|
|
def test_commands_and_rejected_retained_payload(self):
|
|
self.command(b'0')
|
|
self.assertEqual((self.options / 'reconcile_enabled').read_text(), '0\n')
|
|
self.command(b'1', retain=True)
|
|
self.command(b'ON')
|
|
self.command(b'1', option='unknown')
|
|
self.assertEqual((self.options / 'reconcile_enabled').read_text(), '0\n')
|
|
self.command(b'1')
|
|
self.assertEqual((self.options / 'reconcile_enabled').read_text(), '1\n')
|
|
|
|
|
|
class ReportTest(unittest.TestCase):
|
|
def test_outputs_failures_timeouts_and_html_escaping(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
report = Report(directory, timeout=.1)
|
|
report.run('dmesg', [sys.executable, '-c', 'import sys; print("<script>oops</script>"); print("stderr",file=sys.stderr); sys.exit(3)'])
|
|
report.run('missing', ['/no/such/command'])
|
|
report.run('timeout', [sys.executable, '-c', 'import time; time.sleep(10)'])
|
|
report.render()
|
|
root = Path(directory)
|
|
self.assertEqual([r['status'] for r in report.records], [3, 127, 124])
|
|
self.assertIn('stderr', (root / 'outputs/001-dmesg.txt').read_text())
|
|
page = (root / 'index.html').read_text()
|
|
self.assertIn('<script>', page)
|
|
self.assertNotIn('<script>oops</script>', page)
|
|
self.assertIn('output.textContent = await response.text()', page)
|
|
self.assertIn('outputs/001-dmesg.txt', page)
|
|
self.assertIn('stderr', (root / 'kent_report.log').read_text())
|
|
self.assertIn('001-dmesg.txt', (root / 'outputs/index.html').read_text())
|
|
self.assertEqual(len(json.loads((root / 'manifest.json').read_text())['commands']), 3)
|
|
|
|
|
|
class UploadTest(unittest.TestCase):
|
|
def test_directory_upload_artifacts_and_failures(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
report = Report(root / 'report')
|
|
report.run('example', ['printf', 'example'])
|
|
report.render()
|
|
(report.directory / 'trace.dat').write_text('must remain local')
|
|
stub = root / 'kubectl'
|
|
stub.write_text('''#!/bin/bash
|
|
case $1 in
|
|
get) echo brunner-ninja-test ;;
|
|
cp)
|
|
[[ -f $2/index.html && -f $2/outputs/001-example.txt && ! -e $2/trace.dat ]] || exit 9
|
|
exit "${MOCK_CP_STATUS:-0}" ;;
|
|
exec) exit "${MOCK_RENAME_STATUS:-0}" ;;
|
|
esac
|
|
''')
|
|
stub.chmod(0o755)
|
|
env = dict(os.environ, PATH=str(root) + ':' + os.environ['PATH'], TMPDIR=str(root))
|
|
command = ['bash', '-c', 'source "$1"; paste_directory "$2"', 'bash', str(ROOT / 'paste_log.sh'), str(report.directory)]
|
|
result = subprocess.run(command, env=env, capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertTrue(result.stdout.startswith('https://brunner.ninja/pastes/kent-report-'))
|
|
result = subprocess.run(command, env=env | {'MOCK_CP_STATUS': '7'}, capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 7)
|
|
self.assertEqual(result.stdout, '')
|
|
result = subprocess.run(command, env=env | {'MOCK_RENAME_STATUS': '8'}, capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 8)
|
|
self.assertEqual(result.stdout, '')
|
|
self.assertEqual(list(root.glob('kent-report-*')), [])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|