Source code for galaxy.model.unittest_utils.data_app

"""Mock GalaxyApp exposing config + functionality required for galaxy-data package.

There is a more expansive MockApp in test/unit/unittest_utils - but it isn't packaged
and it has dependencies from across the app. This mock application and config is
more appropriate for testing galaxy-data functionality and will be included with
galaxy-data.
"""

import os
import shutil
import tempfile
from typing import Optional

from galaxy import (
    model,
    objectstore,
)
from galaxy.datatypes import registry
from galaxy.files import (
    ConfiguredFileSources,
    NullConfiguredFileSources,
)
from galaxy.model.mapping import (
    GalaxyModelMapping,
    init,
)
from galaxy.model.security import GalaxyRBACAgent
from galaxy.model.tags import GalaxyTagHandler
from galaxy.security.idencoding import IdEncodingHelper
from galaxy.util.bunch import Bunch

GALAXY_TEST_UNITTEST_SECRET = "6e46ed6483a833c100e68cc3f1d0dd76"
GALAXY_TEST_IN_MEMORY_DB_CONNECTION = "sqlite:///:memory:"


[docs]class GalaxyDataTestConfig(Bunch):
    """Minimal Galaxy mock config object that exposes and uses only what is needed for the galaxy-data package."""

    security: IdEncodingHelper
    database_connection: str
    root: str
    data_dir: str
    _remove_root: bool

[docs]    def __init__(self, root=None, **kwd):
        Bunch.__init__(self, **kwd)
        if not root:
            root = tempfile.mkdtemp()
            self._remove_root = True
        else:
            self._remove_root = False
        self.root = root
        self.data_dir = os.path.join(root, "database")

        self.security = IdEncodingHelper(id_secret=GALAXY_TEST_UNITTEST_SECRET)
        self.database_connection = kwd.get("database_connection", GALAXY_TEST_IN_MEMORY_DB_CONNECTION)

        # objectstore config values...
        self.object_store_config_file = ""
        self.object_store_config = None
        self.object_store = "disk"
        self.object_store_check_old_style = False
        self.object_store_cache_path = "/tmp/cache"
        self.object_store_store_by = "uuid"

        self.umask = os.umask(0o77)
        self.gid = os.getgid()
        # objectstore config directories...
        self.jobs_directory = os.path.join(self.data_dir, "jobs_directory")
        self.new_file_path = os.path.join(self.data_dir, "tmp")
        self.file_path = os.path.join(self.data_dir, "files")
        self.server_name = "main"
        self.enable_quotas = False
        self.user_library_import_symlink_allowlist = []
        self.fetch_url_allowlist_ips = []
        self.library_import_dir = None
        self.user_library_import_dir = None
        self.ftp_upload_dir = None
        self.ftp_upload_purge = False

    def __del__(self):
        if self._remove_root:
            shutil.rmtree(self.root)


[docs]class GalaxyDataTestApp:
    """Minimal Galaxy mock app object that exposes and uses only what is needed for the galaxy-data package."""

    security: IdEncodingHelper
    model: GalaxyModelMapping
    security_agent: GalaxyRBACAgent
    file_sources: ConfiguredFileSources = NullConfiguredFileSources()

[docs]    def __init__(self, config: Optional[GalaxyDataTestConfig] = None, **kwd):
        config = config or GalaxyDataTestConfig(**kwd)
        self.config = config
        self.security = config.security
        self.object_store = objectstore.build_object_store_from_config(self.config)
        self.model = init("/tmp", self.config.database_connection, create_tables=True, object_store=self.object_store)
        self.security_agent = self.model.security_agent
        self.tag_handler = GalaxyTagHandler(self.model.context)
        self.init_datatypes()

[docs]    def init_datatypes(self):
        datatypes_registry = registry.Registry()
        datatypes_registry.load_datatypes()
        model.set_datatypes_registry(datatypes_registry)
        datatypes_registry.set_external_metadata_tool = MockSetExternalTool()
        self.datatypes_registry = datatypes_registry


[docs]class MockSetExternalTool:
[docs]    def regenerate_imported_metadata_if_needed(self, *args, **kwds):
        pass