A clone of: https://github.com/nutechsoftware/alarmdecoder This is requires as they dropped support for older firmware releases w/o building in backward compatibility code, and they had previously hardcoded pyserial to a python2 only version.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

316 lines
11 KiB

  1. from unittest import TestCase
  2. from mock import Mock, MagicMock, patch
  3. from serial import Serial, SerialException
  4. from pyftdi.pyftdi.ftdi import Ftdi, FtdiError
  5. from usb.core import USBError, Device as USBCoreDevice
  6. import socket
  7. from OpenSSL import SSL, crypto
  8. from ..devices import USBDevice, SerialDevice, SocketDevice
  9. from ..util import NoDeviceError, CommError, TimeoutError
  10. class TestUSBDevice(TestCase):
  11. def setUp(self):
  12. self._device = USBDevice()
  13. self._device._device = Mock(spec=Ftdi)
  14. self._device._device.usb_dev = Mock(spec=USBCoreDevice)
  15. self._device._device.usb_dev.bus = 0
  16. self._device._device.usb_dev.address = 0
  17. def tearDown(self):
  18. self._device.close()
  19. def test_find_all(self):
  20. with patch.object(USBDevice, 'find_all', return_value=[]) as mock:
  21. devices = USBDevice.find_all()
  22. self.assertEquals(devices, [])
  23. def test_find_all_exception(self):
  24. with patch.object(Ftdi, 'find_all', side_effect=[USBError('testing'), FtdiError]) as mock:
  25. with self.assertRaises(CommError):
  26. devices = USBDevice.find_all()
  27. with self.assertRaises(CommError):
  28. devices = USBDevice.find_all()
  29. def test_open(self):
  30. self._device.interface = ('AD2USB', 0)
  31. with patch.object(self._device._device, 'open') as mock:
  32. self._device.open(no_reader_thread=True)
  33. mock.assert_any_calls()
  34. def test_open_failed(self):
  35. self._device.interface = ('AD2USB', 0)
  36. with patch.object(self._device._device, 'open', side_effect=[USBError('testing'), FtdiError]):
  37. with self.assertRaises(NoDeviceError):
  38. self._device.open(no_reader_thread=True)
  39. with self.assertRaises(NoDeviceError):
  40. self._device.open(no_reader_thread=True)
  41. def test_write(self):
  42. self._device.interface = ('AD2USB', 0)
  43. self._device.open(no_reader_thread=True)
  44. with patch.object(self._device._device, 'write_data') as mock:
  45. self._device.write('test')
  46. mock.assert_called_with('test')
  47. def test_write_exception(self):
  48. with patch.object(self._device._device, 'write_data', side_effect=FtdiError):
  49. with self.assertRaises(CommError):
  50. self._device.write('test')
  51. def test_read(self):
  52. self._device.interface = ('AD2USB', 0)
  53. self._device.open(no_reader_thread=True)
  54. with patch.object(self._device._device, 'read_data') as mock:
  55. self._device.read()
  56. mock.assert_called_with(1)
  57. def test_read_exception(self):
  58. with patch.object(self._device._device, 'read_data', side_effect=[USBError('testing'), FtdiError]):
  59. with self.assertRaises(CommError):
  60. self._device.read()
  61. with self.assertRaises(CommError):
  62. self._device.read()
  63. def test_read_line(self):
  64. with patch.object(self._device._device, 'read_data', side_effect=list("testing\r\n")):
  65. ret = None
  66. try:
  67. ret = self._device.read_line()
  68. except StopIteration:
  69. pass
  70. self.assertEquals(ret, "testing")
  71. def test_read_line_timeout(self):
  72. with patch.object(self._device._device, 'read_data', return_value='a') as mock:
  73. with self.assertRaises(TimeoutError):
  74. self._device.read_line(timeout=0.1)
  75. self.assertIn('a', self._device._buffer)
  76. def test_read_line_exception(self):
  77. with patch.object(self._device._device, 'read_data', side_effect=[USBError('testing'), FtdiError]):
  78. with self.assertRaises(CommError):
  79. self._device.read_line()
  80. with self.assertRaises(CommError):
  81. self._device.read_line()
  82. class TestSerialDevice(TestCase):
  83. def setUp(self):
  84. self._device = SerialDevice()
  85. self._device._device = Mock(spec=Serial)
  86. self._device._device.open = Mock()
  87. def tearDown(self):
  88. self._device.close()
  89. def test_open(self):
  90. self._device.interface = '/dev/ttyS0'
  91. with patch.object(self._device._device, 'open') as mock:
  92. self._device.open(no_reader_thread=True)
  93. mock.assert_called_with()
  94. def test_open_no_interface(self):
  95. with self.assertRaises(NoDeviceError):
  96. self._device.open(no_reader_thread=True)
  97. self.assertFalse(self._device._running)
  98. def test_open_failed(self):
  99. self._device.interface = '/dev/ttyS0'
  100. with patch.object(self._device._device, 'open', side_effect=[SerialException, ValueError]):
  101. with self.assertRaises(NoDeviceError):
  102. self._device.open(no_reader_thread=True)
  103. with self.assertRaises(NoDeviceError):
  104. self._device.open(no_reader_thread=True)
  105. def test_write(self):
  106. self._device.interface = '/dev/ttyS0'
  107. self._device.open(no_reader_thread=True)
  108. with patch.object(self._device._device, 'write') as mock:
  109. self._device.write('test')
  110. mock.assert_called_with('test')
  111. def test_write_exception(self):
  112. with patch.object(self._device._device, 'write', side_effect=SerialException):
  113. with self.assertRaises(CommError):
  114. self._device.write('test')
  115. def test_read(self):
  116. self._device.interface = '/dev/ttyS0'
  117. self._device.open(no_reader_thread=True)
  118. with patch.object(self._device._device, 'read') as mock:
  119. self._device.read()
  120. mock.assert_called_with(1)
  121. def test_read_exception(self):
  122. with patch.object(self._device._device, 'read', side_effect=SerialException):
  123. with self.assertRaises(CommError):
  124. self._device.read()
  125. def test_read_line(self):
  126. with patch.object(self._device._device, 'read', side_effect=list("testing\r\n")):
  127. ret = None
  128. try:
  129. ret = self._device.read_line()
  130. except StopIteration:
  131. pass
  132. self.assertEquals(ret, "testing")
  133. def test_read_line_timeout(self):
  134. with patch.object(self._device._device, 'read', return_value='a') as mock:
  135. with self.assertRaises(TimeoutError):
  136. self._device.read_line(timeout=0.1)
  137. self.assertIn('a', self._device._buffer)
  138. def test_read_line_exception(self):
  139. with patch.object(self._device._device, 'read', side_effect=[OSError, SerialException]):
  140. with self.assertRaises(CommError):
  141. self._device.read_line()
  142. with self.assertRaises(CommError):
  143. self._device.read_line()
  144. class TestSocketDevice(TestCase):
  145. def setUp(self):
  146. self._device = SocketDevice()
  147. self._device._device = Mock(spec=socket.socket)
  148. def tearDown(self):
  149. self._device.close()
  150. def test_open(self):
  151. with patch.object(socket.socket, '__init__', return_value=None):
  152. with patch.object(socket.socket, 'connect', return_value=None) as mock:
  153. self._device.open(no_reader_thread=True)
  154. mock.assert_called_with(self._device.interface)
  155. def test_open_no_interface(self):
  156. with self.assertRaises(NoDeviceError):
  157. self._device.open(no_reader_thread=True)
  158. self.assertFalse(self._device._running)
  159. def test_open_failed(self):
  160. with patch.object(self._device._device, 'connect', side_effect=socket.error):
  161. with self.assertRaises(NoDeviceError):
  162. self._device.open(no_reader_thread=True)
  163. def test_write(self):
  164. with patch.object(socket.socket, '__init__', return_value=None):
  165. with patch.object(socket.socket, 'connect', return_value=None):
  166. self._device.open(no_reader_thread=True)
  167. with patch.object(socket.socket, 'send') as mock:
  168. self._device.write('test')
  169. mock.assert_called_with('test')
  170. def test_write_exception(self):
  171. with patch.object(self._device._device, 'send', side_effect=[SSL.Error, socket.error]):
  172. with self.assertRaises(CommError):
  173. self._device.write('test')
  174. def test_read(self):
  175. with patch.object(socket.socket, '__init__', return_value=None):
  176. with patch.object(socket.socket, 'connect', return_value=None):
  177. self._device.open(no_reader_thread=True)
  178. with patch.object(socket.socket, 'recv') as mock:
  179. self._device.read()
  180. mock.assert_called_with(1)
  181. def test_read_exception(self):
  182. with patch.object(self._device._device, 'recv', side_effect=socket.error):
  183. with self.assertRaises(CommError):
  184. self._device.read()
  185. def test_read_line(self):
  186. with patch.object(self._device._device, 'recv', side_effect=list("testing\r\n")):
  187. ret = None
  188. try:
  189. ret = self._device.read_line()
  190. except StopIteration:
  191. pass
  192. self.assertEquals(ret, "testing")
  193. def test_read_line_timeout(self):
  194. with patch.object(self._device._device, 'recv', return_value='a') as mock:
  195. with self.assertRaises(TimeoutError):
  196. self._device.read_line(timeout=0.1)
  197. self.assertIn('a', self._device._buffer)
  198. def test_read_line_exception(self):
  199. with patch.object(self._device._device, 'recv', side_effect=socket.error):
  200. with self.assertRaises(CommError):
  201. self._device.read_line()
  202. with self.assertRaises(CommError):
  203. self._device.read_line()
  204. def test_ssl(self):
  205. ssl_key = crypto.PKey()
  206. ssl_key.generate_key(crypto.TYPE_RSA, 2048)
  207. ssl_cert = crypto.X509()
  208. ssl_cert.set_pubkey(ssl_key)
  209. ssl_ca_key = crypto.PKey()
  210. ssl_ca_key.generate_key(crypto.TYPE_RSA, 2048)
  211. ssl_ca_cert = crypto.X509()
  212. ssl_ca_cert.set_pubkey(ssl_ca_key)
  213. self._device.ssl = True
  214. self._device.ssl_key = ssl_key
  215. self._device.ssl_certificate = ssl_cert
  216. self._device.ssl_ca = ssl_ca_cert
  217. # ..there has to be a better way..
  218. with patch.object(socket.socket, '__init__', return_value=None):
  219. with patch.object(socket.socket, 'connect', return_value=None) as mock:
  220. with patch.object(socket.socket, '_sock'):
  221. with patch.object(socket.socket, 'fileno', return_value=1):
  222. self._device.open(no_reader_thread=True)
  223. mock.assert_called_with(self._device.interface)
  224. self.assertIsInstance(self._device._device, SSL.Connection)
  225. def test_ssl_exception(self):
  226. self._device.ssl = True
  227. self._device.ssl_key = 'None'
  228. self._device.ssl_certificate = 'None'
  229. self._device.ssl_ca = 'None'
  230. # ..there has to be a better way..
  231. with patch.object(socket.socket, '__init__', return_value=None):
  232. with patch.object(socket.socket, 'connect', return_value=None) as mock:
  233. with patch.object(socket.socket, '_sock'):
  234. with patch.object(socket.socket, 'fileno', return_value=1):
  235. with self.assertRaises(CommError):
  236. self._device.open(no_reader_thread=True)