vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Merge pull request #2826 from jzelinskie/appr-v22

Browse source 
endpoints.appr: move to new v22 format
This commit is contained in:
Jimmy Zelinskie 2017-07-31 07:36:40 -07:00 committed by GitHub
commit fe6760749a
9 changed files with 279 additions and 251 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

0
data/interfaces/__init__.py
View file

9
data/interfaces/test/test_appr.py
View file

@ -1,9 +0,0 @@
import pytest
from data.interfaces.appr import _strip_sha256_header
@pytest.mark.parametrize('digest,expected', [
('sha256:251b6897608fb18b8a91ac9abac686e2e95245d5a041f2d1e78fe7a815e6480a', '251b6897608fb18b8a91ac9abac686e2e95245d5a041f2d1e78fe7a815e6480a'),
('251b6897608fb18b8a91ac9abac686e2e95245d5a041f2d1e78fe7a815e6480a', '251b6897608fb18b8a91ac9abac686e2e95245d5a041f2d1e78fe7a815e6480a'),])
def test_stip_sha256(digest, expected):
assert _strip_sha256_header(digest) == expected

49
endpoints/appr/cnr_backend.py
View file

@ -7,8 +7,7 @@ from cnr.models.db_base import CnrDB
from cnr.models.package_base import PackageBase, manifest_media_type
from app import storage
from data.interfaces.appr import oci_app_model
from data.oci_model import blob # TODO these calls should be through oci_app_model
from endpoints.appr.models_oci import model
class Blob(BlobBase):
@ -17,7 +16,7 @@ class Blob(BlobBase):
return "cnr/blobs/sha256/%s/%s" % (digest[0:2], digest)
def save(self, content_media_type):
oci_app_model.store_blob(self, content_media_type)
model.store_blob(self, content_media_type)
@classmethod
def delete(cls, package_name, digest):
@ -26,7 +25,7 @@ class Blob(BlobBase):
@classmethod
def _fetch_b64blob(cls, package_name, digest):
blobpath = cls.upload_url(digest)
locations = blob.get_blob_locations(digest)
locations = model.get_blob_locations(digest)
if not locations:
raise_package_not_found(package_name, digest)
return base64.b64encode(storage.get_content(locations, blobpath))
@ -34,7 +33,7 @@ class Blob(BlobBase):
@classmethod
def download_url(cls, package_name, digest):
blobpath = cls.upload_url(digest)
locations = blob.get_blob_locations(digest)
locations = model.get_blob_locations(digest)
if not locations:
raise_package_not_found(package_name, digest)
return storage.get_direct_download_url(locations, blobpath)
@ -49,29 +48,29 @@ class Channel(ChannelBase):
def _exists(self):
""" Check if the channel is saved already """
return oci_app_model.channel_exists(self.package, self.name)
return model.channel_exists(self.package, self.name)
@classmethod
def get(cls, name, package):
chanview = oci_app_model.fetch_channel(package, name, with_releases=False)
chanview = model.fetch_channel(package, name, with_releases=False)
return cls(name, package, chanview.current)
def save(self):
oci_app_model.update_channel(self.package, self.name, self.current)
model.update_channel(self.package, self.name, self.current)
def delete(self):
oci_app_model.delete_channel(self.package, self.name)
model.delete_channel(self.package, self.name)
@classmethod
def all(cls, package_name):
return [
Channel(c.name, package_name, c.current) for c in oci_app_model.list_channels(package_name)
Channel(c.name, package_name, c.current) for c in model.list_channels(package_name)
]
@property
def _channel(self):
if self._channel_data is None:
self._channel_data = oci_app_model.fetch_channel(self.package, self.name)
self._channel_data = model.fetch_channel(self.package, self.name)
return self._channel_data
def releases(self):
@ -79,10 +78,10 @@ class Channel(ChannelBase):
return self._channel.releases
def _add_release(self, release):
return oci_app_model.update_channel(self.package, self.name, release)._asdict
return model.update_channel(self.package, self.name, release)._asdict
def _remove_release(self, release):
oci_app_model.delete_channel(self.package, self.name)
model.delete_channel(self.package, self.name)
class User(object):
@ -91,7 +90,7 @@ class User(object):
@classmethod
def get_user(cls, username, password):
""" Returns True if user creds is valid """
return oci_app_model.get_user(username, password)
return model.get_user(username, password)
class Package(PackageBase):
@ -110,55 +109,55 @@ class Package(PackageBase):
@classmethod
def create_repository(cls, package_name, visibility, owner):
oci_app_model.create_application(package_name, visibility, owner)
model.create_application(package_name, visibility, owner)
@classmethod
def exists(cls, package_name):
return oci_app_model.application_exists(package_name)
return model.application_exists(package_name)
@classmethod
def all(cls, organization=None, media_type=None, search=None, username=None, **kwargs):
return [
dict(x._asdict())
for x in oci_app_model.list_applications(namespace=organization, media_type=media_type,
for x in model.list_applications(namespace=organization, media_type=media_type,
search=search, username=username)
]
@classmethod
def _fetch(cls, package_name, release, media_type):
data = oci_app_model.fetch_release(package_name, release, manifest_media_type(media_type))
data = model.fetch_release(package_name, release, manifest_media_type(media_type))
return cls._apptuple_to_dict(data)
@classmethod
def all_releases(cls, package_name, media_type=None):
return oci_app_model.list_releases(package_name, media_type)
return model.list_releases(package_name, media_type)
@classmethod
def search(cls, query, username=None):
return oci_app_model.basic_search(query, username=username)
return model.basic_search(query, username=username)
def _save(self, force=False, **kwargs):
user = kwargs['user']
visibility = kwargs['visibility']
oci_app_model.create_release(self, user, visibility, force)
model.create_release(self, user, visibility, force)
@classmethod
def _delete(cls, package_name, release, media_type):
oci_app_model.delete_release(package_name, release, manifest_media_type(media_type))
model.delete_release(package_name, release, manifest_media_type(media_type))
@classmethod
def isdeleted_release(cls, package, release):
return oci_app_model.release_exists(package, release)
return model.release_exists(package, release)
def channels(self, channel_class, iscurrent=True):
return [
c.name
for c in oci_app_model.list_release_channels(self.package, self.release, active=iscurrent)
for c in model.list_release_channels(self.package, self.release, active=iscurrent)
]
@classmethod
def manifests(cls, package, release=None):
return oci_app_model.list_manifests(package, release)
return model.list_manifests(package, release)
@classmethod
def dump_all(cls, blob_cls):

1
endpoints/appr/decorators.py
View file

@ -3,7 +3,6 @@ import logging
from functools import wraps
from data import model
from util.http import abort

191
endpoints/appr/models_interface.py Normal file
View file

@ -0,0 +1,191 @@
from abc import ABCMeta, abstractmethod
from collections import namedtuple
from six import add_metaclass
class BlobDescriptor(namedtuple('Blob', ['mediaType', 'size', 'digest', 'urls'])):
""" BlobDescriptor describes a blob with its mediatype, size and digest.
A BlobDescriptor is used to retrieves the actual blob.
"""
class ChannelReleasesView(namedtuple('ChannelReleasesView', ['name', 'current', 'releases'])):
""" A channel is a pointer to a Release (current).
Releases are the previous tags pointed by channel (history).
"""
class ChannelView(namedtuple('ChannelView', ['name', 'current'])):
""" A channel is a pointer to a Release (current).
"""
class ApplicationSummaryView(
namedtuple('ApplicationSummaryView', [
'name', 'namespace', 'visibility', 'default', 'manifests', 'channels', 'releases',
'updated_at', 'created_at'
])):
""" ApplicationSummaryView is an aggregated view of an application repository.
"""
class ApplicationManifest(namedtuple('ApplicationManifest', ['mediaType', 'digest', 'content'])):
""" ApplicationManifest embed the BlobDescriptor and some metadata around it.
An ApplicationManifest is content-addressable.
"""
class ApplicationRelease(
namedtuple('ApplicationRelease', ['release', 'name', 'created_at', 'manifest'])):
""" The ApplicationRelease associates an ApplicationManifest to a repository and release.
"""
@add_metaclass(ABCMeta)
class AppRegistryDataInterface(object):
""" Interface that represents all data store interactions required by a App Registry.
"""
@abstractmethod
def list_applications(self, namespace=None, media_type=None, search=None, username=None,
with_channels=False):
""" Lists all repositories that contain applications, with optional filtering to a specific
namespace and/or to those visible to a specific user.
Returns: list of ApplicationSummaryView
"""
pass
@abstractmethod
def application_is_public(self, package_name):
"""
Returns true if the application is public
"""
pass
@abstractmethod
def create_application(self, package_name, visibility, owner):
""" Create a new app repository, owner is the user who creates it """
pass
@abstractmethod
def application_exists(self, package_name):
""" Returns true if the application exists """
pass
@abstractmethod
def basic_search(self, query, username=None):
""" Returns an array of matching application in the format: 'namespace/name'
Note:
* Only 'public' repositories are returned
"""
pass
# @TODO: Paginate
@abstractmethod
def list_releases(self, package_name, media_type=None):
""" Returns the list of all releases(names) of an AppRepository
Example:
>>> get_app_releases('ant31/rocketchat')
['1.7.1', '1.7.0', '1.7.2']
"""
pass
# @TODO: Paginate
@abstractmethod
def list_manifests(self, package_name, release=None):
""" Returns the list of all available manifests type of an Application across all releases or
for a specific one.
Example:
>>> get_app_releases('ant31/rocketchat')
['1.7.1', '1.7.0', '1.7.2']
"""
pass
@abstractmethod
def fetch_release(self, package_name, release, media_type):
"""
Returns an ApplicationRelease
"""
pass
@abstractmethod
def store_blob(self, cnrblob, content_media_type):
"""
Upload the blob content to a storage location and creates a Blob entry in the DB.
Returns a BlobDescriptor
"""
pass
@abstractmethod
def create_release(self, package, user, visibility, force=False):
""" Creates and returns an ApplicationRelease
- package is a data.model.Package object
- user is the owner of the package
- visibility is a string: 'public' or 'private'
"""
pass
@abstractmethod
def release_exists(self, package, release):
""" Return true if a release with that name already exist or
has existed (including deleted ones)
"""
pass
@abstractmethod
def delete_release(self, package_name, release, media_type):
""" Remove/Delete an app-release from an app-repository.
It does not delete the entire app-repository, only a single release
"""
pass
@abstractmethod
def list_release_channels(self, package_name, release, active=True):
""" Returns a list of Channel that are/was pointing to a release.
If active is True, returns only active Channel (lifetime_end not null)
"""
pass
@abstractmethod
def channel_exists(self, package_name, channel_name):
""" Returns true if the channel with the given name exists under the matching package """
pass
@abstractmethod
def update_channel(self, package_name, channel_name, release):
""" Append a new release to the Channel
Returns a new Channel with the release as current
"""
pass
@abstractmethod
def delete_channel(self, package_name, channel_name):
""" Delete a Channel, it doesn't delete/touch the ApplicationRelease pointed by the channel """
# @TODO: Paginate
@abstractmethod
def list_channels(self, package_name):
""" Returns all AppChannel for a package """
pass
@abstractmethod
def fetch_channel(self, package_name, channel_name, with_releases=True):
""" Returns an Channel
Raises: ChannelNotFound, PackageNotFound
"""
pass
@abstractmethod
def log_action(self, event_name, namespace_name, repo_name=None, analytics_name=None,
analytics_sample=1, **kwargs):
""" Logs an action to the audit log. """
pass
@abstractmethod
def get_blob_locations(self, digest):
""" Returns a list of strings for the locations in which a Blob is present. """
pass

251
data/interfaces/appr.py → endpoints/appr/models_oci.py
View file

@ -1,183 +1,26 @@
from abc import ABCMeta, abstractmethod
from collections import namedtuple
from datetime import datetime
import cnr.semver
from cnr.exception import raise_package_not_found, raise_channel_not_found
from six import add_metaclass
import data.model
from app import storage, authentication
from data import model, oci_model
from data import oci_model
from data.database import Tag, Manifest, MediaType, Blob, Repository, Channel
from endpoints.appr.models_interface import (
ApplicationManifest, ApplicationRelease, ApplicationSummaryView, AppRegistryDataInterface,
BlobDescriptor, ChannelView, ChannelReleasesView)
from util.audit import track_and_log
from util.morecollections import AttrDict
from util.names import parse_robot_username
class BlobDescriptor(namedtuple('Blob', ['mediaType', 'size', 'digest', 'urls'])):
""" BlobDescriptor describes a blob with its mediatype, size and digest.
A BlobDescriptor is used to retrieves the actual blob.
"""
class ChannelReleasesView(namedtuple('ChannelReleasesView', ['name', 'current', 'releases'])):
""" A channel is a pointer to a Release (current).
Releases are the previous tags pointed by channel (history).
"""
class ChannelView(namedtuple('ChannelView', ['name', 'current'])):
""" A channel is a pointer to a Release (current).
"""
class ApplicationSummaryView(
namedtuple('ApplicationSummaryView', [
'name', 'namespace', 'visibility', 'default', 'manifests', 'channels', 'releases',
'updated_at', 'created_at'
])):
""" ApplicationSummaryView is an aggregated view of an application repository.
"""
class ApplicationManifest(namedtuple('ApplicationManifest', ['mediaType', 'digest', 'content'])):
""" ApplicationManifest embed the BlobDescriptor and some metadata around it.
An ApplicationManifest is content-addressable.
"""
class ApplicationRelease(
namedtuple('ApplicationRelease', ['release', 'name', 'created_at', 'manifest'])):
""" The ApplicationRelease associates an ApplicationManifest to a repository and release.
"""
@add_metaclass(ABCMeta)
class AppRegistryDataInterface(object):
""" Interface that represents all data store interactions required by a App Registry.
"""
@abstractmethod
def list_applications(self, namespace=None, media_type=None, search=None, username=None,
with_channels=False):
""" Lists all repositories that contain applications, with optional filtering to a specific
namespace and/or to those visible to a specific user.
Returns: list of ApplicationSummaryView
"""
@abstractmethod
def application_is_public(self, package_name):
"""
Returns true if the application is public
"""
@abstractmethod
def create_application(self, package_name, visibility, owner):
""" Create a new app repository, owner is the user who creates it """
@abstractmethod
def application_exists(self, package_name):
""" Returns true if the application exists """
@abstractmethod
def basic_search(self, query, username=None):
""" Returns an array of matching application in the format: 'namespace/name'
Note:
* Only 'public' repositories are returned
"""
# @TODO: Paginate
@abstractmethod
def list_releases(self, package_name, media_type=None):
""" Returns the list of all releases(names) of an AppRepository
Example:
>>> get_app_releases('ant31/rocketchat')
['1.7.1', '1.7.0', '1.7.2']
"""
# @TODO: Paginate
@abstractmethod
def list_manifests(self, package_name, release=None):
""" Returns the list of all available manifests type of an Application across all releases or
for a specific one.
Example:
>>> get_app_releases('ant31/rocketchat')
['1.7.1', '1.7.0', '1.7.2']
"""
@abstractmethod
def fetch_release(self, package_name, release, media_type):
"""
Returns an ApplicationRelease
"""
@abstractmethod
def store_blob(self, cnrblob, content_media_type):
"""
Upload the blob content to a storage location and creates a Blob entry in the DB.
Returns a BlobDescriptor
"""
@abstractmethod
def create_release(self, package, user, visibility, force=False):
""" Creates and returns an ApplicationRelease
- package is a data.model.Package object
- user is the owner of the package
- visibility is a string: 'public' or 'private'
"""
@abstractmethod
def release_exists(self, package, release):
""" Return true if a release with that name already exist or
has existed (including deleted ones)
"""
pass
@abstractmethod
def delete_release(self, package_name, release, media_type):
""" Remove/Delete an app-release from an app-repository.
It does not delete the entire app-repository, only a single release
"""
@abstractmethod
def list_release_channels(self, package_name, release, active=True):
""" Returns a list of Channel that are/was pointing to a release.
If active is True, returns only active Channel (lifetime_end not null)
"""
@abstractmethod
def channel_exists(self, package_name, channel_name):
""" Returns true if the channel with the given name exists under the matching package """
@abstractmethod
def update_channel(self, package_name, channel_name, release):
""" Append a new release to the Channel
Returns a new Channel with the release as current
"""
@abstractmethod
def delete_channel(self, package_name, channel_name):
""" Delete a Channel, it doesn't delete/touch the ApplicationRelease pointed by the channel """
# @TODO: Paginate
@abstractmethod
def list_channels(self, package_name):
""" Returns all AppChannel for a package """
@abstractmethod
def fetch_channel(self, package_name, channel_name, with_releases=True):
""" Returns an Channel
Raises: ChannelNotFound, PackageNotFound
"""
@abstractmethod
def log_action(self, event_name, namespace_name, repo_name=None, analytics_name=None,
analytics_sample=1, **kwargs):
""" Logs an action to the audit log. """
def _strip_sha256_header(digest):
if digest.startswith('sha256:'):
return digest.split('sha256:')[1]
return digest
def _split_package_name(package):
@ -196,27 +39,27 @@ def _timestamp_to_iso(timestamp, in_ms=True):
return datetime.fromtimestamp(timestamp).isoformat()
class OCIAppModel(AppRegistryDataInterface):
def _application(self, package):
ns, name = _split_package_name(package)
repo = model.repository.get_app_repository(ns, name)
if repo is None:
raise_package_not_found(package)
return repo
def _application(package):
ns, name = _split_package_name(package)
repo = data.model.repository.get_app_repository(ns, name)
if repo is None:
raise_package_not_found(package)
return repo
class OCIAppModel(AppRegistryDataInterface):
def log_action(self, event_name, namespace_name, repo_name=None, analytics_name=None,
analytics_sample=1, metadata=None):
metadata = {} if metadata is None else metadata
repo = None
if repo_name is not None:
db_repo = model.repository.get_repository(namespace_name, repo_name,
db_repo = data.model.repository.get_repository(namespace_name, repo_name,
kind_filter='application')
repo = AttrDict({
'id': db_repo.id,
'name': db_repo.name,
'namespace_name': db_repo.namespace_user.username,
})
'namespace_name': db_repo.namespace_user.username,})
track_and_log(event_name, repo, analytics_name=analytics_name,
analytics_sample=analytics_sample, **metadata)
@ -233,14 +76,12 @@ class OCIAppModel(AppRegistryDataInterface):
if not releases:
continue
available_releases = [
str(x) for x in sorted(cnr.semver.versions(releases, False), reverse=True)
]
str(x) for x in sorted(cnr.semver.versions(releases, False), reverse=True)]
channels = None
if with_channels:
channels = [
ChannelView(name=chan.name, current=chan.linked_tag.name)
for chan in oci_model.channel.get_repo_channels(repo)
]
for chan in oci_model.channel.get_repo_channels(repo)]
app_name = _join_package_name(repo.namespace_user.username, repo.name)
manifests = self.list_manifests(app_name, available_releases[0])
@ -263,17 +104,17 @@ class OCIAppModel(AppRegistryDataInterface):
* True if the repository is public
"""
namespace, name = _split_package_name(package_name)
return model.repository.repository_is_public(namespace, name)
return data.model.repository.repository_is_public(namespace, name)
def create_application(self, package_name, visibility, owner):
""" Create a new app repository, owner is the user who creates it """
ns, name = _split_package_name(package_name)
model.repository.create_repository(ns, name, owner, visibility, 'application')
data.model.repository.create_repository(ns, name, owner, visibility, 'application')
def application_exists(self, package_name):
""" Create a new app repository, owner is the user who creates it """
ns, name = _split_package_name(package_name)
return model.repository.get_repository(ns, name, kind_filter='application') is not None
return data.model.repository.get_repository(ns, name, kind_filter='application') is not None
def basic_search(self, query, username=None):
""" Returns an array of matching AppRepositories in the format: 'namespace/name'
@ -285,8 +126,7 @@ class OCIAppModel(AppRegistryDataInterface):
"""
return [
_join_package_name(r.namespace_user.username, r.name)
for r in model.repository.get_app_search(lookup=query, username=username, limit=50)
]
for r in data.model.repository.get_app_search(lookup=query, username=username, limit=50)]
def list_releases(self, package_name, media_type=None):
""" Return the list of all releases of an Application
@ -297,7 +137,7 @@ class OCIAppModel(AppRegistryDataInterface):
Todo:
* Paginate
"""
return oci_model.release.get_releases(self._application(package_name), media_type)
return oci_model.release.get_releases(_application(package_name), media_type)
def list_manifests(self, package_name, release=None):
""" Returns the list of all manifests of an Application.
@ -306,7 +146,7 @@ class OCIAppModel(AppRegistryDataInterface):
* Paginate
"""
try:
repo = self._application(package_name)
repo = _application(package_name)
return list(oci_model.manifest.get_manifest_types(repo, release))
except (Repository.DoesNotExist, Tag.DoesNotExist):
raise_package_not_found(package_name, release)
@ -315,7 +155,7 @@ class OCIAppModel(AppRegistryDataInterface):
"""
Retrieves an AppRelease from it's repository-name and release-name
"""
repo = self._application(package_name)
repo = _application(package_name)
try:
tag, manifest, blob = oci_model.release.get_app_release(repo, release, media_type)
created_at = _timestamp_to_iso(tag.lifetime_start)
@ -348,19 +188,19 @@ class OCIAppModel(AppRegistryDataInterface):
package is an instance of data.cnr.package.Package
"""
data = package.manifest()
manifest = package.manifest()
ns, name = package.namespace, package.name
repo = model.repository.get_or_create_repository(ns, name, user, visibility=visibility,
repo = data.model.repository.get_or_create_repository(ns, name, user, visibility=visibility,
repo_kind='application')
tag_name = package.release
oci_model.release.create_app_release(repo, tag_name,
package.manifest(), data['content']['digest'], force)
package.manifest(), manifest['content']['digest'], force)
def delete_release(self, package_name, release, media_type):
""" Remove/Delete an app-release from an app-repository.
It does not delete the entire app-repository, only a single release
"""
repo = self._application(package_name)
repo = _application(package_name)
try:
oci_model.release.delete_app_release(repo, release, media_type)
except (Channel.DoesNotExist, Tag.DoesNotExist, MediaType.DoesNotExist):
@ -372,7 +212,7 @@ class OCIAppModel(AppRegistryDataInterface):
def channel_exists(self, package_name, channel_name):
""" Returns true if channel exists """
repo = self._application(package_name)
repo = _application(package_name)
return oci_model.tag.tag_exists(repo, channel_name, "channel")
def delete_channel(self, package_name, channel_name):
@ -380,7 +220,7 @@ class OCIAppModel(AppRegistryDataInterface):
Note:
It doesn't delete the AppReleases
"""
repo = self._application(package_name)
repo = _application(package_name)
try:
oci_model.channel.delete_channel(repo, channel_name)
except (Channel.DoesNotExist, Tag.DoesNotExist):
@ -388,13 +228,13 @@ class OCIAppModel(AppRegistryDataInterface):
def list_channels(self, package_name):
""" Returns all AppChannel for a package """
repo = self._application(package_name)
repo = _application(package_name)
channels = oci_model.channel.get_repo_channels(repo)
return [ChannelView(name=chan.name, current=chan.linked_tag.name) for chan in channels]
def fetch_channel(self, package_name, channel_name, with_releases=True):
""" Returns an AppChannel """
repo = self._application(package_name)
repo = _application(package_name)
try:
channel = oci_model.channel.get_channel(repo, channel_name)
@ -412,7 +252,7 @@ class OCIAppModel(AppRegistryDataInterface):
return chanview
def list_release_channels(self, package_name, release, active=True):
repo = self._application(package_name)
repo = _application(package_name)
try:
channels = oci_model.channel.get_tag_channels(repo, release, active=active)
return [ChannelView(name=c.name, current=c.linked_tag.name) for c in channels]
@ -424,7 +264,7 @@ class OCIAppModel(AppRegistryDataInterface):
Returns:
A new AppChannel with the release
"""
repo = self._application(package_name)
repo = _application(package_name)
channel = oci_model.channel.create_or_update_channel(repo, channel_name, release)
return ChannelView(current=channel.linked_tag.name, name=channel.name)
@ -432,18 +272,15 @@ class OCIAppModel(AppRegistryDataInterface):
err_msg = None
if parse_robot_username(username) is not None:
try:
user = model.user.verify_robot(username, password)
except model.InvalidRobotException as exc:
user = data.model.user.verify_robot(username, password)
except data.model.InvalidRobotException as exc:
return (None, exc.message)
else:
user, err_msg = authentication.verify_and_link_user(username, password)
return (user, err_msg)
def _strip_sha256_header(digest):
if digest.startswith('sha256:'):
return digest.split('sha256:')[1]
return digest
def get_blob_locations(self, digest):
return oci_model.blob.get_blob_locations(digest)
oci_app_model = OCIAppModel()
model = OCIAppModel()

