diff --git a/circusweb/tests/test_web.ini b/circusweb/tests/test_web.ini new file mode 100644 index 0000000..e5208be --- /dev/null +++ b/circusweb/tests/test_web.ini @@ -0,0 +1,10 @@ +[circus] +check_delay = 5 +endpoint = tcp://127.0.0.1:5555 +pubsub_endpoint = tcp://127.0.0.1:5556 +stats_endpoint = tcp://127.0.0.1:5557 + +[watcher:sleeper] +cmd = sleep 120 +warmup_delay = 0 +numprocesses = 1 diff --git a/circusweb/tests/test_web.py b/circusweb/tests/test_web.py index d6e17ac..f7331b6 100644 --- a/circusweb/tests/test_web.py +++ b/circusweb/tests/test_web.py @@ -3,14 +3,9 @@ import sys from webtest import TestApp +import gevent -try: - import gevent # NOQA - from circus.web.circushttpd import app - GEVENT = True -except ImportError: - GEVENT = False - +from circusweb.circushttpd import app from circus.tests.support import TestCircus from circus.stream import QueueStream @@ -18,40 +13,40 @@ cfg = os.path.join(os.path.dirname(__file__), 'test_web.ini') -if GEVENT: - class TestHttpd(TestCircus): - def setUp(self): - TestCircus.setUp(self) - self.app = TestApp(app) - self.stream = QueueStream() - # let's run a circus - cmd = [sys.executable, "-c", - "from circus import circusd; circusd.main()", cfg] - self.p = subprocess.Popen(cmd, stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - - def tearDown(self): - self.p.terminate() - try: - with gevent.Timeout(1): - while self.p.poll() is None: - gevent.sleep(0.1) - except gevent.Timeout: - self.p.kill() - - TestCircus.tearDown(self) - - def test_index(self): - # let's open the web app - res = self.app.get('/') - - if res.status_code == 302: - res = res.follow() - - # we have a form to connect to the current app - res = res.form.submit() - - # that should be a 302, redirecting to the connected index - # let's follow it + +class TestHttpd(TestCircus): + def setUp(self): + TestCircus.setUp(self) + self.app = TestApp(app) + self.stream = QueueStream() + # let's run a circus + cmd = [sys.executable, "-c", + "from circus import circusd; circusd.main()", cfg] + self.p = subprocess.Popen(cmd, stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + + def tearDown(self): + self.p.terminate() + try: + with gevent.Timeout(1): + while self.p.poll() is None: + gevent.sleep(0.1) + except gevent.Timeout: + self.p.kill() + + TestCircus.tearDown(self) + + def test_index(self): + # let's open the web app + res = self.app.get('/') + + if res.status_code == 302: res = res.follow() - self.assertTrue('tcp://127.0.0.1:5557' in res.body) + + # we have a form to connect to the current app + res = res.form.submit() + + # that should be a 302, redirecting to the connected index + # let's follow it + res = res.follow() + self.assertTrue('tcp://127.0.0.1:5557' in res.body)