83 lines
4.5 KiB
Python
83 lines
4.5 KiB
Python
"""Contract checks; no Docker required. Does not replace the Debian smoke test."""
|
|
import json
|
|
from pathlib import Path
|
|
import tomllib
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
class Metric:
|
|
def __init__(self, measurement, fields, seconds, **tags):
|
|
self.name, self.fields, self.time = measurement, fields, int(seconds * 1e9)
|
|
self.tags = dict(host='node-a', **tags)
|
|
|
|
class Contracts(unittest.TestCase):
|
|
def setUp(self):
|
|
# enrich.star deliberately uses Python-compatible Starlark constructs.
|
|
# This exercises rate mathematics; it is not a Starlark-runtime validation.
|
|
self.scope = {}
|
|
exec((ROOT / 'agent/enrich.star').read_text(), self.scope)
|
|
self.apply = self.scope['apply']
|
|
|
|
def test_rates_isolate_devices_and_handle_resets_gaps(self):
|
|
self.apply(Metric('net', {'bytes_recv': 100}, 10, interface='eth0'))
|
|
self.apply(Metric('net', {'bytes_recv': 8000}, 10, interface='eth1'))
|
|
got = self.apply(Metric('net', {'bytes_recv': 400}, 25, interface='eth0'))
|
|
self.assertEqual(got.fields['bytes_recv_per_sec'], 20)
|
|
self.assertEqual(got.tags['series'], 'node-a / eth0')
|
|
reset = self.apply(Metric('net', {'bytes_recv': 1}, 40, interface='eth0'))
|
|
self.assertNotIn('bytes_recv_per_sec', reset.fields)
|
|
same = self.apply(Metric('net', {'bytes_recv': 100}, 40, interface='eth0'))
|
|
self.assertNotIn('bytes_recv_per_sec', same.fields)
|
|
recover = self.apply(Metric('net', {'bytes_recv': 31}, 55, interface='eth0'))
|
|
self.assertEqual(recover.fields['bytes_recv_per_sec'], 2)
|
|
gap = self.apply(Metric('net', {'bytes_recv': 99999}, 500, interface='eth0'))
|
|
self.assertNotIn('bytes_recv_per_sec', gap.fields)
|
|
|
|
def test_busy_percent_and_normalized_load(self):
|
|
self.apply(Metric('diskio', {'io_time': 100}, 1, name='sda'))
|
|
m = self.apply(Metric('diskio', {'io_time': 7600}, 16, name='sda'))
|
|
self.assertEqual(m.fields['io_busy_percent'], 50)
|
|
m = self.apply(Metric('system', {'load1': 4, 'n_cpus': 8}, 16))
|
|
self.assertEqual(m.fields['load1_per_cpu'], 0.5)
|
|
|
|
def test_saved_object_graph_fields_and_layouts(self):
|
|
objects = [json.loads(x) for x in (ROOT / 'dashboards/host-metrics.ndjson').read_text().splitlines()]
|
|
self.assertEqual(objects, json.loads((ROOT / 'dashboards/host-metrics.json').read_text()))
|
|
ids = {(o['type'], o['id']) for o in objects}
|
|
self.assertEqual(len(ids), len(objects))
|
|
fields = {f['name'] for f in json.loads(objects[0]['attributes']['fields'])}
|
|
for o in objects:
|
|
for r in o['references']:
|
|
self.assertIn((r['type'], r['id']), ids)
|
|
a = o['attributes']
|
|
if o['type'] == 'visualization':
|
|
for agg in json.loads(a['visState'])['aggs']:
|
|
if 'field' in agg['params']:
|
|
self.assertIn(agg['params']['field'], fields)
|
|
if o['type'] == 'dashboard':
|
|
panels = json.loads(a['panelsJSON'])
|
|
for i, panel in enumerate(panels):
|
|
p = panel['gridData']
|
|
self.assertLessEqual(p['x'] + p['w'], 48)
|
|
for other in panels[i + 1:]:
|
|
q = other['gridData']
|
|
overlaps = p['x'] < q['x'] + q['w'] and q['x'] < p['x'] + p['w'] and p['y'] < q['y'] + q['h'] and q['y'] < p['y'] + p['h']
|
|
self.assertFalse(overlaps)
|
|
self.assertEqual(sum(o['type'] == 'dashboard' for o in objects), 4)
|
|
|
|
def test_configs_and_index_contract(self):
|
|
agent = tomllib.loads((ROOT / 'agent/telegraf.conf').read_text())
|
|
bridge = tomllib.loads((ROOT / 'configs/metrics-bridge/telegraf.conf').read_text())
|
|
self.assertEqual(agent['outputs']['kafka'][0]['topic'], bridge['inputs']['kafka_consumer'][0]['topics'][0])
|
|
self.assertEqual(agent['outputs']['kafka'][0]['json_timestamp_units'], '1ms')
|
|
self.assertEqual(bridge['inputs']['kafka_consumer'][0]['json_v2'][0]['timestamp_format'], 'unix_ms')
|
|
self.assertEqual(set(bridge['inputs']), {'kafka_consumer'})
|
|
template = json.loads((ROOT / 'configs/opensearch/index-template.json').read_text())
|
|
props = template['template']['mappings']['properties']
|
|
self.assertEqual(props['@timestamp']['type'], 'date')
|
|
self.assertEqual(props['fields']['properties']['used_percent']['type'], 'double')
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|