Add registry tests for anonymous access

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/<feature_name>', 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()