91 lines
4.0 KiB
Python
91 lines
4.0 KiB
Python
import json, os, sys, tempfile, unittest
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', 'user', 'remote_access', 'pager-webui'))
|
|
import mk8_events
|
|
|
|
class EventsTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.dir = tempfile.mkdtemp()
|
|
self.old = mk8_events.MK8_DIR
|
|
mk8_events.MK8_DIR = self.dir
|
|
mk8_events.EVENTS_PATH = os.path.join(self.dir, 'events.log')
|
|
|
|
def tearDown(self):
|
|
mk8_events.MK8_DIR = self.old
|
|
mk8_events.EVENTS_PATH = os.path.join(self.old, 'events.log')
|
|
mk8_events.MAX_BYTES = 5 * 1024 * 1024
|
|
|
|
def test_log_and_read_newest_first(self):
|
|
mk8_events.log_event('boot', msg='first')
|
|
mk8_events.log_event('rollback', sev='warn', msg='second', meta={'op': 'wifi'})
|
|
rows = mk8_events.read_events()
|
|
self.assertEqual(rows[0]['kind'], 'rollback')
|
|
self.assertEqual(rows[1]['kind'], 'boot')
|
|
self.assertEqual(rows[0]['meta'], {'op': 'wifi'})
|
|
|
|
def test_counters(self):
|
|
mk8_events.log_event('boot'); mk8_events.log_event('rollback')
|
|
mk8_events.log_event('restart'); mk8_events.log_event('guard_fix')
|
|
c = mk8_events.counters()
|
|
self.assertEqual(c['boots'], 1)
|
|
self.assertEqual(c['rollbacks'], 1)
|
|
self.assertEqual(c['restarts'], 1)
|
|
self.assertEqual(c['guard_fixes'], 1)
|
|
|
|
def test_snapshot_bounded_and_counts(self):
|
|
# Deterministic window: 3 restarts (oldest, outside the scan window),
|
|
# then 40 ticks, then 1 boot LAST so the newest-first window of 30
|
|
# includes it.
|
|
for _ in range(3):
|
|
mk8_events.log_event('restart', msg='old')
|
|
for i in range(40):
|
|
mk8_events.log_event('tick', msg='t%d' % i)
|
|
mk8_events.log_event('boot', msg='final boot')
|
|
snap = mk8_events.snapshot(event_limit=5, scan=30)
|
|
self.assertLessEqual(len(snap['events']), 5)
|
|
ev = snap['events']
|
|
tss = [r.get('ts', 0) for r in ev]
|
|
self.assertEqual(tss, sorted(tss, reverse=True), 'must be newest-first')
|
|
self.assertEqual(ev[0]['kind'], 'boot')
|
|
rel = snap['reliability']
|
|
self.assertEqual(rel['boots'], 1,
|
|
'boot is inside the scanned window')
|
|
self.assertEqual(rel['restarts'], 0,
|
|
'counters must cover only the bounded window')
|
|
self.assertEqual(rel['unexpected_boots'], 0)
|
|
# Full-journal counters() stays available and unbounded for callers
|
|
# that want it.
|
|
self.assertEqual(mk8_events.counters()['restarts'], 3)
|
|
|
|
def test_snapshot_spans_rotation(self):
|
|
mk8_events.MAX_BYTES = 200
|
|
for i in range(8):
|
|
mk8_events.log_event('tick', msg='x' * 20)
|
|
if i == 3:
|
|
mk8_events.log_event('rollback', sev='warn', msg='mid')
|
|
rotated = [i for i in range(1, mk8_events.KEEP + 1)
|
|
if os.path.exists(mk8_events.EVENTS_PATH + '.%d' % i)]
|
|
self.assertGreaterEqual(len(rotated) + 1, 2,
|
|
'journal must actually rotate here')
|
|
snap = mk8_events.snapshot(event_limit=100, scan=1000)
|
|
# Snapshot must span every rotated generation read_events sees.
|
|
self.assertEqual(len(snap['events']),
|
|
len(mk8_events.read_events(limit=1000)))
|
|
tss = [r.get('ts', 0) for r in snap['events']]
|
|
self.assertEqual(tss, sorted(tss, reverse=True))
|
|
kinds = {r['kind'] for r in snap['events']}
|
|
self.assertEqual(kinds, {'tick', 'rollback'})
|
|
self.assertEqual(snap['reliability']['rollbacks'], 1)
|
|
|
|
def test_rotation_keeps_recent(self):
|
|
mk8_events.MAX_BYTES = 200
|
|
for i in range(20):
|
|
mk8_events.log_event('tick', msg='x' * 30)
|
|
rows = mk8_events.read_events()
|
|
self.assertGreater(len(rows), 0)
|
|
self.assertLessEqual(len(rows), 20)
|
|
self.assertTrue(os.path.exists(mk8_events.EVENTS_PATH + '.1'))
|
|
self.assertFalse(os.path.exists(mk8_events.EVENTS_PATH + '.5'))
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|