upload_test.py 2.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990
  1. #!/usr/bin/env python
  2. # -*- coding:utf-8 -*-
  3. import unittest
  4. import upload
  5. import threading
  6. import time
  7. class UploadFileCheckTest(unittest.TestCase):
  8. def generate_content(self):
  9. return {
  10. 'dev_ip': 'local.pc',
  11. 'port': ':9011',
  12. 'type': 'camera',
  13. 'schema': 'onvif',
  14. 'kpi': {
  15. 'img_light': 100,
  16. 'img_cast': 100,
  17. 'img_rgcast': 100,
  18. 'img_lightcast': 100,
  19. 'img_clear': 100,
  20. 'img_ybcast': 100,
  21. }
  22. }
  23. def test_generate_file(self):
  24. host = 'local.pc'
  25. port = 22
  26. user = 'tangs'
  27. passwd = 'local12456'
  28. official = 'test/official'
  29. print("-=-=-=-=")
  30. upload.defaultHandler = upload.Handle(1, 'temp.txt', official, 'storage/rubbish', host, port, user, passwd,
  31. 'test/scp_generate')
  32. t = threading.Thread(target=upload.listen)
  33. t.setDaemon(True)
  34. t.start()
  35. # add task
  36. upload.add_task(self.generate_content())
  37. time.sleep(4)
  38. upload.add_task(self.generate_content())
  39. upload.add_task(self.generate_content())
  40. upload.add_task(self.generate_content())
  41. upload.add_task(self.generate_content())
  42. upload.add_task(self.generate_content())
  43. upload.add_task(self.generate_content())
  44. time.sleep(4)
  45. class UploaderTest(unittest.TestCase):
  46. """
  47. Test uploading single file, single folder or multiple folder to remote server via SCP.
  48. """
  49. def test_upload(self):
  50. host = 'local.pc'
  51. port = 22
  52. username = 'tangs'
  53. password = 'local12456'
  54. # upload file
  55. uploader1 = upload.Uploader(host, port, username, password, 'test/scp_file_1.txt', 'test/scp_file_1.txt')
  56. e = uploader1.upload()
  57. self.assertIsNone(e)
  58. # upload file
  59. uploader2 = upload.Uploader(host, port, username, password, 'test/scp_file_2.txt', 'test/scp_file_2.txt')
  60. e = uploader2.upload()
  61. self.assertIsNone(e)
  62. # upload folder
  63. uploader3 = upload.Uploader(host, port, username, password, 'test/scp_folder', 'test/')
  64. e = uploader3.upload()
  65. self.assertIsNone(e)
  66. # upload a list of files
  67. uploader4 = upload.Uploader(host, port, username, password, ['test/scp_list', 'test/scp_folder'],
  68. 'test/scp_list')
  69. e = uploader4.upload()
  70. self.assertIsNone(e)
  71. if __name__ == '__main__':
  72. unittest.main()