diff --git a/test/fixtures.py b/test/fixtures.py
index b07e63a80..68e461510 100644
--- a/test/fixtures.py
+++ b/test/fixtures.py
@@ -6,9 +6,11 @@ import pytest
 import shutil
 from flask import Flask, jsonify
 from flask_login import LoginManager
+from flask_principal import identity_loaded, Permission, Identity, identity_changed, Principal
 from peewee import SqliteDatabase, savepoint, InternalError
 
 from app import app as application
+from auth.permissions import on_identity_loaded
 from data import model
 from data.database import close_db_filter, db, configure
 from data.model.user import LoginWrappedDBUser
@@ -119,6 +121,9 @@ def appconfig(database_uri):
       'autorollback': True,
     },
     "DB_TRANSACTION_FACTORY": _create_transaction,
+    "DATA_MODEL_CACHE_CONFIG": {
+      'engine': 'inmemory',
+    },
   }
   return conf
 
@@ -168,6 +173,12 @@ def app(appconfig, initialized_db):
   def load_user(user_uuid):
     return LoginWrappedDBUser(user_uuid)
 
+  @identity_loaded.connect_via(app)
+  def on_identity_loaded_for_test(sender, identity):
+    on_identity_loaded(sender, identity)
+
+  Principal(app, use_sessions=False)
+
   app.url_map.converters['regex'] = RegexConverter
   app.url_map.converters['apirepopath'] = APIRepositoryPathConverter
   app.url_map.converters['repopath'] = RepositoryPathConverter
diff --git a/test/registry/fixtures.py b/test/registry/fixtures.py
index abfb4d95c..bb079d56f 100644
--- a/test/registry/fixtures.py
+++ b/test/registry/fixtures.py
@@ -1,5 +1,6 @@
 import copy
 import logging.config
+import json
 import os
 import shutil
 
@@ -26,7 +27,7 @@ def registry_server_executor(app):
     return generate_csrf_token()
 
   def set_supports_direct_download(enabled):
-    storage.put_content(['local_us'], 'supports_direct_download', enabled)
+    storage.put_content(['local_us'], 'supports_direct_download', 'true' if enabled else 'false')
     return 'OK'
 
   def delete_image(image_id):
@@ -61,7 +62,7 @@ def registry_server_executor(app):
                                                       'write')
     another_token.code = 'somecooltokencode'
     another_token.save()
-    return 'OK'
+    return another_token.code
 
   def break_database():
     # Close any existing connection.
@@ -93,6 +94,11 @@ def registry_server_executor(app):
 
     return 'OK'
 
+  def create_app_repository(namespace, name):
+    user = model.user.get_user(namespace)
+    model.repository.create_repository(namespace, name, user, repo_kind='application')
+    return 'OK'
+
   executor = LiveServerExecutor()
   executor.register('generate_csrf', generate_csrf)
   executor.register('set_supports_direct_download', set_supports_direct_download)
@@ -104,6 +110,7 @@ def registry_server_executor(app):
   executor.register('add_token', add_token)
   executor.register('break_database', break_database)
   executor.register('reload_app', reload_app)
+  executor.register('create_app_repository', create_app_repository)
   return executor
 
 
@@ -148,7 +155,46 @@ class FeatureFlagValue(object):
 
   def __enter__(self):
     result = self.executor.set_feature(self.feature_flag, self.test_value)
-    self.old_value = result.json['old_value']
+    self.old_value = result.json()['old_value']
 
   def __exit__(self, type, value, traceback):
     self.executor.set_feature(self.feature_flag, self.old_value)
+
+
+class ApiCaller(object):
+  def __init__(self, liveserver_session, registry_server_executor):
+    self.liveserver_session = liveserver_session
+    self.csrf_token = registry_server_executor.on_session(liveserver_session).generate_csrf()
+
+  def conduct_auth(self, username, password):
+    r = self.post('/api/v1/signin',
+                  data=json.dumps(dict(username=username, password=password)),
+                  headers={'Content-Type': 'application/json'})
+    assert r.status_code == 200
+
+  def _adjust_params(self, kwargs):
+    if 'params' not in kwargs:
+      kwargs['params'] = {}
+
+    kwargs['params'].update({
+      '_csrf_token': self.csrf_token,
+    })
+    return kwargs
+
+  def get(self, url, **kwargs):
+    kwargs = self._adjust_params(kwargs)
+    return self.liveserver_session.get(url, **kwargs)
+
+  def post(self, url, **kwargs):
+    kwargs = self._adjust_params(kwargs)
+    return self.liveserver_session.post(url, **kwargs)
+
+  def change_repo_visibility(self, namespace, repository, visibility):
+    self.post('/api/v1/repository/%s/%s/changevisibility' % (namespace, repository),
+              data=json.dumps(dict(visibility=visibility)),
+              headers={'Content-Type': 'application/json'})
+
+
+@pytest.fixture(scope="function")
+def api_caller(liveserver, registry_server_executor):
+  return ApiCaller(liveserver.new_session(), registry_server_executor)
diff --git a/test/registry/protocol_fixtures.py b/test/registry/protocol_fixtures.py
index 0c678b85a..44ad89964 100644
--- a/test/registry/protocol_fixtures.py
+++ b/test/registry/protocol_fixtures.py
@@ -1,3 +1,6 @@
+import random
+import string
+
 import pytest
 
 from Crypto.PublicKey import RSA
@@ -12,11 +15,23 @@ from test.registry.protocol_v2 import V2Protocol
 def basic_images():
   """ Returns basic images for push and pull testing. """
   # Note: order is from base layer down to leaf.
+  parent_bytes = layer_bytes_for_contents('parent contents')
+  image_bytes = layer_bytes_for_contents('some contents')
   return [
-    Image(id='parentid', bytes=layer_bytes_for_contents('parent contents'),
-          parent_id=None, size=None),
-    Image(id='someid', bytes=layer_bytes_for_contents('some contents'),
-          parent_id='parentid', size=None),
+    Image(id='parentid', bytes=parent_bytes, parent_id=None),
+    Image(id='someid', bytes=image_bytes, parent_id='parentid'),
+  ]
+
+
+@pytest.fixture(scope="session")
+def sized_images():
+  """ Returns basic images (with sizes) for push and pull testing. """
+  # Note: order is from base layer down to leaf.
+  parent_bytes = layer_bytes_for_contents('parent contents', mode='')
+  image_bytes = layer_bytes_for_contents('some contents', mode='')
+  return [
+    Image(id='parentid', bytes=parent_bytes, parent_id=None, size=len(parent_bytes)),
+    Image(id='someid', bytes=image_bytes, parent_id='parentid', size=len(image_bytes)),
   ]
 
 
@@ -25,6 +40,26 @@ def jwk():
   return RSAKey(key=RSA.generate(2048))
 
 
+@pytest.fixture(params=[V2Protocol])
+def v2_protocol(request, jwk):
+  return request.param(jwk)
+
+
+@pytest.fixture(params=[V1Protocol])
+def v1_protocol(request, jwk):
+  return request.param(jwk)
+
+
+@pytest.fixture(params=[V2Protocol])
+def manifest_protocol(request, jwk):
+  return request.param(jwk)
+
+
+@pytest.fixture(params=[V1Protocol, V2Protocol])
+def loginer(request, jwk):
+  return request.param(jwk)
+
+
 @pytest.fixture(params=[V1Protocol, V2Protocol])
 def pusher(request, jwk):
   return request.param(jwk)
@@ -33,3 +68,10 @@ def pusher(request, jwk):
 @pytest.fixture(params=[V1Protocol, V2Protocol])
 def puller(request, jwk):
   return request.param(jwk)
+
+
+@pytest.fixture(scope="session")
+def random_layer_data():
+  size = 4096
+  contents = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(size))
+  return layer_bytes_for_contents(contents)
diff --git a/test/registry/protocol_v1.py b/test/registry/protocol_v1.py
index f38aa6bb2..a3da149b2 100644
--- a/test/registry/protocol_v1.py
+++ b/test/registry/protocol_v1.py
@@ -12,6 +12,7 @@ class V1ProtocolSteps(Enum):
   """ Defines the various steps of the protocol, for matching failures. """
   PUT_IMAGES = 'put-images'
   GET_IMAGES = 'get-images'
+  PUT_TAG = 'put-tag'
 
 
 class V1Protocol(RegistryProtocol):
