205 lines
6.4 KiB
Python
205 lines
6.4 KiB
Python
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
|
|
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
|