Merge pull request #3160 from quay/move-buildquery-tests

Move build query tests to pytest

data/model/test/test_build.py

@@ -2,8 +2,9 @@ import pytest
 
 from mock import patch
 
-from data.database import BUILD_PHASE, RepositoryBuildTrigger
+from data.database import BUILD_PHASE, RepositoryBuildTrigger, RepositoryBuild
-from data.model.build import update_trigger_disable_status
+from data.model.build import (update_trigger_disable_status, create_repository_build,
+                              get_repository_build, update_phase_then_close)
 from test.fixtures import *
 
 TEST_FAIL_THRESHOLD = 5
@@ -43,4 +44,64 @@ def test_update_trigger_disable_status(starting_failure_count, starting_error_co
       assert updated_trigger.disabled_reason is None
       assert updated_trigger.successive_failure_count == 0
       assert updated_trigger.successive_internal_error_count == 0
-  
+
+
+def test_archivable_build_logs(initialized_db):
+  # Make sure there are no archivable logs.
+  result = model.build.get_archivable_build()
+  assert result is None
+
+  # Add a build that cannot (yet) be archived.
+  repo = model.repository.get_repository('devtable', 'simple')
+  token = model.token.create_access_token(repo, 'write')
+  created = RepositoryBuild.create(repository=repo, access_token=token,
+                                   phase=model.build.BUILD_PHASE.WAITING,
+                                   logs_archived=False, job_config='{}',
+                                   display_name='')
+
+  # Make sure there are no archivable logs.
+  result = model.build.get_archivable_build()
+  assert result is None
+
+  # Change the build to being complete.
+  created.phase = model.build.BUILD_PHASE.COMPLETE
+  created.save()
+
+  # Make sure we now find an archivable build.
+  result = model.build.get_archivable_build()
+  assert result.id == created.id
+
+
+def test_update_build_phase(initialized_db):
+  build = create_build(model.repository.get_repository("devtable", "building"))
+
+  repo_build = get_repository_build(build.uuid)
+
+  assert repo_build.phase == BUILD_PHASE.WAITING
+  assert update_phase_then_close(build.uuid, BUILD_PHASE.COMPLETE)
+
+  repo_build = get_repository_build(build.uuid)
+  assert repo_build.phase == BUILD_PHASE.COMPLETE
+
+  repo_build.delete_instance()
+  assert not update_phase_then_close(repo_build.uuid, BUILD_PHASE.PULLING)
+
+
+def create_build(repository):
+  new_token = model.token.create_access_token(repository, 'write', 'build-worker')
+  repo = 'ci.devtable.com:5000/%s/%s' % (repository.namespace_user.username, repository.name)
+  job_config = {
+    'repository': repo,
+    'docker_tags': ['latest'],
+    'build_subdir': '',
+    'trigger_metadata': {
+      'commit': '3482adc5822c498e8f7db2e361e8d57b3d77ddd9',
+      'ref': 'refs/heads/master',
+      'default_branch': 'master'
+    }
+  }
+  build = create_repository_build(repository, new_token, job_config,
+                                  '68daeebd-a5b9-457f-80a0-4363b882f8ea',
+                                  "build_name")
+  build.save()
+  return build

test/test_queries.py

@@ -1,118 +0,0 @@
-import unittest
-from unittest import TestCase
-
-from app import app
-from data.model.build import get_repository_build, update_phase_then_close, create_repository_build
-from initdb import setup_database_for_testing, finished_database_for_testing
-from data import model
-from data.database import RepositoryBuild, Image, ImageStorage, BUILD_PHASE
-
-ADMIN_ACCESS_USER = 'devtable'
-SIMPLE_REPO = 'simple'
-
-
-class TestSpecificQueries(unittest.TestCase):
-  def setUp(self):
-    setup_database_for_testing(self)
-    self.app = app.test_client()
-    self.ctx = app.test_request_context()
-    self.ctx.__enter__()
-
-  def tearDown(self):
-    finished_database_for_testing(self)
-    self.ctx.__exit__(True, None, None)
-
-  def test_archivable_buildlogs(self):
-    # Make sure there are no archivable logs.
-    result = model.build.get_archivable_build()
-    self.assertIsNone(result)
-
-    # Add a build that cannot (yet) be archived.
-    repo = model.repository.get_repository(ADMIN_ACCESS_USER, SIMPLE_REPO)
-    token = model.token.create_access_token(repo, 'write')
-    created = RepositoryBuild.create(repository=repo, access_token=token,
-                                     phase=model.build.BUILD_PHASE.WAITING,
-                                     logs_archived=False, job_config='{}',
-                                     display_name='')
-
-    # Make sure there are no archivable logs.
-    result = model.build.get_archivable_build()
-    self.assertIsNone(result)
-
-    # Change the build to being complete.
-    created.phase = model.build.BUILD_PHASE.COMPLETE
-    created.save()
-
-    # Make sure we now find an archivable build.
-    result = model.build.get_archivable_build()
-    self.assertEquals(created.id, result.id)
-    self.assertEquals(created.uuid, result.uuid)
-
-  def test_lookup_repo_blob(self):
-    repo = model.repository.get_repository(ADMIN_ACCESS_USER, SIMPLE_REPO)
-    expected = list(ImageStorage.select().join(Image).where(Image.repository == repo))
-    self.assertTrue(len(expected) > 0)
-
-    for storage in expected:
-      found = model.blob.get_repo_blob_by_digest(ADMIN_ACCESS_USER, SIMPLE_REPO,
-                                                 storage.content_checksum)
-      self.assertEquals(found.id, storage.id)
-
-    try:
-      model.blob.get_repo_blob_by_digest(ADMIN_ACCESS_USER, SIMPLE_REPO, 'invalidchecksum')
-    except model.BlobDoesNotExist:
-      return
-
-    self.fail('Expected BlobDoesNotExist exception')
-
-
-class TestUpdatePhase(unittest.TestCase):
-  def setUp(self):
-    setup_database_for_testing(self)
-    self.app = app.test_client()
-    self.ctx = app.test_request_context()
-    self.ctx.__enter__()
-
-  def tearDown(self):
-    finished_database_for_testing(self)
-    self.ctx.__exit__(True, None, None)
-
-  def testUpdatePhase(self):
-    build = self.create_build(model.repository.get_repository("devtable", "building"))
-
-    repo_build = get_repository_build(build.uuid)
-
-    self.assertEqual(repo_build.phase, BUILD_PHASE.WAITING)
-    self.assertTrue(update_phase_then_close(build.uuid, BUILD_PHASE.COMPLETE))
-
-    repo_build = get_repository_build(build.uuid)
-
-    self.assertEqual(repo_build.phase, BUILD_PHASE.COMPLETE)
-
-    repo_build.delete_instance()
-
-    self.assertFalse(update_phase_then_close(repo_build.uuid, BUILD_PHASE.PULLING))
-
-  @staticmethod
-  def create_build(repository):
-    new_token = model.token.create_access_token(repository, 'write', 'build-worker')
-    repo = 'ci.devtable.com:5000/%s/%s' % (repository.namespace_user.username, repository.name)
-    job_config = {
-      'repository': repo,
-      'docker_tags': ['latest'],
-      'build_subdir': '',
-      'trigger_metadata': {
-        'commit': '3482adc5822c498e8f7db2e361e8d57b3d77ddd9',
-        'ref': 'refs/heads/master',
-        'default_branch': 'master'
-      }
-    }
-    build = create_repository_build(repository, new_token, job_config,
-                                    '68daeebd-a5b9-457f-80a0-4363b882f8ea',
-                                    "build_name")
-    build.save()
-    return build
-
-
-if __name__ == '__main__':
-  unittest.main()