diff --git a/.gitignore b/.gitignore new file mode 100644 index 00000000..58c55ee7 --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +*.pyc +__pycache__ +.DS_Store +*/tmp/* +build/* +dist/* +*.egg-info +.python-version +logs/% \ No newline at end of file diff --git a/test/__init__.py b/test/__init__.py new file mode 100644 index 00000000..91d07df5 --- /dev/null +++ b/test/__init__.py @@ -0,0 +1,2 @@ +from gevent import monkey +monkey.patch_all() \ No newline at end of file diff --git a/test/api_server.py b/test/api_server.py new file mode 100644 index 00000000..a99d92f7 --- /dev/null +++ b/test/api_server.py @@ -0,0 +1,48 @@ +import json +from flask import Flask +from flask import request, make_response + +app = Flask(__name__) + +""" storage all users' data +data structure: + users_dict = { + 'uid1': { + 'uid': 'uid1', + 'name': 'name1', + 'password': 'pwd1' + }, + 'uid2': { + 'uid': 'uid2', + 'name': 'name2', + 'password': 'pwd2' + } + } +""" +users_dict = {} + +@app.route('/api/user/clear') +def clear_users(): + users_dict.clear() + return "ok" + +@app.route('/api/user/add', methods=['POST']) +def add_user(): + user = request.get_json() + users_dict[user["uid"]] = user + return "ok" + +@app.route('/api/user/') +def get_user(uid): + user = users_dict.get(uid, {}) + response = make_response(json.dumps(user)) + response.headers["Content-Type"] = "application/json" + return response + +@app.route('/api/user/', methods=['DELETE']) +def delete_user(uid): + user = users_dict.pop(uid, None) + if user: + return "ok" + else: + return "not_existed" diff --git a/test/test_apiserver.py b/test/test_apiserver.py new file mode 100644 index 00000000..18b89c52 --- /dev/null +++ b/test/test_apiserver.py @@ -0,0 +1,94 @@ +import gevent +import gevent.pywsgi +import requests +import unittest +from . import api_server + +class TestApiServer(unittest.TestCase): + """ + Test case class that sets up an HTTP server which can be used within the tests + """ + def setUp(self): + super(TestApiServer, self).setUp() + self._api_server = gevent.pywsgi.WSGIServer(("127.0.0.1", 0), api_server.app, log=None) + gevent.spawn(lambda: self._api_server.serve_forever()) + gevent.sleep(0.01) + self.host = "http://127.0.0.1:%i" % self._api_server.server_port + + def tearDown(self): + super(TestApiServer, self).tearDown() + self._api_server.stop_accepting() + self._api_server.stop() + + def clear_users(self): + url = "%s/api/user/clear" % self.host + resp = requests.get(url) + return resp + + def add_user(self, uid, name, password): + url = "%s/api/user/add" % self.host + data = { + 'uid': uid, + 'name': name, + 'password': password + } + resp = requests.post(url, json=data) + return resp + + def test_clear_users(self): + resp = self.clear_users() + self.assertEqual(200, resp.status_code) + self.assertEqual(resp.text, "ok") + + def test_add_user_not_existed(self): + self.clear_users() + resp = self.add_user(1000, 'leo', '123456') + self.assertEqual(200, resp.status_code) + self.assertEqual(resp.text, "ok") + + url = "%s/api/user/1000" % self.host + resp = requests.get(url) + self.assertEqual(200, resp.status_code) + self.assertNotEqual(resp.json(), {}) + + def test_add_user_existed(self): + self.clear_users() + resp = self.add_user(1000, 'leo', '123456') + self.assertEqual(200, resp.status_code) + self.assertEqual(resp.text, "ok") + + self.add_user(1000, 'leo2', '123456') + url = "%s/api/user/1000" % self.host + resp = requests.get(url) + self.assertEqual(200, resp.status_code) + self.assertEqual(resp.json()['name'], 'leo2') + + def test_get_user_not_existed(self): + self.clear_users() + url = "%s/api/user/1000" % self.host + resp = requests.get(url) + self.assertEqual(200, resp.status_code) + self.assertEqual(resp.json(), {}) + + def test_get_user_existed(self): + self.clear_users() + self.add_user(1000, 'leo', '123456') + url = "%s/api/user/1000" % self.host + resp = requests.get(url) + self.assertEqual(200, resp.status_code) + self.assertEqual(resp.json()['name'], 'leo') + + def test_delete_user_not_existed(self): + self.clear_users() + url = "%s/api/user/1000" % self.host + resp = requests.delete(url) + self.assertEqual(200, resp.status_code) + self.assertEqual(resp.text, "not_existed") + + def test_delete_user_existed(self): + self.clear_users() + resp = self.add_user(1000, 'leo', '123456') + url = "%s/api/user/1000" % self.host + resp = requests.delete(url) + self.assertEqual(200, resp.status_code) + self.assertEqual(resp.text, "ok")