test_rest_api.py 6.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232
  1. import json
  2. import unittest
  3. import requests
  4. import time
  5. from flask import Flask
  6. from kalliope.core.ConfigurationManager import BrainLoader
  7. from kalliope.core.ConfigurationManager import SettingLoader
  8. from kalliope.core.RestAPI.FlaskAPI import FlaskAPI
  9. class TestRestAPI(unittest.TestCase):
  10. @classmethod
  11. def setUpClass(cls):
  12. """
  13. executed once at the beginning of the test
  14. """
  15. # rest api config
  16. sl = SettingLoader.Instance()
  17. sl.settings.rest_api.password_protected = False
  18. sl.settings.active = True
  19. sl.settings.port = 5000
  20. # prepare a test brain
  21. brain_to_test = "../Tests/brains/brain_test.yml"
  22. brain_loader = BrainLoader.Instance(file_path=brain_to_test)
  23. brain = brain_loader.brain
  24. app = Flask(__name__)
  25. cls.flask_api = FlaskAPI(app, port=5000, brain=brain)
  26. cls.flask_api.start()
  27. time.sleep(1)
  28. @classmethod
  29. def tearDownClass(cls):
  30. """
  31. executed once at the end of the test
  32. """
  33. url = "http://127.0.0.1:5000/shutdown/"
  34. requests.post(url=url)
  35. def setUp(self):
  36. self.base_url = "http://127.0.0.1:5000"
  37. def test_get_synapses(self):
  38. url = self.base_url+"/synapses/"
  39. result = requests.get(url=url)
  40. expected_content = {
  41. "synapses": [
  42. {
  43. "name": "test",
  44. "neurons": [
  45. {
  46. "say": {
  47. "message": [
  48. "test message"
  49. ]
  50. }
  51. }
  52. ],
  53. "signals": [
  54. {
  55. "order": "test_order"
  56. }
  57. ]
  58. },
  59. {
  60. "name": "test2",
  61. "neurons": [
  62. {
  63. "say": {
  64. "message": [
  65. "test message"
  66. ]
  67. }
  68. }
  69. ],
  70. "signals": [
  71. {
  72. "order": "test_order_2"
  73. }
  74. ]
  75. },
  76. {
  77. "includes": [
  78. "included_brain_test.yml"
  79. ]
  80. },
  81. {
  82. "name": "test3",
  83. "neurons": [
  84. {
  85. "say": {
  86. "message": [
  87. "test message"
  88. ]
  89. }
  90. }
  91. ],
  92. "signals": [
  93. {
  94. "order": "test_order_3"
  95. }
  96. ]
  97. }
  98. ]
  99. }
  100. self.assertEqual(expected_content, json.loads(result.content))
  101. def test_get_one_synapse(self):
  102. url = self.base_url+"/synapses/test"
  103. result = requests.get(url=url)
  104. expected_content = {
  105. "synapses": {
  106. "name": "test",
  107. "neurons": [
  108. {
  109. "say": {
  110. "message": [
  111. "test message"
  112. ]
  113. }
  114. }
  115. ],
  116. "signals": [
  117. {
  118. "order": "test_order"
  119. }
  120. ]
  121. }
  122. }
  123. self.assertEqual(expected_content, json.loads(result.content))
  124. def test_get_synapse_not_found(self):
  125. url = self.base_url + "/synapses/test-none"
  126. result = requests.get(url=url)
  127. expected_content = {
  128. "error": {
  129. "synapse name not found": "test-none"
  130. }
  131. }
  132. self.assertEqual(expected_content, json.loads(result.content))
  133. self.assertEqual(result.status_code, 404)
  134. def test_run_synapse_by_name(self):
  135. url = self.base_url + "/synapses/test"
  136. result = requests.post(url=url)
  137. expected_content = {
  138. "synapses": {
  139. "name": "test",
  140. "neurons": [
  141. {
  142. "say": {
  143. "message": [
  144. "test message"
  145. ]
  146. }
  147. }
  148. ],
  149. "signals": [
  150. {
  151. "order": "test_order"
  152. }
  153. ]
  154. }
  155. }
  156. self.assertEqual(expected_content, json.loads(result.content))
  157. self.assertEqual(result.status_code, 201)
  158. def test_post_synapse_not_found(self):
  159. url = self.base_url + "/synapses/test-none"
  160. result = requests.post(url=url)
  161. expected_content = {
  162. "error": {
  163. "synapse name not found": "test-none"
  164. }
  165. }
  166. self.assertEqual(expected_content, json.loads(result.content))
  167. self.assertEqual(result.status_code, 404)
  168. def test_run_synapse_with_order(self):
  169. url = self.base_url + "/order/"
  170. headers = {"Content-Type": "application/json"}
  171. data = {"order": "test_order"}
  172. result = requests.post(url=url, headers=headers, json=data)
  173. expected_content = {
  174. "synapses": [
  175. {
  176. "name": "test",
  177. "neurons": [
  178. {
  179. "name": "say",
  180. "parameters": "{'message': ['test message']}"
  181. }
  182. ],
  183. "signals": [
  184. {
  185. "order": "test_order"
  186. }
  187. ]
  188. }
  189. ]
  190. }
  191. self.assertEqual(expected_content, json.loads(result.content))
  192. self.assertEqual(result.status_code, 201)
  193. def test_post_synapse_by_order_not_found(self):
  194. url = self.base_url + "/order/"
  195. data = {"order": "non existing order"}
  196. headers = {"Content-Type": "application/json"}
  197. result = requests.post(url=url, headers=headers, json=data)
  198. expected_content = {'error': {'error': "The given order doesn't match any synapses"}}
  199. self.assertEqual(expected_content, json.loads(result.content))
  200. self.assertEqual(result.status_code, 400)
  201. if __name__ == '__main__':
  202. unittest.main()