tegwick
b6d0f78f93
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e6f1-443f-7783-9920-a16b2ffc467f
57 lines
2.9 KiB
Python
57 lines
2.9 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / 'scripts'))
|
|
from prometheus_export import labels, render
|
|
|
|
|
|
class ExportTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.now = datetime(2026, 9, 28, tzinfo=timezone.utc)
|
|
self.contract = json.loads((ROOT / 'contracts/platform-assurance-2026-09-28.json').read_text())
|
|
self.report = dict(schema='railiance-platform.assurance-signal.v1',
|
|
cluster_uid='a553c742-0115-43d4-99a4-a5ca56fe0786',
|
|
evaluated_at=self.now.isoformat(),
|
|
signals={s: dict(state='healthy', owner='railiance-platform') for s in self.contract['signals']},
|
|
transport='unmonitored', guarantees='unsupported',
|
|
threshold_status='local-diagnostic-only', healthy=True)
|
|
|
|
def test_all_states_preserved_including_token_renewal(self):
|
|
for state in ('healthy', 'failed', 'missing', 'unavailable', 'stale'):
|
|
self.report['signals']['eso.token-renewal']['state'] = state
|
|
self.report['healthy'] = state == 'healthy'
|
|
output = render(self.report, self.contract, self.now)
|
|
samples = [line for line in output.splitlines() if line.startswith('railiance_assurance_state')]
|
|
self.assertEqual(len(samples), 120)
|
|
self.assertEqual(sum(line.endswith(' 1') for line in samples), 24)
|
|
self.assertIn(f'signal="eso.token-renewal",state="{state}"}} 1', output)
|
|
|
|
def test_retry_preserves_time_and_does_not_mutate_report(self):
|
|
before = copy.deepcopy(self.report)
|
|
self.assertEqual(render(self.report, self.contract, self.now),
|
|
render(self.report, self.contract, self.now + timedelta(seconds=900)))
|
|
self.assertEqual(self.report, before)
|
|
for delta in (-1, 901):
|
|
with self.assertRaises(ValueError):
|
|
render(self.report, self.contract, self.now + timedelta(seconds=delta))
|
|
|
|
def test_scope_drift_is_rejected(self):
|
|
self.report['signals']['unexpected'] = dict(state='healthy', owner='railiance-platform')
|
|
with self.assertRaises(ValueError): render(self.report, self.contract, self.now)
|
|
|
|
def test_label_escaping(self):
|
|
self.assertEqual(labels({'signal': 'a"b\\c\nd'}), '{signal="a\\"b\\\\c\\nd"}')
|
|
|
|
def test_cli_failure_has_no_metrics_stdout(self):
|
|
result = subprocess.run([sys.executable, str(ROOT / 'scripts/prometheus_export.py'),
|
|
'--contract', str(ROOT / 'contracts/platform-assurance-2026-09-28.json'),
|
|
str(ROOT / 'tests/nonexistent-report.json')], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 2)
|
|
self.assertEqual(result.stdout, '')
|
|
self.assertIn('invalid-or-expired-report', result.stderr)
|