46 lines
1.8 KiB
Python
46 lines
1.8 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_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()
|