GithubBackup/httprunner
1
0
Fork 0
mirror of https://github.com/httprunner/httprunner.git synced 2026-06-02 14:29:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

run api_server in multiprocessing.Thread other than gevent.pywsgi.WSGIServer

Browse Source
This commit is contained in:
debugtalk
2017-06-20 13:25:59 +08:00
parent b775e8c217
commit bb199bd42e
2 changed files with 11 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
test/__init__.py
View File

@@ -1,2 +0,0 @@
from gevent import monkey
monkey.patch_all()

30
test/test_apiserver.py
View File

@@ -1,7 +1,7 @@
import gevent import multiprocessing
import gevent.pywsgi import time
import requests
import unittest import unittest
import requests
from . import api_server from . import api_server
class TestApiServer(unittest.TestCase): class TestApiServer(unittest.TestCase):
@@ -10,16 +10,17 @@ class TestApiServer(unittest.TestCase):
""" """
def setUp(self): def setUp(self):
super(TestApiServer, self).setUp() super(TestApiServer, self).setUp()
self._api_server = gevent.pywsgi.WSGIServer(("127.0.0.1", 0), api_server.app, log=None) self.api_server_process = multiprocessing.Process(
gevent.spawn(lambda: self._api_server.serve_forever()) target=api_server.app.run
gevent.sleep(0.01) )
self.host = "http://127.0.0.1:%i" % self._api_server.server_port self.api_server_process.start()
time.sleep(0.1)
self.host = "http://127.0.0.1:5000"
self.api_client = requests.Session() self.api_client = requests.Session()
def tearDown(self): def tearDown(self):
super(TestApiServer, self).tearDown() super(TestApiServer, self).tearDown()
self._api_server.stop_accepting() self.api_server_process.terminate()
self._api_server.stop()
def clear_users(self): def clear_users(self):
url = "%s/api/users" % self.host url = "%s/api/users" % self.host
@@ -59,24 +60,21 @@ class TestApiServer(unittest.TestCase):
self.assertEqual(True, resp.json()['success']) self.assertEqual(True, resp.json()['success'])
def test_create_user_not_existed(self): def test_create_user_not_existed(self):
self.clear_users()
resp = self.create_user(1000, 'user1', '123456') resp = self.create_user(1000, 'user1', '123456')
self.assertEqual(201, resp.status_code) self.assertEqual(201, resp.status_code)
self.assertEqual(True, resp.json()['success'])
def test_create_user_existed(self): def test_create_user_existed(self):
self.clear_users()
resp = self.create_user(1000, 'user1', '123456') resp = self.create_user(1000, 'user1', '123456')
resp = self.create_user(1000, 'user1', '123456') resp = self.create_user(1000, 'user1', '123456')
self.assertEqual(500, resp.status_code) self.assertEqual(500, resp.status_code)
def test_get_users_empty(self): def test_get_users_empty(self):
self.clear_users()
resp = self.get_users() resp = self.get_users()
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
self.assertEqual(resp.json()['count'], 0) self.assertEqual(resp.json()['count'], 0)
def test_get_users_not_empty(self): def test_get_users_not_empty(self):
self.clear_users()
resp = self.create_user(1000, 'user1', '123456') resp = self.create_user(1000, 'user1', '123456')
resp = self.get_users() resp = self.get_users()
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
@@ -88,39 +86,33 @@ class TestApiServer(unittest.TestCase):
self.assertEqual(resp.json()['count'], 2) self.assertEqual(resp.json()['count'], 2)
def test_get_user_not_existed(self): def test_get_user_not_existed(self):
self.clear_users()
resp = self.get_user(1000) resp = self.get_user(1000)
self.assertEqual(404, resp.status_code) self.assertEqual(404, resp.status_code)
self.assertEqual(resp.json()['success'], False) self.assertEqual(resp.json()['success'], False)
def test_get_user_existed(self): def test_get_user_existed(self):
self.clear_users()
self.create_user(1000, 'user1', '123456') self.create_user(1000, 'user1', '123456')
resp = self.get_user(1000) resp = self.get_user(1000)
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
self.assertEqual(resp.json()['success'], True) self.assertEqual(resp.json()['success'], True)
def test_update_user_not_existed(self): def test_update_user_not_existed(self):
self.clear_users()
resp = self.update_user(1000, 'user1', '123456') resp = self.update_user(1000, 'user1', '123456')
self.assertEqual(404, resp.status_code) self.assertEqual(404, resp.status_code)
self.assertEqual(resp.json()['success'], False) self.assertEqual(resp.json()['success'], False)
def test_update_user_existed(self): def test_update_user_existed(self):
self.clear_users()
self.create_user(1000, 'user1', '123456') self.create_user(1000, 'user1', '123456')
resp = self.update_user(1000, 'user2', '123456') resp = self.update_user(1000, 'user2', '123456')
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
self.assertEqual(resp.json()['data']['name'], 'user2') self.assertEqual(resp.json()['data']['name'], 'user2')
def test_delete_user_not_existed(self): def test_delete_user_not_existed(self):
self.clear_users()
resp = self.delete_user(1000) resp = self.delete_user(1000)
self.assertEqual(404, resp.status_code) self.assertEqual(404, resp.status_code)
self.assertEqual(resp.json()['success'], False) self.assertEqual(resp.json()['success'], False)
def test_delete_user_existed(self): def test_delete_user_existed(self):
self.clear_users()
self.create_user(1000, 'leo', '123456') self.create_user(1000, 'leo', '123456')
resp = self.delete_user(1000) resp = self.delete_user(1000)
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)