mirror of
https://github.com/quay/quay.git
synced 2026-01-26 06:21:37 +03:00
5f63b3a7bb
* chore: drop deprecated tables and remove unused code * isort imports * migration: check for table existence before drop
62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
from test.fixtures import *
|
|
|
|
from app import storage
|
|
from data import database, model
|
|
|
|
ADMIN_ACCESS_USER = "devtable"
|
|
REPO = "simple"
|
|
|
|
|
|
def test_store_blob(initialized_db):
|
|
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
|
|
)
|
|
assert blob_storage.content_checksum == digest
|
|
assert blob_storage.image_size == 1024
|
|
assert 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
|
|
)
|
|
assert blob_storage2.id == blob_storage.id
|
|
|
|
# The sizes should be unchanged.
|
|
assert blob_storage2.image_size == 1024
|
|
assert blob_storage2.uncompressed_size == 5000
|
|
|
|
# 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
|
|
)
|
|
assert blob_storage3.id != blob_storage.id
|
|
assert blob_storage3.image_size == 1234
|
|
assert blob_storage3.uncompressed_size == 5678
|
|
|
|
|
|
def test_get_or_create_shared_blob(initialized_db):
|
|
shared = model.blob.get_or_create_shared_blob("sha256:abcdef", b"somecontent", storage)
|
|
assert shared.content_checksum == "sha256:abcdef"
|
|
|
|
again = model.blob.get_or_create_shared_blob("sha256:abcdef", b"somecontent", storage)
|
|
assert shared == again
|
|
|
|
|
|
def test_lookup_repo_storages_by_content_checksum(initialized_db):
|
|
query = database.ImageStorage.select(database.ImageStorage, database.ManifestBlob).join(
|
|
database.ManifestBlob
|
|
)
|
|
|
|
for imagestorage in query:
|
|
manifestblob = imagestorage.manifestblob
|
|
found = model.storage.lookup_repo_storages_by_content_checksum(
|
|
manifestblob.repository, [imagestorage.content_checksum]
|
|
)
|
|
assert len(found) == 1
|
|
assert found[0].content_checksum == imagestorage.content_checksum
|