Add unit tests and a stronger restriction on the revert API call

data/model/legacy.py

@@ -2829,12 +2829,20 @@ def repository_is_starred(user, repository):
     return False
 
 
-def revert_tag(namespace_name, repository_name, tag_name, docker_image_id):
+def revert_tag(repository, tag_name, docker_image_id):
   """ Reverts a tag to a specific image ID. """
-  image = get_image_by_id(namespace_name, repository_name, docker_image_id)
+  # Verify that the image ID already existed under this repository under the
-  if image is None:
+  # tag.
-    raise DataModelException('Cannot revert to unknown image')
+  try:
+    (RepositoryTag.select()
+                  .join(Image)
+                  .where(RepositoryTag.repository == repository)
+                  .where(RepositoryTag.name == tag_name)
+                  .where(Image.docker_image_id == docker_image_id)
+                  .get())
+  except RepositoryTag.DoesNotExist:
+    raise DataModelException('Cannot revert to unknown or invalid image')
 
-  return create_or_update_tag(namespace_name, repository_name, tag_name, docker_image_id,
+  return create_or_update_tag(repository.namespace_user.username, repository.name,
-                              reversion=True)
+                              tag_name, docker_image_id, reversion=True)
 

endpoints/api/tag.py

@@ -177,7 +177,7 @@ class RevertTag(RepositoryParamResource):
 
     # Revert the tag back to the previous image.
     image_id = request.get_json()['image']
-    model.revert_tag(namespace, repository, tag, image_id)
+    model.revert_tag(tag_image.repository, tag, image_id)
     model.garbage_collect_repository(namespace, repository)
 
     # Log the reversion.

test/test_api_usage.py

@@ -14,7 +14,7 @@ from initdb import setup_database_for_testing, finished_database_for_testing
 from data import model, database
 
 from endpoints.api.team import TeamMember, TeamMemberList, TeamMemberInvite, OrganizationTeam
-from endpoints.api.tag import RepositoryTagImages, RepositoryTag
+from endpoints.api.tag import RepositoryTagImages, RepositoryTag, RevertTag, ListRepositoryTags
 from endpoints.api.search import FindRepositories, EntitySearch, ConductSearch
 from endpoints.api.image import RepositoryImage, RepositoryImageList
 from endpoints.api.build import (RepositoryBuildStatus, RepositoryBuildLogs, RepositoryBuildList,
@@ -1746,6 +1746,45 @@ class TestGetImageChanges(ApiTestCase):
     #                                 image_id=image_id))
 
 
+class TestRevertTag(ApiTestCase):
+  def test_reverttag_invalidtag(self):
+    self.login(ADMIN_ACCESS_USER)
+
+    self.postResponse(RevertTag,
+                      params=dict(repository=ADMIN_ACCESS_USER + '/history', tag='invalidtag'),
+                      data=dict(image='invalid_image'),
+                      expected_code=404)
+
+  def test_reverttag_invalidimage(self):
+    self.login(ADMIN_ACCESS_USER)
+
+    self.postResponse(RevertTag,
+                      params=dict(repository=ADMIN_ACCESS_USER + '/history', tag='latest'),
+                      data=dict(image='invalid_image'),
+                      expected_code=400)
+
+  def test_reverttag(self):
+    self.login(ADMIN_ACCESS_USER)
+
+    json = self.getJsonResponse(ListRepositoryTags,
+                                params=dict(repository=ADMIN_ACCESS_USER + '/history', tag='latest'))
+
+    self.assertEquals(2, len(json['tags']))
+    self.assertFalse('end_ts' in json['tags'][0])
+
+    previous_image_id = json['tags'][1]['docker_image_id']
+
+    self.postJsonResponse(RevertTag,
+                      params=dict(repository=ADMIN_ACCESS_USER + '/history', tag='latest'),
+                      data=dict(image=previous_image_id))
+
+    json = self.getJsonResponse(ListRepositoryTags,
+                                params=dict(repository=ADMIN_ACCESS_USER + '/history', tag='latest'))
+    self.assertEquals(3, len(json['tags']))
+    self.assertFalse('end_ts' in json['tags'][0])
+    self.assertEquals(previous_image_id, json['tags'][0]['docker_image_id'])
+
+
 class TestListAndDeleteTag(ApiTestCase):
   def test_listdeletecreateandmovetag(self):
     self.login(ADMIN_ACCESS_USER)