elec-monitor/upload_test.py

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


import unittest
import upload
import threading
import time


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


class UploaderTest(unittest.TestCase):
    def test_upload(self):
        host = 'local.pc'
        port = 22
        username = 'tangs'
        password = 'local12456'

        # upload file
        uploader1 = upload.Uploader(host, port, username, password, 'test/scp_file_1.txt', 'test/scp_file_1.txt')
        e = uploader1.upload()
        self.assertIsNone(e)

        # upload file
        uploader2 = upload.Uploader(host, port, username, password, 'test/scp_file_2.txt', 'test/scp_file_2.txt')
        e = uploader2.upload()
        self.assertIsNone(e)

        # upload folder
        uploader3 = upload.Uploader(host, port, username, password, 'test/scp_folder', 'test/scp_folder')
        e = uploader3.upload()
        self.assertIsNone(e)


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