Convert storageproxy tests to be pytest-able and use the new liveserver testcase pytest fixture

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):

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
+from test.registry.liveserverfixture import *
+from test.fixtures import *
 
-class TestStorageProxy(LiveServerTestCase):
-  def setUp(self):
-    setup_database_for_testing(self)
 
-  def tearDown(self):
-    finished_database_for_testing(self)
+@pytest.fixture(params=[True, False])
+def is_proxying_enabled(request):
+  return request.param
 
-  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
+@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':
-      self.test_app.config['DEBUG'] = True
+    app.config['DEBUG'] = True
 
-    self.test_app.config['TESTING'] = True
-    self.test_app.config['SERVER_HOSTNAME'] = 'localhost:%s' % _PORT_NUMBER
+  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'
 
-    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'
+  app.config['FEATURE_PROXY_STORAGE'] = is_proxying_enabled
 
-    # UGH... Such a stupid hack!
-    self.test_app.config['FEATURE_PROXY_STORAGE'] = self.id().find('notinstalled') < 0
-
-    self.test_app.config['DISTRIBUTED_STORAGE_CONFIG'] = {
+  app.config['DISTRIBUTED_STORAGE_CONFIG'] = {
     'test': ['FakeStorage', {}],
   }
+  app.config['DISTRIBUTED_STORAGE_PREFERENCE'] = ['test']
+  return app
 
-    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')
+@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 = 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)
-
-
-  @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.
+  direct_download_url = storage.get_direct_download_url(['test'], 'somepath')
   proxy_index = direct_download_url.find('/_storage_proxy/')
-    self.assertTrue(proxy_index > 0)
+  if is_proxying_enabled:
+    assert proxy_index >= 0
+  else:
+    assert proxy_index == -1
 
-    # Ensure that auth returns 200 for the URL pieces.
+  # Ensure that auth returns the expected value.
   headers = {
-      'X-Original-URI': direct_download_url[proxy_index:]
+    'X-Original-URI': direct_download_url[proxy_index:] if proxy_index else 'someurihere'
   }
 
-    resp = requests.get('http://%s/_storage_proxy_auth' % self.test_app.config['SERVER_HOSTNAME'],
-                        headers=headers)
-    self.assertEquals(200, resp.status_code)
-
-
-if __name__ == '__main__':
-  unittest.main()
+  resp = liveserver_session.get('_storage_proxy_auth', headers=headers)
+  assert resp.status_code == (404 if not is_proxying_enabled else 200)