test_rest_api.py 9.9 KB

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