quay/test/test_queries.py

import unittest

from app import app
from initdb import setup_database_for_testing, finished_database_for_testing
from data import model
from data.database import RepositoryBuild, Repository, Image, ImageStorage

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')



if __name__ == '__main__':
  unittest.main()