test_rest_api.py 9.8 KB

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