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

    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')
        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')
        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')
        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')
        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')
        # 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')
        # 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
        timing.set_interval(1.2)
        self.timing_start()
        timing.add_task(1, 1, self.func, 4, 'generate_id')
        time.sleep(3)
        timing.stop()
        # self.assertAlmostEqual(exec_num, 3, delta=1)
        self.assertEqual(exec_num, 3)
        time.sleep(1)
        self.reset()

        # add task with count less than 0, success
        print('-----------------------------------------')
        timing.set_interval(1.2)
        self.timing_start()
        timing.add_task(1, 1, self.func, -1, 'generate_id')
        time.sleep(3)
        timing.stop()
        self.assertEqual(exec_num, 3)
        self.reset()


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