quay/endpoints/appr/test/test_api.py

import os
import shutil
import uuid

import pytest

from cnr.models.db_base import CnrDB
from cnr.tests.test_apiserver import BaseTestServer
from cnr.tests.conftest import *
from cnr.tests.test_models import CnrTestModels
from peewee import SqliteDatabase

from app import app as application
from data.database import close_db_filter, User, db as database
from data.model import user, organization
from data.interfaces.appr import oci_app_model
import data.oci_model.blob as oci_blob
from endpoints.appr import appr_bp, registry
from endpoints.appr.cnr_backend import QuayDB, Channel, Package
from initdb import wipe_database, initialize_database, populate_database

application.register_blueprint(appr_bp, url_prefix='/cnr')


# TODO: avoid direct usage of database
def create_org(namespace, owner):
  try:
    User.get(username=namespace)
  except User.DoesNotExist:
    organization.create_organization(namespace, "%s@test.com" % str(uuid.uuid1()), owner)


class ChannelTest(Channel):
  @classmethod
  def dump_all(cls, package_class=None):
    result = []
    for repo in oci_app_model.list_applications(with_channels=True):
      for chan in repo.channels:
        result.append({'name': chan.name, 'current': chan.current, 'package': repo.name})
    return result


class PackageTest(Package):
  def _save(self, force, **kwargs):
    owner = user.get_user('devtable')
    create_org(self.namespace, owner)
    super(PackageTest, self)._save(force, user=owner, visibility="public")

  @classmethod
  def create_repository(cls, package_name, visibility, owner):
    ns, _ = package_name.split("/")
    owner = user.get_user('devtable')
    visibility = "public"
    create_org(ns, owner)
    return super(PackageTest, cls).create_repository(package_name, visibility, owner)

  @classmethod
  def dump_all(cls, blob_cls):
    result = []
    for repo in oci_app_model.list_applications(with_channels=True):
      package_name = repo.name
      for release in repo.releases:
        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'] = [x.name for x in oci_app_model.list_release_channels(package_name,
                                                                                  package.release,
                                                                                  False)]
          data['blob'] = blob.b64blob
          result.append(data)
    return result


@pytest.fixture(autouse=True)
def quaydb(monkeypatch):
  monkeypatch.setattr('endpoints.appr.cnr_backend.QuayDB.Package', PackageTest)
  monkeypatch.setattr('endpoints.appr.cnr_backend.Package', PackageTest)
  monkeypatch.setattr('endpoints.appr.registry.Package', PackageTest)
  monkeypatch.setattr('cnr.models.Package', PackageTest)

  monkeypatch.setattr('endpoints.appr.cnr_backend.QuayDB.Channel', ChannelTest)
  # monkeypatch.setattr('data.cnrmodel.channel.Channel', ChannelTest)
  monkeypatch.setattr('endpoints.appr.registry.Channel', ChannelTest)
  monkeypatch.setattr('cnr.models.Channel', ChannelTest)


def seed_db():
  create_org("titi", user.get_user("devtable"))


@pytest.fixture()
def sqlitedb_file(tmpdir):
  test_db_file = tmpdir.mkdir("quaydb").join("test.db")
  return str(test_db_file)


@pytest.fixture(scope="module")
def init_db_path(tmpdir_factory):
  sqlitedb_file_loc = str(tmpdir_factory.mktemp("data").join("test.db"))
  sqlitedb = 'sqlite:///{0}'.format(sqlitedb_file_loc)
  conf = {"TESTING": True,
          "DEBUG": True,
          "DB_URI": sqlitedb}
  os.environ['TEST_DATABASE_URI'] = str(sqlitedb)
  os.environ['DB_URI'] = str(sqlitedb)
  database.initialize(SqliteDatabase(sqlitedb_file_loc))
  application.config.update(conf)
  application.config.update({"DB_URI": sqlitedb})
  wipe_database()
  initialize_database()
  populate_database(minimal=True)
  close_db_filter(None)
  seed_db()
  return str(sqlitedb_file_loc)


@pytest.fixture()
def database_uri(monkeypatch, init_db_path, sqlitedb_file):
  shutil.copy2(init_db_path, sqlitedb_file)
  database.initialize(SqliteDatabase(sqlitedb_file))
  db_path = 'sqlite:///{0}'.format(sqlitedb_file)
  monkeypatch.setenv("DB_URI", db_path)
  seed_db()
  return db_path


@pytest.fixture()
def appconfig(database_uri):
  conf = {"TESTING": True,
          "DEBUG": True,
          "DB_URI": database_uri}
  return conf


@pytest.fixture()
def create_app():
  try:
    application.register_blueprint(appr_bp, url_prefix='')
  except:
    pass
  return application


@pytest.fixture(autouse=True)
def app(create_app, appconfig):
  create_app.config.update(appconfig)
  return create_app


@pytest.fixture()
def db():
  return CnrDB


class TestServerQuayDB(BaseTestServer):
  DB_CLASS = QuayDB

  @property
  def token(self):
    return "basic ZGV2dGFibGU6cGFzc3dvcmQ="

  def test_search_package_match(self, db_with_data1, client):
    """ TODO: search cross namespace and package name """
    BaseTestServer.test_search_package_match(self, db_with_data1, client)

  @pytest.mark.xfail
  def test_push_package_already_exists_force(self, db_with_data1, package_b64blob, client):
    """ No force push implemented """
    BaseTestServer.test_push_package_already_exists_force(self, db_with_data1, package_b64blob,
                                                          client)

  @pytest.mark.xfail
  def test_delete_channel_release_absent_release(self, db_with_data1, client):
    BaseTestServer.test_delete_channel_release_absent_release(self, db_with_data1, client)

  @pytest.mark.xfail
  def test_get_absent_blob(self, newdb, client):
    pass


class TestQuayModels(CnrTestModels):
  DB_CLASS = QuayDB

  @pytest.fixture(autouse=True)
  def load_db(self, appconfig):
    return appconfig

  @pytest.mark.xfail
  def test_save_package_exists_force(self, newdb, package_b64blob):
    CnrTestModels.test_save_package_exists_force(self, newdb, package_b64blob)

  @pytest.mark.xfail
  def test_channel_delete_releases(self, db_with_data1):
    """ Can't remove a release from the channel, only delete the channel entirely """
    CnrTestModels.test_channel_delete_releases(self, db_with_data1)

  @pytest.mark.xfail
  def test_forbiddeb_db_reset(self, db_class):
    pass

  @pytest.mark.xfail
  def test_db_restore(self, newdb, dbdata1):
    # This will fail as long as CNR tests use a mediatype with v1.
    pass

  def test_push_same_blob(self, db_with_data1):
    p = db_with_data1.Package.get("titi/rocketchat", ">1.2", 'kpm')
    assert p.package == "titi/rocketchat"
    assert p.release == "2.0.1"
    assert p.digest == "d3b54b7912fe770a61b59ab612a442eac52a8a5d8d05dbe92bf8f212d68aaa80"
    blob = db_with_data1.Blob.get("titi/rocketchat",
                                  p.digest)
    bdb = oci_blob.get_blob(p.digest)
    newblob = db_with_data1.Blob("titi/app2", blob.b64blob)
    p2 = db_with_data1.Package("titi/app2", "1.0.0", "helm", newblob)
    p2.save()
    b2db = oci_blob.get_blob(p2.digest)
    assert b2db.id == bdb.id