@@ -19,10 +20,21 @@ class V1Protocol(RegistryProtocol):
     V1ProtocolSteps.PUT_IMAGES: {
       Failures.UNAUTHENTICATED: 401,
       Failures.UNAUTHORIZED: 403,
+      Failures.APP_REPOSITORY: 405,
+      Failures.INVALID_REPOSITORY: 404,
+      Failures.DISALLOWED_LIBRARY_NAMESPACE: 400,
     },
     V1ProtocolSteps.GET_IMAGES: {
       Failures.UNAUTHENTICATED: 403,
       Failures.UNAUTHORIZED: 403,
+      Failures.APP_REPOSITORY: 405,
+      Failures.ANONYMOUS_NOT_ALLOWED: 401,
+      Failures.DISALLOWED_LIBRARY_NAMESPACE: 400,
+    },
+    V1ProtocolSteps.PUT_TAG: {
+      Failures.MISSING_TAG: 404,
+      Failures.INVALID_TAG: 400,
+      Failures.INVALID_IMAGES: 400,
     },
   }
 
@@ -38,53 +50,70 @@ class V1Protocol(RegistryProtocol):
   def ping(self, session):
     assert session.get('/v1/_ping').status_code == 200
 
+  def login(self, session, username, password, scopes, expect_success):
+    data = {
+      'username': username,
+      'password': password,
+    }
+
+    response = self.conduct(session, 'POST', '/v1/users/', json_data=data, expected_status=400)
+    assert (response.text == '"Username or email already exists"') == expect_success
+
   def pull(self, session, namespace, repo_name, tag_names, images, credentials=None,
            expected_failure=None, options=None):
     options = options or ProtocolOptions()
     auth = self._auth_for_credentials(credentials)
     tag_names = [tag_names] if isinstance(tag_names, str) else tag_names
-    prefix = '/v1/repositories/%s/%s/' % (namespace, repo_name)
+    prefix = '/v1/repositories/%s/' % self.repo_name(namespace, repo_name)
 
     # Ping!
     self.ping(session)
 
     # GET /v1/repositories/{namespace}/{repository}/images
