test_rest_api.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326
  1. import json
  2. import os
  3. import unittest
  4. from flask import Flask
  5. from flask_testing import LiveServerTestCase
  6. from mock import mock
  7. from kalliope._version import version_str
  8. from kalliope.core import LIFOBuffer
  9. from kalliope.core.ConfigurationManager import BrainLoader
  10. from kalliope.core.ConfigurationManager import SettingLoader
  11. from kalliope.core.Models import Singleton
  12. from kalliope.core.RestAPI.FlaskAPI import FlaskAPI
  13. class TestRestAPI(LiveServerTestCase):
  14. def tearDown(self):
  15. Singleton._instances = {}
  16. # clean the lifo
  17. LIFOBuffer.lifo_list = list()
  18. def create_app(self):
  19. """
  20. executed once at the beginning of the test
  21. """
  22. # be sure that the singleton haven't been loaded before
  23. Singleton._instances = {}
  24. current_path = os.getcwd()
  25. if "/Tests" in os.getcwd():
  26. full_path_brain_to_test = current_path + os.sep + "brains/brain_test_api.yml"
  27. self.audio_file = "files/bonjour.wav"
  28. else:
  29. full_path_brain_to_test = current_path + os.sep + "Tests/brains/brain_test_api.yml"
  30. self.audio_file = "Tests/files/bonjour.wav"
  31. # rest api config
  32. sl = SettingLoader()
  33. sl.settings.rest_api.password_protected = False
  34. sl.settings.active = True
  35. sl.settings.port = 5000
  36. sl.settings.allowed_cors_origin = "*"
  37. sl.settings.default_synapse = None
  38. # prepare a test brain
  39. brain_to_test = full_path_brain_to_test
  40. brain_loader = BrainLoader(file_path=brain_to_test)
  41. brain = brain_loader.brain
  42. self.app = Flask(__name__)
  43. self.app.config['TESTING'] = True
  44. self.flask_api = FlaskAPI(self.app, port=5000, brain=brain)
  45. self.client = self.app.test_client()
  46. return self.flask_api.app
  47. def test_server_is_up_and_running(self):
  48. # response = urllib2.urlopen(self.get_server_url())
  49. response = self.client.get(self.get_server_url())
  50. self.assertEqual(response.status_code, 200)
  51. def test_get_main_page(self):
  52. url = self.get_server_url() + "/"
  53. response = self.client.get(url)
  54. expected_content = {
  55. "Kalliope version": "%s" % version_str
  56. }
  57. self.assertEqual(json.dumps(expected_content, sort_keys=True),
  58. json.dumps(json.loads(response.get_data().decode('utf-8')), sort_keys=True))
  59. def test_get_all_synapses(self):
  60. url = self.get_server_url()+"/synapses"
  61. response = self.client.get(url)
  62. expected_content = {
  63. "synapses": [{
  64. "name": "test",
  65. "neurons": [{
  66. "name": "say",
  67. "parameters": {
  68. "message": ["test message"]
  69. }
  70. }],
  71. "signals": [{
  72. "order": "test_order"
  73. }]
  74. }, {
  75. "name": "test2",
  76. "neurons": [{
  77. "name": "say",
  78. "parameters": {
  79. "message": ["test message"]
  80. }
  81. }],
  82. "signals": [{
  83. "order": "bonjour"
  84. }]
  85. }, {
  86. "name": "test4",
  87. "neurons": [{
  88. "name": "say",
  89. "parameters": {
  90. "message": ["test message {{parameter1}}"]
  91. }
  92. }],
  93. "signals": [{
  94. "order": "test_order_with_parameter"
  95. }]
  96. }, {
  97. "name": "test3",
  98. "neurons": [{
  99. "name": "say",
  100. "parameters": {
  101. "message": ["test message"]
  102. }
  103. }],
  104. "signals": [{
  105. "order": "test_order_3"
  106. }]
  107. }]
  108. }
  109. # a lot of char ti process
  110. self.maxDiff = None
  111. self.assertEqual(response.status_code, 200)
  112. self.assertEqual(json.dumps(expected_content, sort_keys=True),
  113. json.dumps(json.loads(response.get_data().decode('utf-8')), sort_keys=True))
  114. def test_get_one_synapse(self):
  115. url = self.get_server_url() + "/synapses/test"
  116. response = self.client.get(url)
  117. expected_content = {
  118. "synapses": {
  119. "name": "test",
  120. "neurons": [
  121. {
  122. "name": "say",
  123. "parameters": {
  124. "message": [
  125. "test message"
  126. ]
  127. }
  128. }
  129. ],
  130. "signals": [
  131. {
  132. "order": "test_order"
  133. }
  134. ]
  135. }
  136. }
  137. self.assertEqual(json.dumps(expected_content, sort_keys=True),
  138. json.dumps(json.loads(response.get_data().decode('utf-8')), sort_keys=True))
  139. def test_get_synapse_not_found(self):
  140. url = self.get_server_url() + "/synapses/test-none"
  141. result = self.client.get(url)
  142. expected_content = {
  143. "error": {
  144. "synapse name not found": "test-none"
  145. }
  146. }
  147. self.assertEqual(expected_content, json.loads(result.get_data().decode('utf-8')))
  148. self.assertEqual(result.status_code, 404)
  149. def test_run_synapse_by_name(self):
  150. url = self.get_server_url() + "/synapses/start/id/test"
  151. result = self.client.post(url)
  152. expected_content = {'status': 'complete',
  153. 'matched_synapses':
  154. [{'matched_order': None,
  155. 'neuron_module_list':
  156. [{'generated_message': 'test message', 'neuron_name': 'Say'}],
  157. 'synapse_name': 'test'}],
  158. 'user_order': None
  159. }
  160. self.assertEqual(json.dumps(expected_content, sort_keys=True),
  161. json.dumps(json.loads(result.get_data().decode('utf-8')), sort_keys=True))
  162. self.assertEqual(result.status_code, 201)
  163. # run a synapse by its name with parameter
  164. url = self.get_server_url() + "/synapses/start/id/test4"
  165. headers = {"Content-Type": "application/json"}
  166. data = {"parameters": {"parameter1": "replaced_value"}}
  167. result = self.client.post(url, headers=headers, data=json.dumps(data))
  168. expected_content = {
  169. "matched_synapses": [
  170. {
  171. "matched_order": None,
  172. "neuron_module_list": [
  173. {
  174. "generated_message": "test message replaced_value",
  175. "neuron_name": "Say"
  176. }
  177. ],
  178. "synapse_name": "test4"
  179. }
  180. ],
  181. "status": "complete",
  182. "user_order": None
  183. }
  184. self.assertEqual(json.dumps(expected_content, sort_keys=True),
  185. json.dumps(json.loads(result.get_data().decode('utf-8')), sort_keys=True))
  186. self.assertEqual(result.status_code, 201)
  187. def test_post_synapse_not_found(self):
  188. url = self.get_server_url() + "/synapses/start/id/test-none"
  189. result = self.client.post(url)
  190. expected_content = {
  191. "error": {
  192. "synapse name not found": "test-none"
  193. }
  194. }
  195. self.assertEqual(json.dumps(expected_content, sort_keys=True),
  196. json.dumps(json.loads(result.get_data().decode('utf-8')), sort_keys=True))
  197. self.assertEqual(result.status_code, 404)
  198. def test_run_synapse_with_order(self):
  199. url = self.get_server_url() + "/synapses/start/order"
  200. headers = {"Content-Type": "application/json"}
  201. data = {"order": "test_order"}
  202. result = self.client.post(url, headers=headers, data=json.dumps(data))
  203. expected_content = {'status': 'complete',
  204. 'matched_synapses':
  205. [
  206. {
  207. 'matched_order': "test_order",
  208. 'neuron_module_list':
  209. [
  210. {
  211. 'generated_message': 'test message', 'neuron_name': 'Say'
  212. }
  213. ],
  214. 'synapse_name': 'test'
  215. }
  216. ],
  217. 'user_order': "test_order"
  218. }
  219. self.assertEqual(json.dumps(expected_content, sort_keys=True),
  220. json.dumps(json.loads(result.get_data().decode('utf-8')), sort_keys=True))
  221. self.assertEqual(result.status_code, 201)
  222. def test_post_synapse_by_order_not_found(self):
  223. url = self.get_server_url() + "/synapses/start/order"
  224. data = {"order": "non existing order"}
  225. headers = {"Content-Type": "application/json"}
  226. result = self.client.post(url,
  227. headers=headers,
  228. data=json.dumps(data))
  229. expected_content = {'status': None, 'matched_synapses': [], 'user_order': u'non existing order'}
  230. self.assertEqual(json.dumps(expected_content, sort_keys=True),
  231. json.dumps(json.loads(result.get_data().decode('utf-8')), sort_keys=True))
  232. self.assertEqual(result.status_code, 201)
  233. # TODO this doesn't work on travis but works locally with python -m unittest discover
  234. # def test_post_synapse_by_audio(self):
  235. # url = self.get_server_url() + "/synapses/start/audio"
  236. # with open(os.path.join(self.audio_file), 'rb') as fp:
  237. # file = FileStorage(fp)
  238. # data = {
  239. # 'file': file
  240. # }
  241. # result = self.client.post(url, data=data, content_type='multipart/form-data')
  242. #
  243. # expected_content = {
  244. # "synapses": [
  245. # {
  246. # "name": "test2",
  247. # "neurons": [
  248. # {
  249. # "name": "say",
  250. # "parameters": {
  251. # "message": [
  252. # "test message"
  253. # ]
  254. # }
  255. # }
  256. # ],
  257. # "signals": [
  258. # {
  259. # "order": "bonjour"
  260. # }
  261. # ]
  262. # }
  263. # ]
  264. # }
  265. #
  266. # self.assertEqual(json.dumps(expected_content), json.dumps(json.loads(result.get_data())))
  267. # self.assertEqual(result.status_code, 201)
  268. def test_convert_to_wav(self):
  269. """
  270. Test the api function to convert incoming sound file to wave.
  271. """
  272. with mock.patch("os.system") as mock_os_system:
  273. # Scenario 1 : input wav file
  274. temp_file = "/tmp/kalliope/tempfile.wav" # tempfile.NamedTemporaryFile(suffix=".wav")
  275. result_file = FlaskAPI._convert_to_wav(temp_file)
  276. self.assertEqual(temp_file, result_file)
  277. mock_os_system.assert_not_called()
  278. # Scenario 2 : input not a wav file
  279. temp_file = "/tmp/kalliope/tempfile.amr" # tempfile.NamedTemporaryFile(suffix=".wav")
  280. expected_result = "/tmp/kalliope/tempfile.wav"
  281. result_file = FlaskAPI._convert_to_wav(temp_file)
  282. self.assertEqual(expected_result, result_file)
  283. mock_os_system.assert_called_once_with("avconv -y -i " + temp_file + " " + expected_result)
  284. if __name__ == '__main__':
  285. unittest.main()
  286. # suite = unittest.TestSuite()
  287. # suite.addTest(TestRestAPI("test_run_synapse_by_name"))
  288. # runner = unittest.TextTestRunner()
  289. # runner.run(suite)