Add tests for image model, fix a small bug and remove an unused function

data/model/image.py

@@ -24,7 +24,7 @@ def get_image_with_storage(docker_image_id, storage_uuid):
   """
   try:
     return (Image
-            .select()
+            .select(Image, ImageStorage)
             .join(ImageStorage)
             .where(Image.docker_image_id == docker_image_id,
                    ImageStorage.uuid == storage_uuid)
@@ -355,25 +355,6 @@ def get_image(repo, docker_image_id):
     return None
 
 
-def get_repo_image_by_storage_checksum(namespace, repository_name, storage_checksum):
-  try:
-    return (Image
-            .select()
-            .join(ImageStorage)
-            .switch(Image)
-            .join(Repository)
-            .join(Namespace, on=(Namespace.id == Repository.namespace_user))
-            .where(Repository.name == repository_name, Namespace.username == namespace,
-                   ImageStorage.content_checksum == storage_checksum,
-                   ImageStorage.uploading == False)
-            .get())
-  except Image.DoesNotExist:
-    msg = 'Image with storage checksum {0} does not exist in repo {1}/{2}'.format(storage_checksum,
-                                                                                  namespace,
-                                                                                  repository_name)
-    raise InvalidImageException(msg)
-
-
 def synthesize_v1_image(repo, image_storage_id, storage_image_size, docker_image_id,
                         created_date_str, comment, command, v1_json_metadata, parent_image=None):
   """ Find an existing image with this docker image id, and if none exists, write one with the

data/model/test/test_image.py

@@ -0,0 +1,104 @@
+import pytest
+
+from collections import defaultdict
+from data.model import image, repository
+from playhouse.test_utils import assert_query_count
+
+from test.fixtures import *
+
+@pytest.fixture()
+def images(initialized_db):
+  images = image.get_repository_images('devtable', 'simple')
+  assert len(images)
+  return images
+
+
+def test_get_image_with_storage(images, initialized_db):
+  for current in images:
+    storage_uuid = current.storage.uuid
+
+    with assert_query_count(1):
+      retrieved = image.get_image_with_storage(current.docker_image_id, storage_uuid)
+      assert retrieved.id == current.id
+      assert retrieved.storage.uuid == storage_uuid
+
+
+def test_get_parent_images(images, initialized_db):
+  for current in images:
+    if not len(current.ancestor_id_list()):
+      continue
+
+    with assert_query_count(1):
+      parent_images = list(image.get_parent_images('devtable', 'simple', current))
+
+    assert len(parent_images) == len(current.ancestor_id_list())
+    assert set(current.ancestor_id_list()) == {i.id for i in parent_images}
+
+    for parent in parent_images:
+      with assert_query_count(0):
+        assert parent.storage.id
+
+
+def test_get_image(images, initialized_db):
+  for current in images:
+    repo = current.repository
+
+    with assert_query_count(1):
+      found = image.get_image(repo, current.docker_image_id)
+
+    assert found.id == current.id
+
+
+def test_placements(images, initialized_db):
+  with assert_query_count(1):
+    placements_map = image.get_placements_for_images(images)
+
+  for current in images:
+    assert current.storage.id in placements_map
+
+    with assert_query_count(2):
+      expected_image, expected_placements = image.get_image_and_placements('devtable', 'simple',
+                                                                           current.docker_image_id)
+
+    assert expected_image.id == current.id
+    assert len(expected_placements) == len(placements_map.get(current.storage.id))
+    assert ({p.id for p in expected_placements} == 
+            {p.id for p in placements_map.get(current.storage.id)})
+
+
+def test_get_repo_image(images, initialized_db):
+  for current in images:
+    with assert_query_count(1):
+      found = image.get_repo_image('devtable', 'simple', current.docker_image_id)
+
+    assert found.id == current.id
+    with assert_query_count(1):
+      assert found.storage.id
+
+
+def test_get_repo_image_and_storage(images, initialized_db):
+  for current in images:
+    with assert_query_count(1):
+      found = image.get_repo_image_and_storage('devtable', 'simple', current.docker_image_id)
+
+    assert found.id == current.id
+    with assert_query_count(0):
+      assert found.storage.id
+
+
+def test_get_repository_images_without_placements(images, initialized_db):
+  ancestors_map = defaultdict(list)
+  for img in images:
+    current = img.parent
+    while current is not None:
+      ancestors_map[current.id].append(img.id)
+      current = current.parent
+
+  for current in images:
+    repo = current.repository
+
+    with assert_query_count(1):
+      found = list(image.get_repository_images_without_placements(repo, with_ancestor=current))
+
+    assert len(found) == len(ancestors_map[current.id]) + 1
+    assert {i.id for i in found} == set(ancestors_map[current.id] + [current.id])