-    result = self.conduct(session, 'GET', prefix + 'images', auth=auth,
+    headers = {'X-Docker-Token': 'true'}
+    result = self.conduct(session, 'GET', prefix + 'images', auth=auth, headers=headers,
                           expected_status=(200, expected_failure, V1ProtocolSteps.GET_IMAGES))
     if expected_failure is not None:
       return
 
+    headers = {}
+    if credentials is not None:
+      headers['Authorization'] = 'token ' + result.headers['www-authenticate']
+    else:
+      assert not 'www-authenticate' in result.headers
+
     # GET /v1/repositories/{namespace}/{repository}/tags
-    tags_result = self.conduct(session, 'GET', prefix + 'tags', auth=auth).json()
-    assert len(tags_result.values()) == len(tag_names)
+    image_ids = self.conduct(session, 'GET', prefix + 'tags', headers=headers).json()
+    assert len(image_ids.values()) == len(tag_names)
 
-    tag_image_id = tags_result['latest']
-    if not options.munge_shas:
-      # Ensure we have a matching image ID.
-      known_ids = {image.id for image in images}
-      assert tag_image_id in known_ids
+    for tag_name in tag_names:
+      tag_image_id = image_ids[tag_name]
+      if not options.munge_shas:
+        # Ensure we have a matching image ID.
+        known_ids = {image.id for image in images}
+        assert tag_image_id in known_ids
 
-    # Retrieve the ancestry of the tagged image.
-    image_prefix = '/v1/images/%s/' % tag_image_id
-    ancestors = self.conduct(session, 'GET', image_prefix + 'ancestry', auth=auth).json()
+      # Retrieve the ancestry of the tagged image.
+      image_prefix = '/v1/images/%s/' % tag_image_id
+      ancestors = self.conduct(session, 'GET', image_prefix + 'ancestry', headers=headers).json()
 
-    assert len(ancestors) == len(images)
-    for index, image_id in enumerate(reversed(ancestors)):
-      # /v1/images/{imageID}/{ancestry, json, layer}
-      image_prefix = '/v1/images/%s/' % image_id
-      self.conduct(session, 'GET', image_prefix + 'ancestry', auth=auth)
+      assert len(ancestors) == len(images)
+      for index, image_id in enumerate(reversed(ancestors)):
+        # /v1/images/{imageID}/{ancestry, json, layer}
+        image_prefix = '/v1/images/%s/' % image_id
+        self.conduct(session, 'GET', image_prefix + 'ancestry', headers=headers)
 
-      result = self.conduct(session, 'GET', image_prefix + 'json', auth=auth)
-      assert result.json()['id'] == image_id
+        result = self.conduct(session, 'GET', image_prefix + 'json', headers=headers)
+        assert result.json()['id'] == image_id
 
-      # Ensure we can HEAD the image layer.
-      self.conduct(session, 'HEAD', image_prefix + 'layer', auth=auth)
+        # Ensure we can HEAD the image layer.
+        self.conduct(session, 'HEAD', image_prefix + 'layer', headers=headers)
 
-      # And retrieve the layer data.
-      result = self.conduct(session, 'GET', image_prefix + 'layer', auth=auth)
-      assert result.content == images[index].bytes
+        # And retrieve the layer data.
+        result = self.conduct(session, 'GET', image_prefix + 'layer', headers=headers)
+        assert result.content == images[index].bytes
 
-    return PullResult(manifests=None)
+    return PullResult(manifests=None, image_ids=image_ids)
 
   def push(self, session, namespace, repo_name, tag_names, images, credentials=None,
            expected_failure=None, options=None):
@@ -95,7 +124,8 @@ class V1Protocol(RegistryProtocol):
     self.ping(session)
 
     # PUT /v1/repositories/{namespace}/{repository}/
-    result = self.conduct(session, 'PUT', '/v1/repositories/%s/%s/' % (namespace, repo_name),
+    result = self.conduct(session, 'PUT',
+                          '/v1/repositories/%s/' % self.repo_name(namespace, repo_name),
                           expected_status=(201, expected_failure, V1ProtocolSteps.PUT_IMAGES),
                           json_data={},
                           auth=auth)
@@ -115,6 +145,9 @@ class V1Protocol(RegistryProtocol):
       if image.parent_id is not None:
         image_json_data['parent'] = image.parent_id
 
+      if image.config is not None:
+        image_json_data['config'] = image.config
+
       self.conduct(session, 'PUT', '/v1/images/%s/json' % image.id,
                    json_data=image_json_data, headers=headers)
 
@@ -133,12 +166,14 @@ class V1Protocol(RegistryProtocol):
     # PUT /v1/repositories/{namespace}/{repository}/tags/latest
     for tag_name in tag_names:
       self.conduct(session, 'PUT',
-                   '/v1/repositories/%s/%s/tags/%s' % (namespace, repo_name, tag_name),
+                   '/v1/repositories/%s/tags/%s' % (self.repo_name(namespace, repo_name), tag_name),
                    data='"%s"' % images[-1].id,
-                   headers=headers)
+                   headers=headers,
+                   expected_status=(200, expected_failure, V1ProtocolSteps.PUT_TAG))
 
     # PUT /v1/repositories/{namespace}/{repository}/images
-    self.conduct(session, 'PUT', '/v1/repositories/%s/%s/images' % (namespace, repo_name),
+    self.conduct(session, 'PUT',
+                 '/v1/repositories/%s/images' % self.repo_name(namespace, repo_name),
                  expected_status=204, headers=headers)
 
-    return PushResult(checksums=None, manifests=None)
+    return PushResult(checksums=None, manifests=None, headers=headers)
diff --git a/test/registry/protocol_v2.py b/test/registry/protocol_v2.py
index ab626bcca..9c4be8dfb 100644
--- a/test/registry/protocol_v2.py
+++ b/test/registry/protocol_v2.py
@@ -14,18 +14,30 @@ class V2ProtocolSteps(Enum):
   AUTH = 'auth'
   BLOB_HEAD_CHECK = 'blob-head-check'
   GET_MANIFEST = 'get-manifest'
+  PUT_MANIFEST = 'put-manifest'
 
 
 class V2Protocol(RegistryProtocol):
   FAILURE_CODES = {
     V2ProtocolSteps.AUTH: {
       Failures.UNAUTHENTICATED: 401,
-      Failures.UNAUTHORIZED: 403,
       Failures.INVALID_REGISTRY: 400,
       Failures.APP_REPOSITORY: 405,
+      Failures.ANONYMOUS_NOT_ALLOWED: 401,
+      Failures.INVALID_REPOSITORY: 400,
     },
     V2ProtocolSteps.GET_MANIFEST: {
       Failures.UNKNOWN_TAG: 404,
+      Failures.UNAUTHORIZED: 403,
+      Failures.DISALLOWED_LIBRARY_NAMESPACE: 400,
+    },
+    V2ProtocolSteps.PUT_MANIFEST: {
+      Failures.DISALLOWED_LIBRARY_NAMESPACE: 400,
+      Failures.MISSING_TAG: 404,
+      Failures.INVALID_TAG: 400,
+      Failures.INVALID_IMAGES: 400,
+      Failures.INVALID_BLOB: 400,
+      Failures.UNSUPPORTED_CONTENT_TYPE: 415,
     },
   }
 
@@ -37,7 +49,27 @@ class V2Protocol(RegistryProtocol):
     assert result.status_code == 401
     assert result.headers['Docker-Distribution-API-Version'] == 'registry/2.0'
 
-  def auth(self, session, credentials, namespace, repository, scopes=None,
+  def login(self, session, username, password, scopes, expect_success):
+    scopes = scopes if isinstance(scopes, list) else [scopes]
+    params = {
+      'account': username,
+      'service': 'localhost:5000',
+      'scope': scopes,
+    }
+
+    auth = (username, password)
+    if not username or not password:
+      auth = None
+
+    response = session.get('/v2/auth', params=params, auth=auth)
+    if expect_success:
+      assert response.status_code / 100 == 2
+    else:
+      assert response.status_code / 100 == 4
+
+    return response
+
+  def auth(self, session, credentials, namespace, repo_name, scopes=None,
            expected_failure=None):
     """
     Performs the V2 Auth flow, returning the token (if any) and the response.
@@ -47,6 +79,8 @@ class V2Protocol(RegistryProtocol):
 
     scopes = scopes or []
     auth = None
+    username = None
+
     if credentials is not None:
       username, _ = credentials
       auth = credentials
@@ -57,7 +91,8 @@ class V2Protocol(RegistryProtocol):
     }
 
     if scopes:
-      params['scope'] = 'repository:%s/%s:%s' % (namespace, repository, ','.join(scopes))
+      params['scope'] = 'repository:%s:%s' % (self.repo_name(namespace, repo_name),
+                                              ','.join(scopes))
 
     response = self.conduct(session, 'GET', '/v2/auth', params=params, auth=auth,
                             expected_status=(200, expected_failure, V2ProtocolSteps.AUTH))
@@ -99,7 +134,14 @@ class V2Protocol(RegistryProtocol):
         if options.manifest_invalid_blob_references:
           checksum = 'sha256:' + hashlib.sha256('notarealthing').hexdigest()
 
-        builder.add_layer(checksum, json.dumps({'id': image.id, 'parent': image.parent_id}))
+        layer_dict = {'id': image.id, 'parent': image.parent_id}
+        if image.config is not None:
+          layer_dict['config'] = image.config
+
+        if image.size is not None:
+          layer_dict['Size'] = image.size
+
+        builder.add_layer(checksum, json.dumps(layer_dict))
 
       # Build the manifest.
       manifests[tag_name] = builder.build(self.jwk)
@@ -110,13 +152,16 @@ class V2Protocol(RegistryProtocol):
       checksum = 'sha256:' + hashlib.sha256(image.bytes).hexdigest()
       checksums[image.id] = checksum
 
-      # Layer data should not yet exist.
-      self.conduct(session, 'HEAD', '/v2/%s/%s/blobs/%s' % (namespace, repo_name, checksum),
-                   expected_status=(404, expected_failure, V2ProtocolSteps.BLOB_HEAD_CHECK),
-                   headers=headers)
+      if not options.skip_head_checks:
+        # Layer data should not yet exist.
+        self.conduct(session, 'HEAD',
+                     '/v2/%s/blobs/%s' % (self.repo_name(namespace, repo_name), checksum),
+                     expected_status=(404, expected_failure, V2ProtocolSteps.BLOB_HEAD_CHECK),
+                     headers=headers)
 
       # Start a new upload of the layer data.
-      response = self.conduct(session, 'POST', '/v2/%s/%s/blobs/uploads/' % (namespace, repo_name),
+      response = self.conduct(session, 'POST',
+                              '/v2/%s/blobs/uploads/' % self.repo_name(namespace, repo_name),
                               expected_status=202,
                               headers=headers)
 
@@ -153,7 +198,8 @@ class V2Protocol(RegistryProtocol):
             return
 
           # Retrieve the upload status at each point, and ensure it is valid.
-          status_url = '/v2/%s/%s/blobs/uploads/%s' % (namespace, repo_name, upload_uuid)
+          status_url = '/v2/%s/blobs/uploads/%s' % (self.repo_name(namespace, repo_name),
+                                                    upload_uuid)
           response = self.conduct(session, 'GET', status_url, expected_status=204, headers=headers)
           assert response.headers['Docker-Upload-UUID'] == upload_uuid
           assert response.headers['Range'] == "bytes=0-%s" % end_byte
@@ -163,7 +209,7 @@ class V2Protocol(RegistryProtocol):
                      headers=headers)
 
         # Ensure the upload was canceled.
-        status_url = '/v2/%s/%s/blobs/uploads/%s' % (namespace, repo_name, upload_uuid)
+        status_url = '/v2/%s/blobs/uploads/%s' % (self.repo_name(namespace, repo_name), upload_uuid)
         self.conduct(session, 'GET', status_url, expected_status=404, headers=headers)
         return
 
@@ -174,14 +220,15 @@ class V2Protocol(RegistryProtocol):
 
       # Ensure the layer exists now.
       response = self.conduct(session, 'HEAD',
-                              '/v2/%s/%s/blobs/%s' % (namespace, repo_name, checksum),
+                              '/v2/%s/blobs/%s' % (self.repo_name(namespace, repo_name), checksum),
                               expected_status=200, headers=headers)
 
       assert response.headers['Docker-Content-Digest'] == checksum
       assert response.headers['Content-Length'] == str(len(image.bytes))
 
       # And retrieve the layer data.
-      result = self.conduct(session, 'GET', '/v2/%s/%s/blobs/%s' % (namespace, repo_name, checksum),
+      result = self.conduct(session, 'GET',
+                            '/v2/%s/blobs/%s' % (self.repo_name(namespace, repo_name), checksum),
                             headers=headers, expected_status=200)
       assert result.content == image.bytes
 
@@ -195,11 +242,42 @@ class V2Protocol(RegistryProtocol):
       manifest_headers = {'Content-Type': 'application/json'}
       manifest_headers.update(headers)
 
-      self.conduct(session, 'PUT', '/v2/%s/%s/manifests/%s' % (namespace, repo_name, tag_name),
-                   data=manifest.bytes, expected_status=put_code,
+      if options.manifest_content_type is not None:
+        manifest_headers['Content-Type'] = options.manifest_content_type
+
+      self.conduct(session, 'PUT',
+                   '/v2/%s/manifests/%s' % (self.repo_name(namespace, repo_name), tag_name),
+                   data=manifest.bytes,
+                   expected_status=(put_code, expected_failure, V2ProtocolSteps.PUT_MANIFEST),
                    headers=manifest_headers)
 
-    return PushResult(checksums=checksums, manifests=manifests)
+    return PushResult(checksums=checksums, manifests=manifests, headers=headers)
+
+
+  def delete(self, session, namespace, repo_name, tag_names, credentials=None,
+             expected_failure=None, options=None):
+    options = options or ProtocolOptions()
+    scopes = options.scopes or ['*']
+    tag_names = [tag_names] if isinstance(tag_names, str) else tag_names
+
+    # Ping!
+    self.ping(session)
+
+    # Perform auth and retrieve a token.
+    token, _ = self.auth(session, credentials, namespace, repo_name, scopes=scopes,
+                         expected_failure=expected_failure)
+    if token is None:
+      return None
+
+    headers = {
+      'Authorization': 'Bearer ' + token,
+    }
+
+    for tag_name in tag_names:
+      self.conduct(session, 'DELETE',
+                   '/v2/%s/manifests/%s' % (self.repo_name(namespace, repo_name), tag_name),
+                   headers=headers,
+                   expected_status=202)
 
 
   def pull(self, session, namespace, repo_name, tag_names, images, credentials=None,
@@ -222,10 +300,12 @@ class V2Protocol(RegistryProtocol):
     }
 
     manifests = {}
+    image_ids = {}
     for tag_name in tag_names:
       # Retrieve the manifest for the tag or digest.
       response = self.conduct(session, 'GET',
-                              '/v2/%s/%s/manifests/%s' % (namespace, repo_name, tag_name),
+                              '/v2/%s/manifests/%s' % (self.repo_name(namespace, repo_name),
+                                                       tag_name),
                               expected_status=(200, expected_failure, V2ProtocolSteps.GET_MANIFEST),
                               headers=headers)
       if expected_failure is not None:
@@ -234,13 +314,58 @@ class V2Protocol(RegistryProtocol):
       # Ensure the manifest returned by us is valid.
       manifest = DockerSchema1Manifest(response.text)
       manifests[tag_name] = manifest
+      image_ids[tag_name] = manifest.leaf_layer.v1_metadata.image_id
 
       # Verify the layers.
       for index, layer in enumerate(manifest.layers):
         result = self.conduct(session, 'GET',
-                              '/v2/%s/%s/blobs/%s' % (namespace, repo_name, layer.digest),
+                              '/v2/%s/blobs/%s' % (self.repo_name(namespace, repo_name),
+                                                   layer.digest),
                               expected_status=200,
                               headers=headers)
         assert result.content == images[index].bytes
 
-    return PullResult(manifests=manifests)
+    return PullResult(manifests=manifests, image_ids=image_ids)
+
+
+  def catalog(self, session, page_size=2, credentials=None, options=None, expected_failure=None,
+              namespace=None, repo_name=None):
+    options = options or ProtocolOptions()
+    scopes = options.scopes or []
+
+    # Ping!
+    self.ping(session)
+
+    # Perform auth and retrieve a token.
+    headers = {}
+    if credentials is not None:
+      token, _ = self.auth(session, credentials, namespace, repo_name, scopes=scopes,
+                           expected_failure=expected_failure)
+      if token is None:
+        return None
+
+      headers = {
+        'Authorization': 'Bearer ' + token,
+      }
+
+    results = []
+    url = '/v2/_catalog'
+    params = {}
+    if page_size is not None:
+      params['n'] = page_size
+
+    while True:
+      response = self.conduct(session, 'GET', url, headers=headers, params=params)
+      data = response.json()
+
+      assert len(data['repositories']) <= page_size
+      results.extend(data['repositories'])
+
+      if not response.headers.get('Link'):
+        return results
+
+      link_url = response.headers['Link']
+      v2_index = link_url.find('/v2/')
+      url = link_url[v2_index:]
+
+    return results
diff --git a/test/registry/protocols.py b/test/registry/protocols.py
index a7b720afa..3b4edba33 100644
--- a/test/registry/protocols.py
+++ b/test/registry/protocols.py
@@ -7,12 +7,14 @@ from cStringIO import StringIO
 from enum import Enum, unique
 from six import add_metaclass
 
-Image = namedtuple('Image', ['id', 'parent_id', 'size', 'bytes'])
-PushResult = namedtuple('PushResult', ['checksums', 'manifests'])
-PullResult = namedtuple('PullResult', ['manifests'])
+Image = namedtuple('Image', ['id', 'parent_id', 'bytes', 'size', 'config'])
+Image.__new__.__defaults__ = (None, None)
+
+PushResult = namedtuple('PushResult', ['checksums', 'manifests', 'headers'])
+PullResult = namedtuple('PullResult', ['manifests', 'image_ids'])
 
 
-def layer_bytes_for_contents(contents):
+def layer_bytes_for_contents(contents, mode='|gz'):
   layer_data = StringIO()
 
   def add_file(name, contents):
@@ -21,7 +23,7 @@ def layer_bytes_for_contents(contents):
     tar_file_info.size = len(contents)
     tar_file_info.mtime = 1
 
-    tar_file = tarfile.open(fileobj=layer_data, mode='w|gz')
+    tar_file = tarfile.open(fileobj=layer_data, mode='w' + mode)
     tar_file.addfile(tar_file_info, StringIO(contents))
     tar_file.close()
 
@@ -38,8 +40,16 @@ class Failures(Enum):
   UNAUTHENTICATED = 'unauthenticated'
   UNAUTHORIZED = 'unauthorized'
   INVALID_REGISTRY = 'invalid-registry'
+  INVALID_REPOSITORY = 'invalid-repository'
   APP_REPOSITORY = 'app-repository'
   UNKNOWN_TAG = 'unknown-tag'
+  ANONYMOUS_NOT_ALLOWED = 'anonymous-not-allowed'
+  DISALLOWED_LIBRARY_NAMESPACE = 'disallowed-library-namespace'
+  MISSING_TAG = 'missing-tag'
+  INVALID_TAG = 'invalid-tag'
+  INVALID_IMAGES = 'invalid-images'
+  UNSUPPORTED_CONTENT_TYPE = 'unsupported-content-type'
+  INVALID_BLOB = 'invalid-blob'
 
 
 class ProtocolOptions(object):
@@ -49,6 +59,8 @@ class ProtocolOptions(object):
     self.cancel_blob_upload = False
     self.manifest_invalid_blob_references = False
     self.chunks_for_upload = None
+    self.skip_head_checks = False
+    self.manifest_content_type = None
 
 
 @add_metaclass(ABCMeta)
@@ -56,6 +68,10 @@ class RegistryProtocol(object):
   """ Interface for protocols. """
   FAILURE_CODES = {}
 
+  @abstractmethod
+  def login(self, session, username, password, scopes, expect_success):
+    """ Performs the login flow with the given credentials, over the given scopes. """
+
   @abstractmethod
   def pull(self, session, namespace, repo_name, tag_names, images, credentials=None,
            expected_failure=None, options=None):
@@ -70,6 +86,12 @@ class RegistryProtocol(object):
         the given credentials.
     """
 
+  def repo_name(self, namespace, repo_name):
+    if namespace:
+      return '%s/%s' % (namespace, repo_name)
+
+    return repo_name
+
   def conduct(self, session, method, url, expected_status=200, params=None, data=None,
               json_data=None, headers=None, auth=None):
     if json_data is not None:
diff --git a/test/registry/registry_tests.py b/test/registry/registry_tests.py
index 69d23c6f5..030528075 100644
--- a/test/registry/registry_tests.py
+++ b/test/registry/registry_tests.py
@@ -1,9 +1,23 @@
+# pylint: disable=W0401, W0621, W0613, W0614, R0913
+import hashlib
+import tarfile
+
+from cStringIO import StringIO
+
+import binascii
+import bencode
+import resumablehashlib
+
 from test.fixtures import *
 from test.registry.liveserverfixture import *
 from test.registry.fixtures import *
 from test.registry.protocol_fixtures import *
 
-from test.registry.protocols import Failures
+from test.registry.protocols import Failures, Image, layer_bytes_for_contents, ProtocolOptions
+
+from app import instance_keys
+from util.security.registry_jwt import decode_bearer_header
+from util.timedeltastring import convert_to_timedelta
 
 
 def test_basic_push_pull(pusher, puller, basic_images, liveserver_session, app_reloader):
@@ -19,6 +33,21 @@ def test_basic_push_pull(pusher, puller, basic_images, liveserver_session, app_r
               credentials=credentials)
 
 
+def test_basic_push_pull_by_manifest(manifest_protocol, basic_images, liveserver_session,
+                                     app_reloader):
+  """ Test: Basic push and pull-by-manifest of an image to a new repository. """
+  credentials = ('devtable', 'password')
+
+  # Push a new repository.
+  result = manifest_protocol.push(liveserver_session, 'devtable', 'newrepo', 'latest', basic_images,
+                                  credentials=credentials)
+
+  # Pull the repository by digests to verify.
+  digests = [str(manifest.digest) for manifest in result.manifests.values()]
+  manifest_protocol.pull(liveserver_session, 'devtable', 'newrepo', digests, basic_images,
+                         credentials=credentials)
+
+
 def test_push_invalid_credentials(pusher, basic_images, liveserver_session, app_reloader):
   """ Test: Ensure we get auth errors when trying to push with invalid credentials. """
   invalid_credentials = ('devtable', 'notcorrectpassword')
@@ -33,3 +62,831 @@ def test_pull_invalid_credentials(puller, basic_images, liveserver_session, app_
 
   puller.pull(liveserver_session, 'devtable', 'newrepo', 'latest', basic_images,
               credentials=invalid_credentials, expected_failure=Failures.UNAUTHENTICATED)
+
+
+def test_push_pull_formerly_bad_repo_name(pusher, puller, basic_images, liveserver_session,
+                                          app_reloader):
+  """ Test: Basic push and pull of an image to a new repository with a name that formerly
+            failed. """
+  credentials = ('devtable', 'password')
+
+  # Push a new repository.
+  pusher.push(liveserver_session, 'devtable', 'foo.bar', 'latest', basic_images,
+              credentials=credentials)
+
+  # Pull the repository to verify.
+  puller.pull(liveserver_session, 'devtable', 'foo.bar', 'latest', basic_images,
+              credentials=credentials)
+
+
+def test_application_repo(pusher, puller, basic_images, liveserver_session, app_reloader,
+                          registry_server_executor, liveserver):
+  """ Test: Attempting to push or pull from an *application* repository raises a 405. """
+  credentials = ('devtable', 'password')
+  registry_server_executor.on(liveserver).create_app_repository('devtable', 'someapprepo')
+
+  # Attempt to push to the repository.
+  pusher.push(liveserver_session, 'devtable', 'someapprepo', 'latest', basic_images,
+              credentials=credentials, expected_failure=Failures.APP_REPOSITORY)
+
+  # Attempt to pull from the repository.
+  puller.pull(liveserver_session, 'devtable', 'someapprepo', 'latest', basic_images,
+              credentials=credentials, expected_failure=Failures.APP_REPOSITORY)
+
+
+def test_middle_layer_different_sha(manifest_protocol, v1_protocol, liveserver_session,
+                                    app_reloader):
+  """ Test: Pushing of a 3-layer image with the *same* V1 ID's, but the middle layer having
+            different bytes, must result in new IDs being generated for the leaf layer, as
+            they point to different "images".
+  """
+  credentials = ('devtable', 'password')
+  first_images = [
+    Image(id='baseimage', parent_id=None, size=None, bytes=layer_bytes_for_contents('base')),
+    Image(id='middleimage', parent_id='baseimage', size=None,
+          bytes=layer_bytes_for_contents('middle')),
+    Image(id='leafimage', parent_id='middleimage', size=None,
+          bytes=layer_bytes_for_contents('leaf')),
+  ]
+
+  # First push and pull the images, to ensure we have the basics setup and working.
+  manifest_protocol.push(liveserver_session, 'devtable', 'newrepo', 'latest', first_images,
+                         credentials=credentials)
+  first_pull_result = manifest_protocol.pull(liveserver_session, 'devtable', 'newrepo', 'latest',
+                                             first_images, credentials=credentials)
+  first_manifest = first_pull_result.manifests['latest']
+  assert set([image.id for image in first_images]) == set(first_manifest.image_ids)
+  assert first_pull_result.image_ids['latest'] == 'leafimage'
+
+  # Next, create an image list with the middle image's *bytes* changed.
+  second_images = list(first_images)
+  second_images[1] = Image(id='middleimage', parent_id='baseimage', size=None,
+                           bytes=layer_bytes_for_contents('different middle bytes'))
+
+  # Push and pull the image, ensuring that the produced ID for the middle and leaf layers
+  # are synthesized.
+  options = ProtocolOptions()
+  options.munge_shas = True
+  options.skip_head_checks = True
+
+  manifest_protocol.push(liveserver_session, 'devtable', 'newrepo', 'latest', second_images,
+                         credentials=credentials, options=options)
+  second_pull_result = v1_protocol.pull(liveserver_session, 'devtable', 'newrepo', 'latest',
+                                        second_images, credentials=credentials, options=options)
+
+  assert second_pull_result.image_ids['latest'] != 'leafimage'
+
+
+def add_robot(api_caller, _):
+  api_caller.conduct_auth('devtable', 'password')
+  resp = api_caller.get('/api/v1/organization/buynlarge/robots/ownerbot')
+  return ('buynlarge+ownerbot', resp.json()['token'])
+
+
+def add_token(_, executor):
+  return ('$token', executor.add_token().text)
+
+
+@pytest.mark.parametrize('credentials, namespace, expected_performer', [
+  (('devtable', 'password'), 'devtable', 'devtable'),
+  (add_robot, 'buynlarge', 'buynlarge+ownerbot'),
+  (('$oauthtoken', 'test'), 'devtable', 'devtable'),
+  (('$app', 'test'), 'devtable', 'devtable'),
+  (add_token, 'devtable', None),
+])
+def test_push_pull_logging(credentials, namespace, expected_performer, pusher, puller, basic_images,
+                           liveserver_session, liveserver, api_caller, app_reloader,
+                           registry_server_executor):
+  """ Test: Basic push and pull, ensuring that logs are added for each operation. """
+
+  # Create the repository before the test push.
+  start_images = [Image(id='startimage', parent_id=None, size=None,
+                        bytes=layer_bytes_for_contents('start image'))]
+  pusher.push(liveserver_session, namespace, 'newrepo', 'latest', start_images,
+              credentials=('devtable', 'password'))
+
+  # Retrieve the credentials to use. This must be after the repo is created, because
+  # some credentials creation code adds the new entity to the repository.
+  if not isinstance(credentials, tuple):
+    credentials = credentials(api_caller, registry_server_executor.on(liveserver))
+
+  # Push to the repository with the specified credentials.
+  pusher.push(liveserver_session, namespace, 'newrepo', 'latest', basic_images,
+              credentials=credentials)
+
+  # Check the logs for the push.
+  api_caller.conduct_auth('devtable', 'password')
+
+  result = api_caller.get('/api/v1/repository/%s/newrepo/logs' % namespace)
+  logs = result.json()['logs']
+  assert len(logs) == 2
+  assert logs[0]['kind'] == 'push_repo'
+  assert logs[0]['metadata']['namespace'] == namespace
+  assert logs[0]['metadata']['repo'] == 'newrepo'
+
+  if expected_performer is not None:
+    assert logs[0]['performer']['name'] == expected_performer
+
+  # Pull the repository to verify.
+  puller.pull(liveserver_session, namespace, 'newrepo', 'latest', basic_images,
+              credentials=credentials)
+
+  # Check the logs for the pull.
+  result = api_caller.get('/api/v1/repository/%s/newrepo/logs' % namespace)
+  logs = result.json()['logs']
+  assert len(logs) == 3
+  assert logs[0]['kind'] == 'pull_repo'
+  assert logs[0]['metadata']['namespace'] == namespace
+  assert logs[0]['metadata']['repo'] == 'newrepo'
+
+  if expected_performer is not None:
+    assert logs[0]['performer']['name'] == expected_performer
+
+
+def test_pull_publicrepo_anonymous(pusher, puller, basic_images, liveserver_session,
+                                   app_reloader, api_caller, liveserver):
+  """ Test: Pull a public repository anonymously. """
+  # Add a new repository under the public user, so we have a repository to pull.
+  pusher.push(liveserver_session, 'public', 'newrepo', 'latest', basic_images,
+              credentials=('public', 'password'))
+
+  # First try to pull the (currently private) repo anonymously, which should fail (since it is
+  # private)
+  puller.pull(liveserver_session, 'public', 'newrepo', 'latest', basic_images,
+              expected_failure=Failures.UNAUTHORIZED)
+
+  # Using a non-public user should also fail.
+  puller.pull(liveserver_session, 'public', 'newrepo', 'latest', basic_images,
+              credentials=('devtable', 'password'),
+              expected_failure=Failures.UNAUTHORIZED)
+
+  # Make the repository public.
+  api_caller.conduct_auth('public', 'password')
+  api_caller.change_repo_visibility('public', 'newrepo', 'public')
+
+  # Pull the repository anonymously, which should succeed because the repository is public.
+  puller.pull(liveserver_session, 'public', 'newrepo', 'latest', basic_images)
+
+
+def test_pull_publicrepo_no_anonymous_access(pusher, puller, basic_images, liveserver_session,
+                                             app_reloader, api_caller, liveserver,
+                                             registry_server_executor):
+  """ Test: Attempts to pull a public repository anonymously, with the feature flag disabled. """
+  # Add a new repository under the public user, so we have a repository to pull.
+  pusher.push(liveserver_session, 'public', 'newrepo', 'latest', basic_images,
+              credentials=('public', 'password'))
+
+  # First try to pull the (currently private) repo anonymously, which should fail (since it is
+  # private)
+  puller.pull(liveserver_session, 'public', 'newrepo', 'latest', basic_images,
+              expected_failure=Failures.UNAUTHORIZED)
+
+  # Using a non-public user should also fail.
+  puller.pull(liveserver_session, 'public', 'newrepo', 'latest', basic_images,
+              credentials=('devtable', 'password'),
+              expected_failure=Failures.UNAUTHORIZED)
+
+  # Make the repository public.
+  api_caller.conduct_auth('public', 'password')
+  api_caller.change_repo_visibility('public', 'newrepo', 'public')
+
+  with FeatureFlagValue('ANONYMOUS_ACCESS', False, registry_server_executor.on(liveserver)):
+    # Attempt again to pull the (now public) repo anonymously, which should fail since
+    # the feature flag for anonymous access is turned off.
+    puller.pull(liveserver_session, 'public', 'newrepo', 'latest', basic_images,
+                expected_failure=Failures.ANONYMOUS_NOT_ALLOWED)
+
+    # Using a non-public user should now succeed.
+    puller.pull(liveserver_session, 'public', 'newrepo', 'latest', basic_images,
+                credentials=('devtable', 'password'))
+
+
+def test_basic_organization_flow(pusher, puller, basic_images, liveserver_session, app_reloader):
+  """ Test: Basic push and pull of an image to a new repository by members of an organization. """
+  # Add a new repository under the organization via the creator user.
+  pusher.push(liveserver_session, 'buynlarge', 'newrepo', 'latest', basic_images,
+              credentials=('creator', 'password'))
+
+  # Ensure that the creator can pull it.
+  puller.pull(liveserver_session, 'buynlarge', 'newrepo', 'latest', basic_images,
+              credentials=('creator', 'password'))
+
+  # Ensure that the admin can pull it.
+  puller.pull(liveserver_session, 'buynlarge', 'newrepo', 'latest', basic_images,
+              credentials=('devtable', 'password'))
+
+  # Ensure that the reader *cannot* pull it.
+  puller.pull(liveserver_session, 'buynlarge', 'newrepo', 'latest', basic_images,
+              credentials=('reader', 'password'),
+              expected_failure=Failures.UNAUTHORIZED)
+
+
+def test_library_support(pusher, puller, basic_images, liveserver_session, app_reloader):
+  """ Test: Pushing and pulling from the implicit library namespace. """
+  credentials = ('devtable', 'password')
+
+  # Push a new repository.
+  pusher.push(liveserver_session, '', 'newrepo', 'latest', basic_images,
+              credentials=credentials)
+
+  # Pull the repository to verify.
+  puller.pull(liveserver_session, '', 'newrepo', 'latest', basic_images,
+              credentials=credentials)
+
+  # Pull the repository from the library namespace to verify.
+  puller.pull(liveserver_session, 'library', 'newrepo', 'latest', basic_images,
+              credentials=credentials)
+
+
+def test_library_namespace_with_support_disabled(pusher, puller, basic_images, liveserver_session,
+                                                 app_reloader, liveserver,
+                                                 registry_server_executor):
+  """ Test: Pushing and pulling from the explicit library namespace, even when the
+            implicit one is disabled.
+  """
+  credentials = ('devtable', 'password')
+
+  with FeatureFlagValue('LIBRARY_SUPPORT', False, registry_server_executor.on(liveserver)):
+    # Push a new repository.
+    pusher.push(liveserver_session, 'library', 'newrepo', 'latest', basic_images,
+                credentials=credentials)
+
+    # Pull the repository from the library namespace to verify.
+    puller.pull(liveserver_session, 'library', 'newrepo', 'latest', basic_images,
+                credentials=credentials)
+
+
+def test_push_library_with_support_disabled(pusher, basic_images, liveserver_session,
+                                            app_reloader, liveserver,
+                                            registry_server_executor):
+  """ Test: Pushing to the implicit library namespace, when disabled,
+            should fail.
+  """
+  credentials = ('devtable', 'password')
+
+  with FeatureFlagValue('LIBRARY_SUPPORT', False, registry_server_executor.on(liveserver)):
+    # Attempt to push a new repository.
+    pusher.push(liveserver_session, '', 'newrepo', 'latest', basic_images,
+                credentials=credentials,
+                expected_failure=Failures.DISALLOWED_LIBRARY_NAMESPACE)
+
+
+def test_pull_library_with_support_disabled(puller, basic_images, liveserver_session,
+                                            app_reloader, liveserver,
+                                            registry_server_executor):
+  """ Test: Pushing to the implicit library namespace, when disabled,
+            should fail.
+  """
+  credentials = ('devtable', 'password')
+
+  with FeatureFlagValue('LIBRARY_SUPPORT', False, registry_server_executor.on(liveserver)):
+    # Attempt to pull the repository from the library namespace.
+    puller.pull(liveserver_session, '', 'newrepo', 'latest', basic_images,
+                credentials=credentials,
+                expected_failure=Failures.DISALLOWED_LIBRARY_NAMESPACE)
+
+
+def test_image_replication(pusher, basic_images, liveserver_session, app_reloader, liveserver,
+                           registry_server_executor):
+  """ Test: Ensure that entries are created for replication of the images pushed. """
+  credentials = ('devtable', 'password')
+
+  with FeatureFlagValue('STORAGE_REPLICATION', True, registry_server_executor.on(liveserver)):
+    pusher.push(liveserver_session, 'devtable', 'newrepo', 'latest', basic_images,
+                credentials=credentials)
+
+    # Ensure that entries were created for each image.
+    for image in basic_images:
+      r = registry_server_executor.on(liveserver).get_storage_replication_entry(image.id)
+      assert r.text == 'OK'
+
+
+def test_push_reponame_with_slashes(pusher, basic_images, liveserver_session, app_reloader):
+  """ Test: Attempt to add a repository name with slashes. This should fail as we do not
+      support it.
+  """
+  credentials = ('devtable', 'password')
+
+  pusher.push(liveserver_session, 'devtable', 'some/slash', 'latest', basic_images,
+              credentials=credentials,
+              expected_failure=Failures.INVALID_REPOSITORY)
+
+
+@pytest.mark.parametrize('tag_name, expected_failure', [
+  ('l', None),
+  ('1', None),
+  ('x' * 128, None),
+
+  ('', Failures.MISSING_TAG),
+  ('x' * 129, Failures.INVALID_TAG),
+  ('.fail', Failures.INVALID_TAG),
+  ('-fail', Failures.INVALID_TAG),
+])
+def test_tag_validaton(tag_name, expected_failure, pusher, basic_images, liveserver_session,
+                       app_reloader):
+  """ Test: Various forms of tags and whether they succeed or fail as expected. """
+  credentials = ('devtable', 'password')
+
+  pusher.push(liveserver_session, 'devtable', 'newrepo', tag_name, basic_images,
+              credentials=credentials,
+              expected_failure=expected_failure)
+
+
+def test_invalid_parent(pusher, liveserver_session, app_reloader):
+  """ Test: Attempt to push an image with an invalid/missing parent. """
+  images = [
+    Image(id='childimage', parent_id='parentimage', size=None,
+          bytes=layer_bytes_for_contents('child')),
+  ]
+
+  credentials = ('devtable', 'password')
+
+  pusher.push(liveserver_session, 'devtable', 'newrepo', 'latest', images,
+              credentials=credentials,
+              expected_failure=Failures.INVALID_IMAGES)
+
+
+def test_wrong_image_order(pusher, liveserver_session, app_reloader):
+  """ Test: Attempt to push an image with its layers in the wrong order. """
+  images = [
+    Image(id='childimage', parent_id='parentimage', size=None,
+          bytes=layer_bytes_for_contents('child')),
+    Image(id='parentimage', parent_id=None, size=None,
+          bytes=layer_bytes_for_contents('parent')),
+  ]
+
+  credentials = ('devtable', 'password')
+
+  pusher.push(liveserver_session, 'devtable', 'newrepo', 'latest', images,
+              credentials=credentials,
+              expected_failure=Failures.INVALID_IMAGES)
+
+
+@pytest.mark.parametrize('labels', [
+  # Basic labels.
+  [('foo', 'bar', 'text/plain'), ('baz', 'meh', 'text/plain')],
+
+  # Theoretically invalid, but allowed when pushed via registry protocol.
+  [('theoretically-invalid--label', 'foo', 'text/plain')],
+
+  # JSON label.
+  [('somejson', '{"some": "json"}', 'application/json'), ('plain', '', 'text/plain')],
+
+  # JSON-esque (but not valid JSON) labels.
+  [('foo', '[hello world]', 'text/plain'), ('bar', '{wassup?!}', 'text/plain')],
+])
+def test_labels(labels, manifest_protocol, liveserver_session, api_caller, app_reloader):
+  """ Test: Image pushed with labels has those labels found in the database after the
+      push succeeds.
+  """
+  images = [
+    Image(id='theimage', parent_id=None, bytes=layer_bytes_for_contents('image'),
+          config={'Labels': {key: value for (key, value, _) in labels}}),
+  ]
+
+  credentials = ('devtable', 'password')
+  result = manifest_protocol.push(liveserver_session, 'devtable', 'newrepo', 'latest', images,
+                                  credentials=credentials)
+
+  digest = result.manifests['latest'].digest
+  api_caller.conduct_auth('devtable', 'password')
+
+  data = api_caller.get('/api/v1/repository/devtable/newrepo/manifest/%s/labels' % digest).json()
+  labels_found = data['labels']
+  assert len(labels_found) == len(labels)
+
+  labels_found_map = {l['key']: l for l in labels_found}
+  assert set(images[0].config['Labels'].keys()) == set(labels_found_map.keys())
+
+  for key, _, media_type in labels:
+    assert labels_found_map[key]['source_type'] == 'manifest'
+    assert labels_found_map[key]['media_type'] == media_type
+
+
+@pytest.mark.parametrize('label_value, expected_expiration', [
+  ('1d', True),
+  ('1h', True),
+  ('2w', True),
+
+  ('1g', False),
+  ('something', False),
+])
+def test_expiration_label(label_value, expected_expiration, manifest_protocol, liveserver_session,
+                          api_caller, app_reloader):
+  """ Test: Tag pushed with a valid `quay.expires-after` will have its expiration set to its
+            start time plus the duration specified. If the duration is invalid, no expiration will
+            be set.
+  """
+  images = [
+    Image(id='theimage', parent_id=None, bytes=layer_bytes_for_contents('image'),
+          config={'Labels': {'quay.expires-after': label_value}}),
+  ]
+
+  credentials = ('devtable', 'password')
+  manifest_protocol.push(liveserver_session, 'devtable', 'newrepo', 'latest', images,
+                         credentials=credentials)
+
+  api_caller.conduct_auth('devtable', 'password')
+
+  tag_data = api_caller.get('/api/v1/repository/devtable/newrepo/tag').json()['tags'][0]
+  if expected_expiration:
+    diff = convert_to_timedelta(label_value).total_seconds()
+    assert tag_data['end_ts'] == tag_data['start_ts'] + diff
+  else:
+    assert tag_data.get('end_ts') is None
+
+
+@pytest.mark.parametrize('content_type', [
+  'application/vnd.oci.image.manifest.v1+json',
+  'application/vnd.docker.distribution.manifest.v2+json',
+  'application/vnd.foo.bar',
+])
+def test_unsupported_manifest_content_type(content_type, manifest_protocol, basic_images,
+                                           liveserver_session, app_reloader):
+  """ Test: Attempt to push a manifest with an unsupported media type. """
+  credentials = ('devtable', 'password')
+
+  options = ProtocolOptions()
+  options.manifest_content_type = content_type
+
+  # Attempt to push a new repository.
+  manifest_protocol.push(liveserver_session, 'devtable', 'newrepo', 'latest', basic_images,
+                         credentials=credentials,
+                         options=options,
+                         expected_failure=Failures.UNSUPPORTED_CONTENT_TYPE)
+
+
+def test_invalid_blob_reference(manifest_protocol, basic_images, liveserver_session, app_reloader):
+  """ Test: Attempt to push a manifest with an invalid blob reference. """
+  credentials = ('devtable', 'password')
+
+  options = ProtocolOptions()
+  options.manifest_invalid_blob_references = True
+
+  # Attempt to push a new repository.
+  manifest_protocol.push(liveserver_session, 'devtable', 'newrepo', 'latest', basic_images,
+                         credentials=credentials,
+                         options=options,
+                         expected_failure=Failures.INVALID_BLOB)
+
+
+def test_delete_tag(manifest_protocol, puller, basic_images, liveserver_session,
+                    app_reloader):
+  """ Test: Push a repository, delete a tag, and attempt to pull. """
+  credentials = ('devtable', 'password')
+
+  # Push the tags.
+  result = manifest_protocol.push(liveserver_session, 'devtable', 'newrepo', ['one', 'two'],
+                                  basic_images, credentials=credentials)
+
+  # Delete tag `one` by digest.
+  manifest_protocol.delete(liveserver_session, 'devtable', 'newrepo',
+                           result.manifests['one'].digest,
+                           credentials=credentials)
+
+  # Attempt to pull tag `one` and ensure it doesn't work.
+  puller.pull(liveserver_session, 'devtable', 'newrepo', 'one', basic_images,
+              credentials=credentials,
+              expected_failure=Failures.UNKNOWN_TAG)
+
+  # Pull tag `two` to verify it works.
+  puller.pull(liveserver_session, 'devtable', 'newrepo', 'two', basic_images,
+              credentials=credentials)
+
+
+def test_cancel_upload(manifest_protocol, basic_images, liveserver_session, app_reloader):
+  """ Test: Cancelation of blob uploads. """
+  credentials = ('devtable', 'password')
+
+  options = ProtocolOptions()
+  options.cancel_blob_upload = True
+
+  manifest_protocol.push(liveserver_session, 'devtable', 'newrepo', 'latest',
+                         basic_images, credentials=credentials)
+
+
+def test_blob_caching(manifest_protocol, basic_images, liveserver_session, app_reloader,
+                      liveserver, registry_server_executor):
+  """ Test: Pulling of blobs after initially pulled will result in the blobs being cached. """
+  credentials = ('devtable', 'password')
+
+  # Push a tag.
+  result = manifest_protocol.push(liveserver_session, 'devtable', 'newrepo', 'latest',
+                                  basic_images, credentials=credentials)
+
+
+  # Conduct the initial pull to prime the cache.
+  manifest_protocol.pull(liveserver_session, 'devtable', 'newrepo', 'latest', basic_images,
+                         credentials=credentials)
+
+  # Disconnect the server from the database.
+  registry_server_executor.on(liveserver).break_database()
+
+  # Pull each blob, which should succeed due to caching. If caching is broken, this will
+  # fail when it attempts to hit the database.
+  for layer in result.manifests['latest'].layers:
+    blob_id = str(layer.digest)
+    r = liveserver_session.get('/v2/devtable/newrepo/blobs/%s' % blob_id, headers=result.headers)
+    assert r.status_code == 200
+
+
+@pytest.mark.parametrize('chunks', [
+  # Two chunks.
+  [(0, 100), (100, None)],
+
+  # Multiple chunks.
+  [(0, 10), (10, 20), (20, None)],
+  [(0, 10), (10, 20), (20, 30), (30, 40), (40, 50), (50, None)],
+
+  # Overlapping chunks.
+  [(0, 1024), (10, None)],
+])
+def test_chunked_blob_uploading(chunks, random_layer_data, manifest_protocol, puller,
+                                liveserver_session, app_reloader):
+  """ Test: Uploading of blobs as chunks. """
+  credentials = ('devtable', 'password')
+
+  adjusted_chunks = []
+  for start_byte, end_byte in chunks:
+    adjusted_chunks.append((start_byte, end_byte if end_byte else len(random_layer_data)))
+
+  images = [
+    Image(id='theimage', parent_id=None, bytes=random_layer_data),
+  ]
+
+  options = ProtocolOptions()
+  options.chunks_for_upload = adjusted_chunks
+
+  # Push the image, using the specified chunking.
+  manifest_protocol.push(liveserver_session, 'devtable', 'newrepo', 'latest',
+                         images, credentials=credentials, options=options)
+
+  # Pull to verify the image was created.
+  puller.pull(liveserver_session, 'devtable', 'newrepo', 'latest', images,
+              credentials=credentials)
+
+
+def test_chunked_uploading_mismatched_chunks(manifest_protocol, random_layer_data,
+                                             liveserver_session, app_reloader):
+  """ Test: Attempt to upload chunks with data missing. """
+  credentials = ('devtable', 'password')
+
+  images = [
+    Image(id='theimage', parent_id=None, bytes=random_layer_data),
+  ]
+
+  # Note: Byte #100 is missing.
+  options = ProtocolOptions()
+  options.chunks_for_upload = [(0, 100), (101, len(random_layer_data), 416)]
+
+  # Attempt to push, with the chunked upload failing.
+  manifest_protocol.push(liveserver_session, 'devtable', 'newrepo', 'latest',
+                         images, credentials=credentials, options=options)
+
+
+@pytest.mark.parametrize('public_catalog, credentials, expected_repos', [
+  # No public access and no credentials => No results.
+  (False, None, None),
+
+  # Public access and no credentials => public repositories.
+  (True, None, ['public/publicrepo']),
+
+  # Private creds => private repositories.
+  (False, ('devtable', 'password'), ['devtable/simple', 'devtable/complex', 'devtable/gargantuan']),
+  (True, ('devtable', 'password'), ['devtable/simple', 'devtable/complex', 'devtable/gargantuan']),
+])
+@pytest.mark.parametrize('page_size', [
+  1,
+  2,
+  10,
+  50,
+  100,
+])
+def test_catalog(public_catalog, credentials, expected_repos, page_size, v2_protocol,
+                 liveserver_session, app_reloader, liveserver, registry_server_executor):
+  """ Test: Retrieving results from the V2 catalog. """
+  with FeatureFlagValue('PUBLIC_CATALOG', public_catalog, registry_server_executor.on(liveserver)):
+    results = v2_protocol.catalog(liveserver_session, page_size=page_size, credentials=credentials,
+                                  namespace='devtable', repo_name='simple')
+
+  if expected_repos is None:
+    assert len(results) == 0
+  else:
+    assert set(expected_repos).issubset(set(results))
+
+
+def test_pull_torrent(pusher, basic_images, liveserver_session, liveserver,
+                      registry_server_executor, app_reloader):
+  """ Test: Retrieve a torrent for pulling the image via the Quay CLI. """
+  credentials = ('devtable', 'password')
+
+  # Push an image to download.
+  pusher.push(liveserver_session, 'devtable', 'newrepo', 'latest', basic_images,
+              credentials=credentials)
+
+  # Required for torrent.
+  registry_server_executor.on(liveserver).set_supports_direct_download(True)
+
+  # For each layer, retrieve a torrent for the blob.
+  for image in basic_images:
+    digest = 'sha256:' + hashlib.sha256(image.bytes).hexdigest()
+    response = liveserver_session.get('/c1/torrent/devtable/newrepo/blobs/%s' % digest,
+                                      auth=credentials)
+    torrent_info = bencode.bdecode(response.content)
+
+    # Check the announce URL.
+    assert torrent_info['url-list'] == 'http://somefakeurl?goes=here'
+
+    # Check the metadata.
+    assert torrent_info.get('info', {}).get('pieces') is not None
+    assert torrent_info.get('announce') is not None
+
+    # Check the pieces.
+    sha = resumablehashlib.sha1()
+    sha.update(image.bytes)
+
+    expected = binascii.hexlify(sha.digest())
+    found = binascii.hexlify(torrent_info['info']['pieces'])
+    assert expected == found
+
+
+@pytest.mark.parametrize('use_estimates', [
+  False,
+  True,
+])
+def test_squashed_images(use_estimates, pusher, sized_images, liveserver_session,
+                         liveserver, registry_server_executor, app_reloader):
+  """ Test: Pulling of squashed images. """
+  credentials = ('devtable', 'password')
+
+  # Push an image to download.
+  pusher.push(liveserver_session, 'devtable', 'newrepo', 'latest', sized_images,
+              credentials=credentials)
+
+  if use_estimates:
+    # Clear the uncompressed size stored for the images, to ensure that we estimate instead.
+    for image in sized_images:
+      registry_server_executor.on(liveserver).clear_uncompressed_size(image.id)
+
+  # Pull the squashed version.
+  response = liveserver_session.get('/c1/squash/devtable/newrepo/latest', auth=credentials)
+  tar = tarfile.open(fileobj=StringIO(response.content))
+
+  # Verify the squashed image.
+  expected_image_id = '13a2d9711a3e242fcd50a7627c02d86901ac801b78dcea3147e8ff640078de52'
+  expected_names = ['repositories',
+                    expected_image_id,
+                    '%s/json' % expected_image_id,
+                    '%s/VERSION' % expected_image_id,
+                    '%s/layer.tar' % expected_image_id]
+
+  assert tar.getnames() == expected_names
+
+  # Verify the JSON image data.
+  json_data = (tar.extractfile(tar.getmember('%s/json' % expected_image_id)).read())
+
+  # Ensure the JSON loads and parses.
+  result = json.loads(json_data)
+  assert result['id'] == expected_image_id
+
+  # Ensure that squashed layer tar can be opened.
+  tarfile.open(fileobj=tar.extractfile(tar.getmember('%s/layer.tar' % expected_image_id)))
+
+
+def get_robot_password(api_caller):
+  api_caller.conduct_auth('devtable', 'password')
+  resp = api_caller.get('/api/v1/organization/buynlarge/robots/ownerbot')
+  return resp.json()['token']
+
+
+def get_encrypted_password(api_caller):
+  api_caller.conduct_auth('devtable', 'password')
+  resp = api_caller.post('/api/v1/user/clientkey', data=json.dumps(dict(password='password')),
+                         headers={'Content-Type': 'application/json'})
+  return resp.json()['key']
+
+
+@pytest.mark.parametrize('username, password, expect_success', [
+  # Invalid username.
+  ('invaliduser', 'somepassword', False),
+
+  # Invalid password.
+  ('devtable', 'invalidpassword', False),
+
+  # Invalid OAuth token.
+  ('$oauthtoken', 'unknown', False),
+
+  # Invalid CLI token.
+  ('$app', 'invalid', False),
+
+  # Valid.
+  ('devtable', 'password', True),
+
+  # Robot.
+  ('buynlarge+ownerbot', get_robot_password, True),
+
+  # Encrypted password.
+  ('devtable', get_encrypted_password, True),
+
+  # OAuth.
+  ('$oauthtoken', 'test', True),
+
+  # CLI Token.
+  ('$app', 'test', True),
+])
+def test_login(username, password, expect_success, loginer, liveserver_session,
+               api_caller, app_reloader):
+  """ Test: Login flow. """
+  if not isinstance(password, str):
+    password = password(api_caller)
+
+  loginer.login(liveserver_session, username, password, [], expect_success)
+
+
+@pytest.mark.parametrize('username, password, scopes, expected_access, expect_success', [
+  # No scopes.
+  ('devtable', 'password', [], [], True),
+
+  # Basic pull.
+  ('devtable', 'password', ['repository:devtable/simple:pull'], [
+    {'type': 'repository', 'name': 'devtable/simple', 'actions': ['pull']},
+  ], True),
+
+  # Basic push.
+  ('devtable', 'password', ['repository:devtable/simple:push'], [
+    {'type': 'repository', 'name': 'devtable/simple', 'actions': ['push']},
+  ], True),
+
+  # Basic push/pull.
+  ('devtable', 'password', ['repository:devtable/simple:push,pull'], [
+    {'type': 'repository', 'name': 'devtable/simple', 'actions': ['push', 'pull']},
+  ], True),
+
+  # Admin.
+  ('devtable', 'password', ['repository:devtable/simple:push,pull,*'], [
+    {'type': 'repository', 'name': 'devtable/simple', 'actions': ['push', 'pull', '*']},
+  ], True),
+
+  # Basic pull with endpoint.
+  ('devtable', 'password', ['repository:localhost:5000/devtable/simple:pull'], [
+    {'type': 'repository', 'name': 'localhost:5000/devtable/simple', 'actions': ['pull']},
+  ], True),
+
+  # Basic pull with invalid endpoint.
+  ('devtable', 'password', ['repository:someinvalid/devtable/simple:pull'], [], False),
+
+  # Pull with no access.
+  ('public', 'password', ['repository:devtable/simple:pull'], [
+    {'type': 'repository', 'name': 'devtable/simple', 'actions': []},
+  ], True),
+
+  # Anonymous push and pull on a private repository.
+  ('', '', ['repository:devtable/simple:pull,push'], [
+    {'type': 'repository', 'name': 'devtable/simple', 'actions': []},
+  ], True),
+
+  # Pull and push with no push access.
+  ('reader', 'password', ['repository:buynlarge/orgrepo:pull,push'], [
+    {'type': 'repository', 'name': 'buynlarge/orgrepo', 'actions': ['pull']},
+  ], True),
+
+  # OAuth.
+  ('$oauthtoken', 'test', ['repository:public/publicrepo:pull,push'], [
+    {'type': 'repository', 'name': 'public/publicrepo', 'actions': ['pull']},
+  ], True),
+
+  # Anonymous public repo.
+  ('', '', ['repository:public/publicrepo:pull,push'], [
+    {'type': 'repository', 'name': 'public/publicrepo', 'actions': ['pull']},
+  ], True),
+
+  # Multiple scopes.
+  ('devtable', 'password',
+   ['repository:devtable/simple:push,pull,*', 'repository:buynlarge/orgrepo:pull'], [
+     {'type': 'repository', 'name': 'devtable/simple', 'actions': ['push', 'pull', '*']},
+     {'type': 'repository', 'name': 'buynlarge/orgrepo', 'actions': ['pull']},
+   ], True),
+
+  # Multiple scopes.
+  ('devtable', 'password',
+   ['repository:devtable/simple:push,pull,*', 'repository:public/publicrepo:push,pull'], [
+     {'type': 'repository', 'name': 'devtable/simple', 'actions': ['push', 'pull', '*']},
+     {'type': 'repository', 'name': 'public/publicrepo', 'actions': ['pull']},
+   ], True),
+])
+def test_login_scopes(username, password, scopes, expected_access, expect_success, v2_protocol,
+                      liveserver_session, api_caller, app_reloader):
+  """ Test: Login via the V2 auth protocol reacts correctly to requested scopes. """
+  if not isinstance(password, str):
+    password = password(api_caller)
+
+  response = v2_protocol.login(liveserver_session, username, password, scopes, expect_success)
+  if not expect_success:
+    return
+
+  # Validate the returned token.
+  encoded = response.json()['token']
+  payload = decode_bearer_header('Bearer ' + encoded, instance_keys,
+                                 {'SERVER_HOSTNAME': 'localhost:5000'})
+  assert payload is not None
+  assert payload['access'] == expected_access