quay/test/test_storagereplication.py

import hashlib

import pytest

from app import storage
from data import model, database
from storage.basestorage import StoragePaths
from workers.storagereplication import StorageReplicationWorker

from test.fixtures import *


@pytest.fixture()
def storage_user(app):
  user = model.user.get_user('devtable')
  database.UserRegion.create(user=user,
                             location=database.ImageStorageLocation.get(name='local_us'))
  database.UserRegion.create(user=user,
                             location=database.ImageStorageLocation.get(name='local_eu'))
  return user


@pytest.fixture()
def storage_paths():
  return StoragePaths()


@pytest.fixture()
def replication_worker():
  return StorageReplicationWorker(None)


def test_storage_replication_v1(storage_user, storage_paths, replication_worker, app):
  # Add a storage entry with a V1 path.
  v1_storage = model.storage.create_v1_storage('local_us')
  content_path = storage_paths.v1_image_layer_path(v1_storage.uuid)
  storage.put_content(['local_us'], content_path, 'some content')

  # Call replicate on it and verify it replicates.
  result = replication_worker.replicate_storage(storage_user, v1_storage.uuid)
  assert result

  # Ensure that the data was replicated to the other "region".
  assert storage.get_content(['local_eu'], content_path) == 'some content'

  locations = model.storage.get_storage_locations(v1_storage.uuid)
  assert len(locations) == 2


def test_storage_replication_cas(storage_user, storage_paths, replication_worker, app):
  # Add a storage entry with a CAS path.
  content_checksum = 'sha256:' + hashlib.sha256('some content').hexdigest()
  cas_storage = database.ImageStorage.create(cas_path=True, content_checksum=content_checksum)

  location = database.ImageStorageLocation.get(name='local_us')
  database.ImageStoragePlacement.create(storage=cas_storage, location=location)

  content_path = storage_paths.blob_path(cas_storage.content_checksum)
  storage.put_content(['local_us'], content_path, 'some content')

  # Call replicate on it and verify it replicates.
  result = replication_worker.replicate_storage(storage_user, cas_storage.uuid)
  assert result

  # Ensure that the data was replicated to the other "region".
  assert storage.get_content(['local_eu'], content_path) == 'some content'

  locations = model.storage.get_storage_locations(cas_storage.uuid)
  assert len(locations) == 2