elec-monitor/upload_test.py

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


import unittest
import upload
import threading
import time


class UploadTest(unittest.TestCase):
    def generate_content(self):
        return {
            'dev_ip': 'local.pc',
            'port': ':9011',
            'type': 'camera',
            'schema': 'onvif',
            'kpi': {
                'img_light': 100,
                'img_cast': 100,
                'img_rgcast': 100,
                'img_lightcast': 100,
                'img_clear': 100,
                'img_ybcast': 100,
            }
        }

    def test_generate_file(self):
        print("-=-=-=-=")
        upload.defaultHandler = upload.Handle(1, 'temp.txt', './')

        t = threading.Thread(target=upload.listen)
        t.setDaemon(True)
        t.start()

        # add task
        upload.add_task(self.generate_content())
        time.sleep(4)

        upload.add_task(self.generate_content())
        upload.add_task(self.generate_content())
        upload.add_task(self.generate_content())
        upload.add_task(self.generate_content())
        upload.add_task(self.generate_content())
        upload.add_task(self.generate_content())
        time.sleep(4)


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