68 lines
No EOL
2.3 KiB
Python
68 lines
No EOL
2.3 KiB
Python
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() |