Add squash testing code to registry tests

Fixes #896

storage/fakestorage.py

@@ -33,7 +33,7 @@ class FakeStorage(BaseStorageV2):
       yield buf
 
   def stream_read_file(self, path):
-    return StringIO(_FAKE_STORAGE_MAP[path])
+    return StringIO.StringIO(self.get_content(path))
 
   def stream_write(self, path, fp, content_type=None, content_encoding=None):
     out_fp = _FAKE_STORAGE_MAP[path]

test/registry_tests.py

@@ -14,6 +14,7 @@ from app import app
 from data.database import close_db_filter, configure
 from endpoints.v1 import v1_bp
 from endpoints.v2 import v2_bp
+from endpoints.verbs import verbs
 from endpoints.v2.manifest import SignedManifestBuilder
 from endpoints.api import api_bp
 from initdb import wipe_database, initialize_database, populate_database
@@ -40,6 +41,7 @@ from digest.checksums import compute_simple
 try:
   app.register_blueprint(v1_bp, url_prefix='/v1')
   app.register_blueprint(v2_bp, url_prefix='/v2')
+  app.register_blueprint(verbs, url_prefix='/c1')
   app.register_blueprint(api_bp, url_prefix='/api')
 except ValueError:
   # Blueprint was already registered
@@ -920,5 +922,52 @@ class V1PullV2PushRegistryTests(V1RegistryPullMixin, V2RegistryPushMixin, Regist
                                 RegistryTestCaseMixin, LiveServerTestCase):
   """ Tests for V1 pull, V2 push registry. """
 
+
+class VerbTests(RegistryTestCaseMixin, V1RegistryPushMixin, LiveServerTestCase):
+  """ Tests for registry verbs. """
+
+  def test_multilayer_squashing(self):
+    images = [
+      {
+        'id': 'latestid',
+        'contents': 'the latest image',
+        'parent': 'baseid',
+      },
+      {
+        'id': 'baseid',
+        'contents': 'The base image',
+      }
+    ]
+
+    # Create the repo.
+    self.do_push('devtable', 'newrepo', 'devtable', 'password', images=images)
+
+    # Pull the squashed version of the tag.
+    response = self.conduct('GET', '/c1/squash/devtable/newrepo/latest', auth='sig')
+    tar = tarfile.open(fileobj=StringIO(response.content))
+
+    expected_image_id = 'bd590ae79fba5ebc6550aaf016c0bd0f49b1d78178e0f83e0ca1c56c2bb7e7bf'
+
+    expected_names = ['repositories',
+                      expected_image_id,
+                      '%s/json' % expected_image_id,
+                      '%s/VERSION' % expected_image_id,
+                      '%s/layer.tar' % expected_image_id]
+
+    self.assertEquals(expected_names, tar.getnames())
+    self.assertEquals('1.0', tar.extractfile(tar.getmember('%s/VERSION' % expected_image_id)).read())
+
+    json_data = (tar.extractfile(tar.getmember('%s/json' % expected_image_id)).read())
+
+    # Ensure the JSON loads and parses.
+    result = json.loads(json_data)
+    self.assertEquals(expected_image_id, result['id'])
+
+    # Ensure that the "image_name" file refers to the latest image, as it is the top layer.
+    layer_tar = tarfile.open(fileobj=tar.extractfile(tar.getmember('%s/layer.tar' % expected_image_id)))
+    image_name = layer_tar.extractfile(layer_tar.getmember('image_name')).read()
+    self.assertEquals('latestid', image_name)
+
+
 if __name__ == '__main__':
   unittest.main()