diff --git a/test/registry_tests.py b/test/registry_tests.py index bacc9ee8a..06c7c68b7 100644 --- a/test/registry_tests.py +++ b/test/registry_tests.py @@ -1,6 +1,7 @@ import unittest import requests +from flask import request, jsonify from flask.blueprints import Blueprint from flask.ext.testing import LiveServerTestCase @@ -14,6 +15,7 @@ from endpoints.csrf import generate_csrf_token import endpoints.decorated import json +import features import tarfile @@ -30,15 +32,46 @@ except ValueError: pass -# Add a test blueprint for generating CSRF tokens. +# Add a test blueprint for generating CSRF tokens and setting feature flags. testbp = Blueprint('testbp', __name__) + @testbp.route('/csrf', methods=['GET']) def generate_csrf(): return generate_csrf_token() +@testbp.route('/feature/', methods=['POST']) +def set_feature(feature_name): + import features + old_value = features._FEATURES[feature_name].value + features._FEATURES[feature_name].value = request.get_json()['value'] + return jsonify({'old_value': old_value}) + app.register_blueprint(testbp, url_prefix='/__test') +class TestFeature(object): + """ Helper object which temporarily sets the value of a feature flag. + """ + + def __init__(self, test_case, feature_flag, test_value): + self.test_case = test_case + self.feature_flag = feature_flag + self.test_value = test_value + self.old_value = None + + def __enter__(self): + result = self.test_case.conduct('POST', '/__test/feature/' + self.feature_flag, + data=json.dumps(dict(value=self.test_value)), + headers={'Content-Type': 'application/json'}) + + result_data = json.loads(result.text) + self.old_value = result_data['old_value'] + + def __exit__(self, type, value, traceback): + self.test_case.conduct('POST', '/__test/feature/' + self.feature_flag, + data=json.dumps(dict(value=self.old_value)), + headers={'Content-Type': 'application/json'}) + class RegistryTestCase(LiveServerTestCase): maxDiff = None @@ -243,5 +276,46 @@ class RegistryTests(RegistryTestCase): self.do_pull('devtable', 'newrepo', 'devtable', 'password') + def test_public_no_anonymous_access(self): + # Turn off anonymous access. + with TestFeature(self, 'ANONYMOUS_ACCESS', False): + # Add a new repository under the public user, so we have a real repository to pull. + images = [{ + 'id': 'onlyimagehere' + }] + self.do_push('public', 'newrepo', 'public', 'password', images) + self.clearSession() + + # First try to pull the (currently private) repo as devtable, which should fail as it belongs + # to public. + self.do_pull('public', 'newrepo', 'devtable', 'password', expected_code=403) + + # Make the repository public. + self.conduct_api_login('public', 'password') + self.change_repo_visibility('public', 'newrepo', 'public') + self.clearSession() + + # Pull the repository as devtable, which should succeed because the repository is public. + self.do_pull('public', 'newrepo', 'devtable', 'password') + + + def test_private_no_anonymous_access(self): + # Turn off anonymous access. + with TestFeature(self, 'ANONYMOUS_ACCESS', False): + # Add a new repository under the public user, so we have a real repository to pull. + images = [{ + 'id': 'onlyimagehere' + }] + self.do_push('public', 'newrepo', 'public', 'password', images) + self.clearSession() + + # First try to pull the (currently private) repo as devtable, which should fail as it belongs + # to public. + self.do_pull('public', 'newrepo', 'devtable', 'password', expected_code=403) + + # Pull the repository as public, which should succeed because the repository is owned by public. + self.do_pull('public', 'newrepo', 'public', 'password') + + if __name__ == '__main__': unittest.main()