6
endpoints/appr/registry.py
View file

@ -12,11 +12,11 @@ from flask import jsonify, request
from auth.auth_context import get_authenticated_user
from auth.decorators import process_auth
from auth.permissions import (CreateRepositoryPermission, ModifyRepositoryPermission)
from data.interfaces.appr import oci_app_model as model
from endpoints.appr import (appr_bp, require_app_repo_read, require_app_repo_write)
from auth.permissions import CreateRepositoryPermission, ModifyRepositoryPermission
from endpoints.appr import appr_bp, require_app_repo_read, require_app_repo_write
from endpoints.appr.cnr_backend import Blob, Channel, Package, User
from endpoints.appr.decorators import disallow_for_image_repository
from endpoints.appr.models_oci import model
from endpoints.decorators import anon_allowed, anon_protect
from util.names import REPOSITORY_NAME_REGEX, TAG_REGEX

12
endpoints/appr/test/test_api.py
View file

@ -9,10 +9,10 @@ from cnr.tests.test_models import CnrTestModels
import data.oci_model.blob as oci_blob
from data.database import User
from data.interfaces.appr import oci_app_model
from data.model import organization, user
from endpoints.appr import registry # Needed to register the endpoint
from endpoints.appr.cnr_backend import Channel, Package, QuayDB
from endpoints.appr.models_oci import model as oci_app_model
from test.fixtures import *
@ -57,14 +57,14 @@ class PackageTest(Package):
for mtype in cls.manifests(package_name, release):
package = oci_app_model.fetch_release(package_name, release, mtype)
blob = blob_cls.get(package_name, package.manifest.content.digest)
data = cls._apptuple_to_dict(package)
data.pop('digest')
data['channels'] = [
app_data = cls._apptuple_to_dict(package)
app_data.pop('digest')
app_data['channels'] = [
x.name
for x in oci_app_model.list_release_channels(package_name, package.release, False)
]
data['blob'] = blob.b64blob
result.append(data)
app_data['blob'] = blob.b64blob
result.append(app_data)
return result

11
endpoints/appr/test/test_digest_prefix.py Normal file
View file

@ -0,0 +1,11 @@
import pytest
from endpoints.appr.models_oci import _strip_sha256_header
@pytest.mark.parametrize('digest,expected', [
('sha256:251b6897608fb18b8a91ac9abac686e2e95245d5a041f2d1e78fe7a815e6480a',
'251b6897608fb18b8a91ac9abac686e2e95245d5a041f2d1e78fe7a815e6480a'),
('251b6897608fb18b8a91ac9abac686e2e95245d5a041f2d1e78fe7a815e6480a',
'251b6897608fb18b8a91ac9abac686e2e95245d5a041f2d1e78fe7a815e6480a'),])
def test_stip_sha256(digest, expected):
assert _strip_sha256_header(digest) == expected