test.py 6.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198
  1. import time, os
  2. import unittest
  3. import paho.mqtt.client as mqtt
  4. import subprocess
  5. import json
  6. dir_path = os.path.dirname(os.path.realpath(__file__))
  7. count = 0
  8. cardReady = False
  9. client = mqtt.Client()
  10. client.connect('localhost')
  11. client.subscribe('/card')
  12. client.subscribe('/audio')
  13. messageBuff = []
  14. lastCardUpdate = {}
  15. lastAudioUpdate = {}
  16. def resetBuff () :
  17. global messageBuff
  18. global lastCardUpdate
  19. global lastAudioUpdate
  20. messageBuff = []
  21. lastCardUpdate = {}
  22. lastAudioUpdate = {}
  23. def on_message(client, userdata, message):
  24. payload = json.loads(message.payload)
  25. if message.topic == '/card':
  26. if payload['cmd'] == 'new_card':
  27. resp = {
  28. "action": "create",
  29. "attention": payload['responseTopic'],
  30. "cardId": 0
  31. }
  32. client.publish('/cardResponse', json.dumps(resp))
  33. client.publish('/cardResponse', json.dumps({
  34. "action":"select",
  35. "cardId": 0 }))
  36. global cardReady
  37. cardReady = True
  38. if payload['cmd'] == 'update_card':
  39. global messageBuff
  40. global lastCardUpdate
  41. messageBuff.append(payload)
  42. lastCardUpdate = payload
  43. elif message.topic == '/audio':
  44. global lastAudioUpdate
  45. lastAudioUpdate = payload
  46. client.on_message=on_message
  47. client.loop_start() #start the loop
  48. class TestTimer(unittest.TestCase):
  49. def setUp(self):
  50. self.myProcess = subprocess.Popen(['oboo-runtime/ort', '%s/timer-card/timer.js'%(dir_path)])
  51. while True:
  52. if cardReady:
  53. print 'Card Ready ...'
  54. break
  55. else:
  56. time.sleep(0.5)
  57. def tearDown(self):
  58. self.myProcess.kill()
  59. global cardReady
  60. cardReady = False
  61. def test_increase(self):
  62. for i in range(10):
  63. client.publish('/button', json.dumps({"action": "press", "id": 3}))
  64. time.sleep(0.1)
  65. time.sleep(0.5)
  66. self.assertEqual(lastCardUpdate['elements'][0]['value'], '20:00')
  67. resetBuff()
  68. def test_decrease(self):
  69. for i in range(10):
  70. client.publish('/button', json.dumps({"action": "press", "id": 2}))
  71. time.sleep(0.1)
  72. time.sleep(0.5)
  73. self.assertEqual(lastCardUpdate['elements'][0]['value'], '00:00')
  74. resetBuff()
  75. def test_increase_max(self):
  76. for i in range(70):
  77. client.publish('/button', json.dumps({"action": "press", "id": 3}))
  78. time.sleep(0.1)
  79. time.sleep(0.5)
  80. self.assertEqual(lastCardUpdate['elements'][0]['value'], '60:00')
  81. resetBuff()
  82. def test_decrease_min(self):
  83. for i in range(30):
  84. client.publish('/button', json.dumps({"action": "press", "id": 2}))
  85. time.sleep(0.1)
  86. time.sleep(0.5)
  87. self.assertEqual(lastCardUpdate['elements'][0]['value'], '00:00')
  88. resetBuff()
  89. def test_start(self):
  90. client.publish('/button', json.dumps({"action": "press", "id": 1}))
  91. time.sleep(3)
  92. self.assertEqual(lastCardUpdate['elements'][0]['value'], '09:57')
  93. resetBuff()
  94. def test_stop(self):
  95. client.publish('/button', json.dumps({"action": "press", "id": 1}))
  96. time.sleep(3)
  97. client.publish('/button', json.dumps({"action": "press", "id": 0}))
  98. time.sleep(2)
  99. self.assertEqual(lastCardUpdate['elements'][0]['value'], '10:00')
  100. time.sleep(2)
  101. self.assertEqual(lastCardUpdate['elements'][0]['value'], '10:00')
  102. resetBuff()
  103. def test_expire(self):
  104. client.publish('/test/set3', '')
  105. time.sleep(0.1)
  106. # start the timer
  107. client.publish('/button', json.dumps({"action": "press", "id": 1}))
  108. time.sleep(4)
  109. self.assertEqual(lastAudioUpdate['action'], 'play')
  110. self.assertTrue(len(lastAudioUpdate['file']) > 0)
  111. time.sleep(2)
  112. self.assertEqual(lastCardUpdate['elements'][0]['value'], '00:00')
  113. def test_modify_while_running(self):
  114. client.publish('/test/set3', '')
  115. time.sleep(0.1)
  116. # start the timer
  117. client.publish('/button', json.dumps({"action": "press", "id": 1}))
  118. time.sleep(0.5)
  119. # increment the timer
  120. for i in range(3):
  121. client.publish('/button', json.dumps({"action": "press", "id": 3}))
  122. time.sleep(0.1)
  123. self.assertEqual(lastCardUpdate['elements'][0]['value'], '03:02')
  124. # decrement the timer
  125. for i in range(6):
  126. client.publish('/button', json.dumps({"action": "press", "id": 2}))
  127. time.sleep(0.1)
  128. self.assertEqual(lastCardUpdate['elements'][0]['value'], '00:00')
  129. self.assertEqual(lastAudioUpdate['action'], 'play')
  130. self.assertTrue(len(lastAudioUpdate['file']) > 0)
  131. def test_restart_after_expiry(self):
  132. client.publish('/test/set3', '')
  133. time.sleep(0.1)
  134. # start the timer
  135. client.publish('/button', json.dumps({"action": "press", "id": 1}))
  136. time.sleep(4)
  137. self.assertEqual(lastAudioUpdate['action'], 'play')
  138. self.assertTrue(len(lastAudioUpdate['file']) > 0)
  139. time.sleep(2)
  140. # start the timer again
  141. client.publish('/button', json.dumps({"action": "press", "id": 1}))
  142. time.sleep(2)
  143. self.assertEqual(lastCardUpdate['elements'][0]['value'], '00:01')
  144. def test_stop_after_expiry(self):
  145. client.publish('/test/set3', '')
  146. time.sleep(0.1)
  147. # start the timer
  148. client.publish('/button', json.dumps({"action": "press", "id": 1}))
  149. time.sleep(4)
  150. self.assertEqual(lastAudioUpdate['action'], 'play')
  151. self.assertTrue(len(lastAudioUpdate['file']) > 0)
  152. time.sleep(2)
  153. # hit stop
  154. client.publish('/button', json.dumps({"action": "press", "id": 0}))
  155. time.sleep(2)
  156. self.assertEqual(lastCardUpdate['elements'][0]['value'], '00:03')
  157. def test_increment_while_running(self):
  158. # start the timer
  159. client.publish('/button', json.dumps({"action": "press", "id": 1}))
  160. time.sleep(4)
  161. # increment many times
  162. for i in range(70):
  163. client.publish('/button', json.dumps({"action": "press", "id": 3}))
  164. time.sleep(0.1)
  165. time.sleep(0.5)
  166. self.assertEqual(lastCardUpdate['elements'][0]['value'].split(':')[0], '59')
  167. if __name__ == '__main__':
  168. # time.sleep(40)
  169. unittest.main()