bea8b9ac53
Implement the minimal changes to the local filesystem storage driver and feed them through the distributed storage driver. Create a digest package which contains digest_tools and checksums. Fix the tests to use the new v1 endpoint locations. Fix repository.delete_instance to properly filter the generated queries to avoid most subquery deletes, but still generate them when not explicitly filtered.
108 lines
3.4 KiB
Python
108 lines
3.4 KiB
Python
import unittest
|
|
|
|
from app import app
|
|
from util.names import parse_namespace_repository
|
|
from initdb import setup_database_for_testing, finished_database_for_testing
|
|
from specs import build_index_specs
|
|
from endpoints.v1 import v1_bp
|
|
|
|
|
|
app.register_blueprint(v1_bp, url_prefix='/v1')
|
|
|
|
|
|
NO_ACCESS_USER = 'freshuser'
|
|
READ_ACCESS_USER = 'reader'
|
|
ADMIN_ACCESS_USER = 'devtable'
|
|
|
|
|
|
class EndpointTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
setup_database_for_testing(self)
|
|
|
|
def tearDown(self):
|
|
finished_database_for_testing(self)
|
|
|
|
|
|
class _SpecTestBuilder(type):
|
|
@staticmethod
|
|
def _test_generator(url, expected_status, open_kwargs, session_var_list):
|
|
def test(self):
|
|
with app.test_client() as c:
|
|
if session_var_list:
|
|
# Temporarily remove the teardown functions
|
|
teardown_funcs = []
|
|
if None in app.teardown_request_funcs:
|
|
teardown_funcs = app.teardown_request_funcs[None]
|
|
app.teardown_request_funcs[None] = []
|
|
|
|
with c.session_transaction() as sess:
|
|
for sess_key, sess_val in session_var_list:
|
|
sess[sess_key] = sess_val
|
|
|
|
# Restore the teardown functions
|
|
app.teardown_request_funcs[None] = teardown_funcs
|
|
|
|
rv = c.open(url, **open_kwargs)
|
|
msg = '%s %s: %s expected: %s' % (open_kwargs['method'], url,
|
|
rv.status_code, expected_status)
|
|
self.assertEqual(rv.status_code, expected_status, msg)
|
|
return test
|
|
|
|
|
|
def __new__(cls, name, bases, attrs):
|
|
with app.test_request_context() as ctx:
|
|
specs = attrs['spec_func']()
|
|
for test_spec in specs:
|
|
url, open_kwargs = test_spec.get_client_args()
|
|
|
|
if attrs['auth_username']:
|
|
basic_auth = test_spec.gen_basic_auth(attrs['auth_username'],
|
|
'password')
|
|
open_kwargs['headers'] = [('authorization', '%s' % basic_auth)]
|
|
|
|
session_vars = []
|
|
if test_spec.sess_repo:
|
|
ns, repo = parse_namespace_repository(test_spec.sess_repo)
|
|
session_vars.append(('namespace', ns))
|
|
session_vars.append(('repository', repo))
|
|
|
|
expected_status = getattr(test_spec, attrs['result_attr'])
|
|
test = _SpecTestBuilder._test_generator(url, expected_status,
|
|
open_kwargs,
|
|
session_vars)
|
|
|
|
test_name_url = url.replace('/', '_').replace('-', '_')
|
|
sess_repo = str(test_spec.sess_repo).replace('/', '_')
|
|
test_name = 'test_%s%s_%s' % (open_kwargs['method'].lower(),
|
|
test_name_url, sess_repo)
|
|
attrs[test_name] = test
|
|
|
|
return type(name, bases, attrs)
|
|
|
|
|
|
class TestAnonymousAccess(EndpointTestCase):
|
|
__metaclass__ = _SpecTestBuilder
|
|
spec_func = build_index_specs
|
|
result_attr = 'anon_code'
|
|
auth_username = None
|
|
|
|
|
|
class TestNoAccess(EndpointTestCase):
|
|
__metaclass__ = _SpecTestBuilder
|
|
spec_func = build_index_specs
|
|
result_attr = 'no_access_code'
|
|
auth_username = NO_ACCESS_USER
|
|
|
|
|
|
class TestReadAccess(EndpointTestCase):
|
|
__metaclass__ = _SpecTestBuilder
|
|
spec_func = build_index_specs
|
|
result_attr = 'read_code'
|
|
auth_username = READ_ACCESS_USER
|
|
|
|
|
|
class TestAdminAccess(EndpointTestCase):
|
|
__metaclass__ = _SpecTestBuilder
|
|
spec_func = build_index_specs
|
|
result_attr = 'admin_code'
|
|
auth_username = ADMIN_ACCESS_USER
|