quay/endpoints/api/test/test_repository.py

import pytest

from mock import patch, ANY, MagicMock

from data import model, database
from data.appr_model import release, channel, blob
from endpoints.appr.models_cnr import model as appr_model
from endpoints.api.test.shared import conduct_api_call
from endpoints.api.repository import RepositoryTrust, Repository, RepositoryList
from endpoints.test.shared import client_with_identity
from features import FeatureNameValue

from test.fixtures import *

INVALID_RESPONSE = {
  u'detail': u"u'invalid_req' is not of type 'boolean'",
  u'error_message': u"u'invalid_req' is not of type 'boolean'",
  u'error_type': u'invalid_request',
  u'status': 400,
  u'title': u'invalid_request',
  u'type': u'http://localhost/api/v1/error/invalid_request'
}

NOT_FOUND_RESPONSE = {
  u'detail':
    u'Not Found',
  u'error_message':
    u'Not Found',
  u'error_type':
    u'not_found',
  u'message':
    u'You have requested this URI [/api/v1/repository/devtable/repo/changetrust] but did you mean /api/v1/repository/<apirepopath:repository>/changetrust or /api/v1/repository/<apirepopath:repository>/changevisibility or /api/v1/repository/<apirepopath:repository>/tag/<tag>/images ?',
  u'status':
    404,
  u'title':
    u'not_found',
  u'type':
    u'http://localhost/api/v1/error/not_found'
}


@pytest.mark.parametrize('trust_enabled,repo_found,expected_body,expected_status', [
  (True, True, {
    'success': True
  }, 200),
  (False, True, {
    'success': True
  }, 200),
  (False, False, NOT_FOUND_RESPONSE, 404),
  ('invalid_req', False, INVALID_RESPONSE, 400),
])
def test_post_changetrust(trust_enabled, repo_found, expected_body, expected_status, client):
  with patch('endpoints.api.repository.tuf_metadata_api') as mock_tuf:
    with patch(
        'endpoints.api.repository_models_pre_oci.model.repository.get_repository') as mock_model:
      mock_model.return_value = MagicMock() if repo_found else None
      mock_tuf.get_default_tags_with_expiration.return_value = ['tags', 'expiration']
      with client_with_identity('devtable', client) as cl:
        params = {'repository': 'devtable/repo'}
        request_body = {'trust_enabled': trust_enabled}
        assert expected_body == conduct_api_call(cl, RepositoryTrust, 'POST', params, request_body,
                                                 expected_status).json


def test_signing_disabled(client):
  with patch('features.SIGNING', FeatureNameValue('SIGNING', False)):
    with client_with_identity('devtable', client) as cl:
      params = {'repository': 'devtable/simple'}
      response = conduct_api_call(cl, Repository, 'GET', params).json
      assert not response['trust_enabled']


def test_list_starred_repos(client):
  with client_with_identity('devtable', client) as cl:
    params = {
      'starred': 'true',
    }

    response = conduct_api_call(cl, RepositoryList, 'GET', params).json
    repos = {r['namespace'] + '/' + r['name'] for r in response['repositories']}
    assert 'devtable/simple' in repos
    assert 'public/publicrepo' not in repos

    # Add a star on publicrepo.
    publicrepo = model.repository.get_repository('public', 'publicrepo')
    model.repository.star_repository(model.user.get_user('devtable'), publicrepo)

    # Ensure publicrepo shows up.
    response = conduct_api_call(cl, RepositoryList, 'GET', params).json
    repos = {r['namespace'] + '/' + r['name'] for r in response['repositories']}
    assert 'devtable/simple' in repos
    assert 'public/publicrepo' in repos

    # Make publicrepo private and ensure it disappears.
    model.repository.set_repository_visibility(publicrepo, 'private')

    response = conduct_api_call(cl, RepositoryList, 'GET', params).json
    repos = {r['namespace'] + '/' + r['name'] for r in response['repositories']}
    assert 'devtable/simple' in repos
    assert 'public/publicrepo' not in repos


@pytest.mark.parametrize('repo_name, expected_status', [
  pytest.param('x' * 255, 201, id='Maximum allowed length'),
  pytest.param('x' * 256, 400, id='Over allowed length'),
  pytest.param('a|b', 400, id='Invalid name'),
])
def test_create_repository(repo_name, expected_status, client):
  with client_with_identity('devtable', client) as cl:
    body = {
      'namespace': 'devtable',
      'repository': repo_name,
      'visibility': 'public',
      'description': 'foo',
    }

    result = conduct_api_call(client, RepositoryList, 'post', None, body,
                              expected_code=expected_status).json
    if expected_status == 201:
      assert result['name'] == repo_name
      assert model.repository.get_repository('devtable', repo_name).name == repo_name


@pytest.mark.parametrize('has_tag_manifest', [
  True,
  False,
])
def test_get_repo(has_tag_manifest, client, initialized_db):
  with client_with_identity('devtable', client) as cl:
    if not has_tag_manifest:
      database.TagManifestLabel.delete().execute()
      database.TagManifest.delete().execute()

    params = {'repository': 'devtable/simple'}
    response = conduct_api_call(cl, Repository, 'GET', params).json
    assert response['kind'] == 'image'


def test_get_app_repo(client, initialized_db):
  with client_with_identity('devtable', client) as cl:
    devtable = model.user.get_user('devtable')
    repo = model.repository.create_repository('devtable', 'someappr', devtable,
                                              repo_kind='application')

    models_ref = appr_model.models_ref
    blob.get_or_create_blob('sha256:somedigest', 0, 'application/vnd.cnr.blob.v0.tar+gzip',
                            ['local_us'], models_ref)

    release.create_app_release(repo, 'test',
                               dict(mediaType='application/vnd.cnr.package-manifest.helm.v0.json'),
                               'sha256:somedigest', models_ref, False)

    channel.create_or_update_channel(repo, 'somechannel', 'test', models_ref)

    params = {'repository': 'devtable/someappr'}
    response = conduct_api_call(cl, Repository, 'GET', params).json
    assert response['kind'] == 'application'
    assert response['channels']
    assert response['releases']