elec-monitor/timing_test.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import unittest
import timing
import threading
import time

exec_num = 0


class TimingTest(unittest.TestCase):
    def func(self):
        global exec_num
        self.name = 'TANGS TIMING TEST'
        exec_num += 1
        print(self.name)

    @staticmethod
    def reset():
        global exec_num
        exec_num = 0

    @staticmethod
    def timing_start():
        t = threading.Thread(target=timing.run)
        t.setDaemon(True)
        t.start()
        print('New Thread', t.name)
        return t

    def test_timing_run(self):
        # add task but invalid task_id
        err = None
        try:
            timing.add_task(1, 1, self.func, 1, 1)
        except Exception as e:
            err = str(e)
        self.assertEqual(err, 'task_id must be str')

        # add task but invalid interval
        err = None
        try:
            timing.add_task(1, '1', self.func, 1, 'generate_id_1')
        except Exception as e:
            err = str(e)
        self.assertEqual(err, 'interval must be int')

        # add task but interval must be bigger than 0
        err = None
        try:
            timing.add_task(1, -1, self.func, 1, 'generate_id_2')
        except Exception as e:
            err = str(e)
        self.assertEqual(err, 'interval must be bigger than 0')

        # add task but func is not function
        err = None
        try:
            timing.add_task(1, 1, 1, 1, 'generate_id_3')
        except Exception as e:
            err = str(e)
        self.assertEqual(err, 'func must be func')

        # add task but count is not int
        err = None
        try:
            timing.add_task(1, 1, self.func, '1', 'generate_id_4')
        except Exception as e:
            err = str(e)
        self.assertEqual(err, 'count must be int')

        timing.set_interval(1)
        timer = timing.Timing()
        timing.defaultTiming = timer

        # add task success
        self.timing_start()
        timing.add_task(1, 1, self.func, 1, 'generate_id_5')
        self.assertEqual(len(timer.task), 1)
        time.sleep(1.5)
        timing.stop()
        self.assertEqual(exec_num, 1)
        self.assertEqual(len(timer.task), 0)
        time.sleep(1)
        self.reset()

        # add task with count 3, wait 3.5, success
        self.timing_start()
        timing.add_task(1, 1, self.func, 3, 'generate_id_6')
        self.assertEqual(len(timer.task), 1)
        time.sleep(3.5)
        timing.stop()
        self.assertEqual(exec_num, 3)
        self.assertEqual(len(timer.task), 0)
        time.sleep(1)
        self.reset()

        # add task with count 4, wait 3.5, success
        t = self.timing_start()
        timing.add_task('name_4', 1, self.func, 4, 'generate_id_7')
        time.sleep(3.5)
        timing.stop()
        # self.assertAlmostEqual(exec_num, 3, delta=1)
        time.sleep(5)
        self.assertEqual(exec_num, 3)
        self.assertEqual(len(timer.task), 1)
        self.assertEqual(t.is_alive(), False)
        timer.task = {}
        self.reset()

        # add task with count less than 0, success
        print('-----------------------------------------')
        t = self.timing_start()
        timing.add_task('name', 1, self.func, -1, 'generate_id_8')
        time.sleep(3.5)
        timing.stop()
        time.sleep(5)
        self.assertEqual(exec_num, 3)
        self.assertEqual(len(timer.task), 1)
        self.assertEqual(t.is_alive(), False)
        timer.task = {}
        self.reset()

        # add multi task, success
        self.timing_start()
        timing.add_task('name_9', 1, self.func, 0, 'generate_id_9')
        timing.add_task('name_10', 1, self.func, 0, 'generate_id_10')
        timing.add_task('name_11', 1, self.func, 0, 'generate_id_11')
        timing.add_task('name_12', 1, self.func, 0, 'generate_id_12')
        timing.add_task('name_13', 1, self.func, 0, 'generate_id_13')
        timing.add_task('name_14', 1, self.func, 0, 'generate_id_14')
        timing.add_task('name_15', 1, self.func, 0, 'generate_id_15')
        timing.add_task('name_16', 1, self.func, 0, 'generate_id_16')
        timing.add_task('name_17', 1, self.func, 0, 'generate_id_17')
        timing.add_task('name_18', 1, self.func, 0, 'generate_id_18')
        timing.add_task('name_19', 1, self.func, 0, 'generate_id_19')
        timing.add_task('name_20', 1, self.func, 0, 'generate_id_20')
        time.sleep(10)
        self.assertAlmostEqual(exec_num, 100, delta=10)
        timing.stop()
        print(exec_num)


if __name__ == '__main__':
    unittest.main()