Skip to content
Snippets Groups Projects
Commit 6c1127f8 authored by John Chilton's avatar John Chilton
Browse files

Refactor test app and server creation into smaller pieces.

Will want to create a different kind of test server to mimic a potential Galaxy job files API.
parent 5e4b359d
No related branches found
No related tags found
No related merge requests found
......@@ -8,8 +8,9 @@ def test_standard_requests():
""" Tests app controller methods. These tests should be
compartmentalized. Also these methods should be made to not retest
the behavior of the associated Manager class. """
from .test_utils import test_app
with test_app(test_conf={"extra_environ": {"REMOTE_ADDR": "127.101.101.98"}}) as app:
from .test_utils import test_lwr_app
with test_lwr_app(test_conf={"extra_environ": {"REMOTE_ADDR": "127.101.101.98"}}) as app:
staging_directory = app.app.staging_directory
setup_response = app.get("/setup?job_id=12345")
setup_config = json.loads(setup_response.body)
......
......@@ -26,14 +26,14 @@ class BaseIntegrationTest(TempDirectoryTestCase):
options["jobs_directory"] = staging_directory
if kwds.get("direct_interface", None):
from .test_utils import test_app
with test_app({}, app_conf, {}) as app:
from .test_utils import test_lwr_app
with test_lwr_app({}, app_conf, {}) as app:
options = Bunch(job_manager=next(itervalues(app.app.managers)), file_cache=app.app.file_cache, **kwds)
update_options_for_app(options, app.app)
run(options)
else:
from .test_utils import test_server
with test_server(app_conf=app_conf) as server:
from .test_utils import test_lwr_server
with test_lwr_server(app_conf=app_conf) as server:
options = Bunch(url=server.application_url, **kwds)
update_options_for_app(options, server.test_app.application)
run(options)
......
......@@ -36,8 +36,8 @@ class LwrObjectStoreTest(TempDirectoryTestCase):
object_store_config_file=object_store_config_file,
private_key="12345",
)
from .test_utils import test_server
with test_server(app_conf=app_conf) as server:
from .test_utils import test_lwr_server
with test_lwr_server(app_conf=app_conf) as server:
url = server.application_url
# Define a proxy LWR object store.
proxy_object_store_config_file = join(self.temp_directory, "proxy_object_store_conf.xml")
......
......@@ -120,24 +120,30 @@ class TestDependencyManager(object):
@contextmanager
def test_server(global_conf={}, app_conf={}, test_conf={}):
with test_app(global_conf, app_conf, test_conf) as app:
try:
from paste.exceptions.errormiddleware import ErrorMiddleware
error_app = ErrorMiddleware(app.app, debug=True, error_log="errors")
except ImportError:
# paste.exceptions not available for Python 3.
error_app = app
server = StopableWSGIServer.create(error_app)
try:
server.wait()
yield server
finally:
server.shutdown()
def server_for_test_app(app):
try:
from paste.exceptions.errormiddleware import ErrorMiddleware
error_app = ErrorMiddleware(app.app, debug=True, error_log="errors")
except ImportError:
# paste.exceptions not available for Python 3.
error_app = app
server = StopableWSGIServer.create(error_app)
try:
server.wait()
yield server
finally:
server.shutdown()
@contextmanager
def test_lwr_server(global_conf={}, app_conf={}, test_conf={}):
with test_lwr_app(global_conf, app_conf, test_conf) as app:
with server_for_test_app(app) as test_lwr_server:
yield test_lwr_server
@contextmanager
def test_app(global_conf={}, app_conf={}, test_conf={}):
def test_lwr_app(global_conf={}, app_conf={}, test_conf={}):
staging_directory = mkdtemp()
# Make staging directory world executable for run as user tests.
mode = stat(staging_directory).st_mode
......@@ -149,8 +155,7 @@ def test_app(global_conf={}, app_conf={}, test_conf={}):
from lwr.app import app_factory
app = app_factory(global_conf, **app_conf)
test_app = TestApp(app, **test_conf)
yield test_app
yield TestApp(app, **test_conf)
finally:
try:
app.shutdown()
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment