55 lines
1.9 KiB
Python
55 lines
1.9 KiB
Python
|
import unittest
|
||
|
import time
|
||
|
|
||
|
from app import app, storage
|
||
|
from initdb import setup_database_for_testing, finished_database_for_testing
|
||
|
from data import model, database
|
||
|
|
||
|
ADMIN_ACCESS_USER = 'devtable'
|
||
|
REPO = 'simple'
|
||
|
|
||
|
class TestBlobModel(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_store_blob(self):
|
||
|
location = database.ImageStorageLocation.select().get()
|
||
|
|
||
|
# Create a new blob at a unique digest.
|
||
|
digest = 'somecooldigest'
|
||
|
blob_storage = model.blob.store_blob_record_and_temp_link(ADMIN_ACCESS_USER, REPO, digest,
|
||
|
location, 1024, 0, 5000)
|
||
|
|
||
|
self.assertEquals(blob_storage.content_checksum, digest)
|
||
|
self.assertEquals(blob_storage.image_size, 1024)
|
||
|
self.assertEquals(blob_storage.uncompressed_size, 5000)
|
||
|
|
||
|
# Link to the same digest.
|
||
|
blob_storage2 = model.blob.store_blob_record_and_temp_link(ADMIN_ACCESS_USER, REPO, digest,
|
||
|
location, 2048, 0, 6000)
|
||
|
|
||
|
self.assertEquals(blob_storage2.id, blob_storage.id)
|
||
|
self.assertEquals(blob_storage2.image_size, 2048)
|
||
|
self.assertEquals(blob_storage2.uncompressed_size, 6000)
|
||
|
|
||
|
# Add a new digest, ensure it has a new record.
|
||
|
otherdigest = 'anotherdigest'
|
||
|
blob_storage3 = model.blob.store_blob_record_and_temp_link(ADMIN_ACCESS_USER, REPO, otherdigest,
|
||
|
location, 1234, 0, 5678)
|
||
|
|
||
|
self.assertNotEquals(blob_storage3.id, blob_storage.id)
|
||
|
self.assertEquals(blob_storage3.image_size, 1234)
|
||
|
self.assertEquals(blob_storage3.uncompressed_size, 5678)
|
||
|
|
||
|
|
||
|
if __name__ == '__main__':
|
||
|
unittest.main()
|