update api_server and its tests

test/api_server.py

@@ -8,12 +8,10 @@ app = Flask(__name__)
 data structure:
     users_dict = {
         'uid1': {
-            'uid': 'uid1',
             'name': 'name1',
             'password': 'pwd1'
         },
         'uid2': {
-            'uid': 'uid2',
             'name': 'name2',
             'password': 'pwd2'
         }
@@ -21,28 +19,102 @@ data structure:
 """
 users_dict = {}
 
-@app.route('/api/user/clear')
+@app.route('/api/users')
-def clear_users():
+def get_users():
-    users_dict.clear()
+    users_list = [user for uid, user in users_dict.items()]
-    return "ok"
+    users = {
-
+        'success': True,
-@app.route('/api/user/add', methods=['POST'])
+        'count': len(users_list),
-def add_user():
+        'items': users_list
-    user = request.get_json()
+    }
-    users_dict[user["uid"]] = user
+    response = make_response(json.dumps(users))
-    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'])
+@app.route('/api/users', methods=['DELETE'])
-def delete_user(uid):
+def clear_users():
-    user = users_dict.pop(uid, None)
+    users_dict.clear()
-    if user:
+    result = {
-        return "ok"
+        'success': True
+    }
+    response = make_response(json.dumps(result))
+    response.headers["Content-Type"] = "application/json"
+    return response
+
+@app.route('/api/users/<int:uid>', methods=['POST'])
+def create_user(uid):
+    user = request.get_json()
+    if uid not in users_dict:
+        result = {
+            'success': True,
+            'msg': "user created successfully."
+        }
+        status_code = 201
+        users_dict[uid] = user
     else:
-        return "not_existed"
+        result = {
+            'success': False,
+            'msg': "user already existed."
+        }
+        status_code = 500
+
+    response = make_response(json.dumps(result), status_code)
+    response.headers["Content-Type"] = "application/json"
+    return response
+
+@app.route('/api/users/<int:uid>')
+def get_user(uid):
+    user = users_dict.get(uid, {})
+    if user:
+        result = {
+            'success': True,
+            'data': user
+        }
+        status_code = 200
+    else:
+        result = {
+            'success': False,
+            'data': user
+        }
+        status_code = 404
+
+    response = make_response(json.dumps(result), status_code)
+    response.headers["Content-Type"] = "application/json"
+    return response
+
+@app.route('/api/users/<int:uid>', methods=['PUT'])
+def update_user(uid):
+    user = users_dict.get(uid, {})
+    if user:
+        user = request.get_json()
+        success = True
+        status_code = 200
+    else:
+        success = False
+        status_code = 404
+
+    result = {
+        'success': success,
+        'data': user
+    }
+    response = make_response(json.dumps(result), status_code)
+    response.headers["Content-Type"] = "application/json"
+    return response
+
+@app.route('/api/users/<int:uid>', methods=['DELETE'])
+def delete_user(uid):
+    user = users_dict.pop(uid, {})
+    if user:
+        success = True
+        status_code = 200
+    else:
+        success = False
+        status_code = 404
+
+    result = {
+        'success': success,
+        'data': user
+    }
+    response = make_response(json.dumps(result), status_code)
+    response.headers["Content-Type"] = "application/json"
+    return response

test/test_apiserver.py

@@ -14,6 +14,7 @@ class TestApiServer(unittest.TestCase):
         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
+        self.api_client = requests.Session()
 
     def tearDown(self):
         super(TestApiServer, self).tearDown()
@@ -21,74 +22,106 @@ class TestApiServer(unittest.TestCase):
         self._api_server.stop()
 
     def clear_users(self):
-        url = "%s/api/user/clear" % self.host
+        url = "%s/api/users" % self.host
-        resp = requests.get(url)
+        return self.api_client.delete(url)
-        return resp
 
-    def add_user(self, uid, name, password):
+    def get_users(self):
-        url = "%s/api/user/add" % self.host
+        url = "%s/api/users" % self.host
+        return self.api_client.get(url)
+
+    def create_user(self, uid, name, password):
+        url = "%s/api/users/%d" % (self.host, uid)
         data = {
-            'uid': uid,
             'name': name,
             'password': password
         }
-        resp = requests.post(url, json=data)
+        return self.api_client.post(url, json=data)
-        return resp
+
+    def get_user(self, uid):
+        url = "%s/api/users/%d" % (self.host, uid)
+        return self.api_client.get(url)
+
+    def update_user(self, uid, name, password):
+        url = "%s/api/users/%d" % (self.host, uid)
+        data = {
+            'name': name,
+            'password': password
+        }
+        return self.api_client.put(url, json=data)
+
+    def delete_user(self, uid):
+        url = "%s/api/users/%d" % (self.host, uid)
+        return self.api_client.delete(url)
 
     def test_clear_users(self):
         resp = self.clear_users()
         self.assertEqual(200, resp.status_code)
-        self.assertEqual(resp.text, "ok")
+        self.assertEqual(True, resp.json()['success'])
 
-    def test_add_user_not_existed(self):
+    def test_create_user_not_existed(self):
         self.clear_users()
-        resp = self.add_user(1000, 'leo', '123456')
+        resp = self.create_user(1000, 'user1', '123456')
-        self.assertEqual(200, resp.status_code)
+        self.assertEqual(201, resp.status_code)
-        self.assertEqual(resp.text, "ok")
 
-        url = "%s/api/user/1000" % self.host
+    def test_create_user_existed(self):
-        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')
+        resp = self.create_user(1000, 'user1', '123456')
-        self.assertEqual(200, resp.status_code)
+        resp = self.create_user(1000, 'user1', '123456')
-        self.assertEqual(resp.text, "ok")
+        self.assertEqual(500, resp.status_code)
 
-        self.add_user(1000, 'leo2', '123456')
+    def test_get_users_empty(self):
-        url = "%s/api/user/1000" % self.host
+        self.clear_users()
-        resp = requests.get(url)
+        resp = self.get_users()
         self.assertEqual(200, resp.status_code)
-        self.assertEqual(resp.json()['name'], 'leo2')
+        self.assertEqual(resp.json()['count'], 0)
+
+    def test_get_users_not_empty(self):
+        self.clear_users()
+        resp = self.create_user(1000, 'user1', '123456')
+        resp = self.get_users()
+        self.assertEqual(200, resp.status_code)
+        self.assertEqual(resp.json()['count'], 1)
+
+        resp = self.create_user(1001, 'user2', '123456')
+        resp = self.get_users()
+        self.assertEqual(200, resp.status_code)
+        self.assertEqual(resp.json()['count'], 2)
 
     def test_get_user_not_existed(self):
         self.clear_users()
-        url = "%s/api/user/1000" % self.host
+        resp = self.get_user(1000)
-        resp = requests.get(url)
+        self.assertEqual(404, resp.status_code)
-        self.assertEqual(200, resp.status_code)
+        self.assertEqual(resp.json()['success'], False)
-        self.assertEqual(resp.json(), {})
 
     def test_get_user_existed(self):
         self.clear_users()
-        self.add_user(1000, 'leo', '123456')
+        self.create_user(1000, 'user1', '123456')
-        url = "%s/api/user/1000" % self.host
+        resp = self.get_user(1000)
-        resp = requests.get(url)
         self.assertEqual(200, resp.status_code)
-        self.assertEqual(resp.json()['name'], 'leo')
+        self.assertEqual(resp.json()['success'], True)
+
+    def test_update_user_not_existed(self):
+        self.clear_users()
+        resp = self.update_user(1000, 'user1', '123456')
+        self.assertEqual(404, resp.status_code)
+        self.assertEqual(resp.json()['success'], False)
+
+    def test_update_user_existed(self):
+        self.clear_users()
+        self.create_user(1000, 'user1', '123456')
+        resp = self.update_user(1000, 'user2', '123456')
+        self.assertEqual(200, resp.status_code)
+        self.assertEqual(resp.json()['data']['name'], 'user2')
 
     def test_delete_user_not_existed(self):
         self.clear_users()
-        url = "%s/api/user/1000" % self.host
+        resp = self.delete_user(1000)
-        resp = requests.delete(url)
+        self.assertEqual(404, resp.status_code)
-        self.assertEqual(200, resp.status_code)
+        self.assertEqual(resp.json()['success'], False)
-        self.assertEqual(resp.text, "not_existed")
 
     def test_delete_user_existed(self):
         self.clear_users()
-        resp = self.add_user(1000, 'leo', '123456')
+        self.create_user(1000, 'leo', '123456')
-        url = "%s/api/user/1000" % self.host
+        resp = self.delete_user(1000)
-        resp = requests.delete(url)
         self.assertEqual(200, resp.status_code)
-        self.assertEqual(resp.text, "ok")
+        self.assertEqual(resp.json()['success'], True)