"""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()