From cafde813221da5f088278ac09c8572e5ebc23a00 Mon Sep 17 00:00:00 2001 From: Jimmy Zelinskie Date: Wed, 22 Mar 2017 22:25:19 -0400 Subject: [PATCH] endpoints.appr.test: init --- endpoints/appr/test/test_api.py | 193 ++++++++++++++++++++++++++++++++ 1 file changed, 193 insertions(+) create mode 100644 endpoints/appr/test/test_api.py diff --git a/endpoints/appr/test/test_api.py b/endpoints/appr/test/test_api.py new file mode 100644 index 000000000..c013141fa --- /dev/null +++ b/endpoints/appr/test/test_api.py @@ -0,0 +1,193 @@ +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.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 +from endpoints.appr.cnr_backend import QuayDB, Channel, Package +from initdb import wipe_database, initialize_database, populate_database + + +# 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) + + +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_forbidden_db_reset(self, db_class): + pass