create api server for test
This commit is contained in:
parent
62844038b2
commit
7cbbaa3e49
|
@ -0,0 +1,9 @@
|
|||
*.pyc
|
||||
__pycache__
|
||||
.DS_Store
|
||||
*/tmp/*
|
||||
build/*
|
||||
dist/*
|
||||
*.egg-info
|
||||
.python-version
|
||||
logs/%
|
|
@ -0,0 +1,2 @@
|
|||
from gevent import monkey
|
||||
monkey.patch_all()
|
|
@ -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/<int:uid>')
|
||||
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/<int:uid>', methods=['DELETE'])
|
||||
def delete_user(uid):
|
||||
user = users_dict.pop(uid, None)
|
||||
if user:
|
||||
return "ok"
|
||||
else:
|
||||
return "not_existed"
|
|
@ -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")
|
Loading…
Reference in New Issue