From 544c8f0adfa67e10ac0bc0f3a4b5d1baee772aa5 Mon Sep 17 00:00:00 2001 From: Joseph Schorr Date: Fri, 1 Jun 2018 17:07:15 -0400 Subject: [PATCH] Convert storageproxy tests to be pytest-able and use the new liveserver testcase pytest fixture --- test/registry/liveserverfixture.py | 1 + test/test_storageproxy.py | 161 +++++++++++++++-------------- 2 files changed, 83 insertions(+), 79 deletions(-) diff --git a/test/registry/liveserverfixture.py b/test/registry/liveserverfixture.py index 18185a5d1..e506ef689 100644 --- a/test/registry/liveserverfixture.py +++ b/test/registry/liveserverfixture.py @@ -75,6 +75,7 @@ class liveFlaskServer(object): raise RuntimeError("Failed to start the server after %d seconds. " % timeout) if self._can_connect(): + self.app.config['SERVER_HOSTNAME'] = 'localhost:%s' % self._port_value.value break def _can_connect(self): diff --git a/test/test_storageproxy.py b/test/test_storageproxy.py index 174c3c854..b0a4577d4 100644 --- a/test/test_storageproxy.py +++ b/test/test_storageproxy.py @@ -1,92 +1,95 @@ import os + +import pytest import requests -import unittest from flask import Flask from flask_testing import LiveServerTestCase -from initdb import setup_database_for_testing, finished_database_for_testing from storage import Storage from util.security.instancekeys import InstanceKeys -_PORT_NUMBER = 5001 - -class TestStorageProxy(LiveServerTestCase): - def setUp(self): - setup_database_for_testing(self) - - def tearDown(self): - finished_database_for_testing(self) - - def create_app(self): - global _PORT_NUMBER - _PORT_NUMBER = _PORT_NUMBER + 1 - - self.test_app = Flask('teststorageproxy') - self.test_app.config['LIVESERVER_PORT'] = _PORT_NUMBER - - if os.environ.get('DEBUG') == 'true': - self.test_app.config['DEBUG'] = True - - self.test_app.config['TESTING'] = True - self.test_app.config['SERVER_HOSTNAME'] = 'localhost:%s' % _PORT_NUMBER - - self.test_app.config['INSTANCE_SERVICE_KEY_KID_LOCATION'] = 'test/data/test.kid' - self.test_app.config['INSTANCE_SERVICE_KEY_LOCATION'] = 'test/data/test.pem' - self.test_app.config['INSTANCE_SERVICE_KEY_SERVICE'] = 'quay' - - # UGH... Such a stupid hack! - self.test_app.config['FEATURE_PROXY_STORAGE'] = self.id().find('notinstalled') < 0 - - self.test_app.config['DISTRIBUTED_STORAGE_CONFIG'] = { - 'test': ['FakeStorage', {}], - } - - instance_keys = InstanceKeys(self.test_app) - self.storage = Storage(self.test_app, instance_keys=instance_keys) - self.test_app.config['DISTRIBUTED_STORAGE_PREFERENCE'] = ['test'] - return self.test_app - - @unittest.skipIf(os.environ.get('TEST_DATABASE_URI'), "not supported for non SQLite testing") - def test_storage_proxy_auth_notinstalled(self): - # Active direct download on the fake storage. - self.storage.put_content(['test'], 'supports_direct_download', 'true') - - # Get the unwrapped URL. - direct_download_url = self.storage.get_direct_download_url(['test'], 'somepath') - self.assertEquals(-1, direct_download_url.find('/_storage_proxy/')) - - # Ensure that auth returns 404. - headers = { - 'X-Original-URI': 'someurihere' - } - - resp = requests.get('http://%s/_storage_proxy_auth' % self.test_app.config['SERVER_HOSTNAME'], - headers=headers) - self.assertEquals(404, resp.status_code) +from test.registry.liveserverfixture import * +from test.fixtures import * - @unittest.skipIf(os.environ.get('TEST_DATABASE_URI'), "not supported for non SQLite testing") - def test_storage_proxy_auth(self): - # Active direct download on the fake storage. - self.storage.put_content(['test'], 'supports_direct_download', 'true') - - # Get the wrapped URL. - direct_download_url = self.storage.get_direct_download_url(['test'], 'somepath') - - # Ensure it refers to the storage proxy. - proxy_index = direct_download_url.find('/_storage_proxy/') - self.assertTrue(proxy_index > 0) - - # Ensure that auth returns 200 for the URL pieces. - headers = { - 'X-Original-URI': direct_download_url[proxy_index:] - } - - resp = requests.get('http://%s/_storage_proxy_auth' % self.test_app.config['SERVER_HOSTNAME'], - headers=headers) - self.assertEquals(200, resp.status_code) +@pytest.fixture(params=[True, False]) +def is_proxying_enabled(request): + return request.param -if __name__ == '__main__': - unittest.main() \ No newline at end of file +@pytest.fixture() +def server_executor(app): + def reload_app(server_hostname): + # Close any existing connection. + close_db_filter(None) + + # Reload the database config. + app.config['SERVER_HOSTNAME'] = server_hostname[len('http://'):] + configure(app.config) + return 'OK' + + executor = LiveServerExecutor() + executor.register('reload_app', reload_app) + return executor + + +@pytest.fixture() +def liveserver_app(app, server_executor, init_db_path, is_proxying_enabled): + server_executor.apply_blueprint_to_app(app) + + if os.environ.get('DEBUG') == 'true': + app.config['DEBUG'] = True + + app.config['TESTING'] = True + app.config['INSTANCE_SERVICE_KEY_KID_LOCATION'] = 'test/data/test.kid' + app.config['INSTANCE_SERVICE_KEY_LOCATION'] = 'test/data/test.pem' + app.config['INSTANCE_SERVICE_KEY_SERVICE'] = 'quay' + + app.config['FEATURE_PROXY_STORAGE'] = is_proxying_enabled + + app.config['DISTRIBUTED_STORAGE_CONFIG'] = { + 'test': ['FakeStorage', {}], + } + app.config['DISTRIBUTED_STORAGE_PREFERENCE'] = ['test'] + return app + + +@pytest.fixture() +def instance_keys(liveserver_app): + return InstanceKeys(liveserver_app) + + +@pytest.fixture() +def storage(liveserver_app, instance_keys): + return Storage(liveserver_app, instance_keys=instance_keys) + + +@pytest.fixture() +def app_reloader(liveserver, server_executor): + server_executor.on(liveserver).reload_app(liveserver.url) + yield + + +@pytest.mark.skipif(os.environ.get('TEST_DATABASE_URI'), + reason="not supported for non SQLite testing") +def test_storage_proxy_auth(storage, liveserver_app, liveserver_session, is_proxying_enabled, + app_reloader): + # Activate direct download on the fake storage. + storage.put_content(['test'], 'supports_direct_download', 'true') + + # Get the unwrapped URL. + direct_download_url = storage.get_direct_download_url(['test'], 'somepath') + proxy_index = direct_download_url.find('/_storage_proxy/') + if is_proxying_enabled: + assert proxy_index >= 0 + else: + assert proxy_index == -1 + + # Ensure that auth returns the expected value. + headers = { + 'X-Original-URI': direct_download_url[proxy_index:] if proxy_index else 'someurihere' + } + + resp = liveserver_session.get('_storage_proxy_auth', headers=headers) + assert resp.status_code == (404 if not is_proxying_enabled else 200)