Source code for galaxy.app_unittest_utils.galaxy_mock

"""
Mock infrastructure for testing ModelManagers.
"""
import os
import shutil
import tempfile
from typing import Any

from galaxy import (
    di,
    quota,
)
from galaxy.auth import AuthManager
from galaxy.celery import set_thread_app
from galaxy.config import CommonConfigurationMixin
from galaxy.jobs.manager import NoopManager
from galaxy.managers.users import UserManager
from galaxy.model import tags
from galaxy.model.base import SharedModelMapping
from galaxy.model.mapping import GalaxyModelMapping
from galaxy.model.scoped_session import galaxy_scoped_session
from galaxy.model.unittest_utils import (
    GalaxyDataTestApp,
    GalaxyDataTestConfig,
)
from galaxy.security import idencoding
from galaxy.structured_app import (
    BasicSharedApp,
    MinimalManagerApp,
    StructuredApp,
)
from galaxy.tool_util.deps.containers import NullContainerFinder
from galaxy.tools.data import ToolDataTableManager
from galaxy.util import StructuredExecutionTimer
from galaxy.util.bunch import Bunch
from galaxy.util.dbkeys import GenomeBuilds
from galaxy.web.short_term_storage import (
    ShortTermStorageAllocator,
    ShortTermStorageConfiguration,
    ShortTermStorageManager,
    ShortTermStorageMonitor,
)
from galaxy.web_stack import ApplicationStack


# =============================================================================
[docs]def buildMockEnviron(**kwargs):
    environ = {
        "CONTENT_LENGTH": "0",
        "CONTENT_TYPE": "",
        "HTTP_ACCEPT": "*/*",
        "HTTP_ACCEPT_ENCODING": "gzip, deflate",
        "HTTP_ACCEPT_LANGUAGE": "en-US,en;q=0.8,zh;q=0.5,ja;q=0.3",
        "HTTP_CACHE_CONTROL": "no-cache",
        "HTTP_CONNECTION": "keep-alive",
        "HTTP_DNT": "1",
        "HTTP_HOST": "localhost:8000",
        "HTTP_ORIGIN": "http://localhost:8000",
        "HTTP_PRAGMA": "no-cache",
        "HTTP_REFERER": "http://localhost:8000",
        "HTTP_USER_AGENT": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.10; rv:43.0) Gecko/20100101 Firefox/43.0",
        "PATH_INFO": "/",
        "QUERY_STRING": "",
        "REMOTE_ADDR": "127.0.0.1",
        "REQUEST_METHOD": "GET",
        "SCRIPT_NAME": "",
        "SERVER_NAME": "127.0.0.1",
        "SERVER_PORT": "8080",
        "SERVER_PROTOCOL": "HTTP/1.1",
    }
    environ.update(**kwargs)
    return environ


[docs]class MockApp(di.Container, GalaxyDataTestApp):
    config: "MockAppConfig"

[docs]    def __init__(self, config=None, **kwargs):
        super().__init__()
        config = config or MockAppConfig(**kwargs)
        GalaxyDataTestApp.__init__(self, config=config, **kwargs)
        self[BasicSharedApp] = self
        self[MinimalManagerApp] = self
        self[StructuredApp] = self
        self[idencoding.IdEncodingHelper] = self.security
        self.name = kwargs.get("name", "galaxy")
        self[SharedModelMapping] = self.model
        self[GalaxyModelMapping] = self.model
        sts_config = ShortTermStorageConfiguration(short_term_storage_directory=os.path.join(config.data_dir, "sts"))
        sts_manager = ShortTermStorageManager(sts_config)
        self[ShortTermStorageAllocator] = sts_manager
        self[ShortTermStorageMonitor] = sts_manager
        self[galaxy_scoped_session] = self.model.context
        self.visualizations_registry = MockVisualizationsRegistry()
        self.tag_handler = tags.GalaxyTagHandler(self.model.context)
        self[tags.GalaxyTagHandler] = self.tag_handler
        self.quota_agent = quota.DatabaseQuotaAgent(self.model)
        self.job_config = Bunch(dynamic_params=None, destinations={}, assign_handler=lambda *args, **kwargs: None)
        self.tool_data_tables = ToolDataTableManager(tool_data_path=self.config.tool_data_path)
        self.dataset_collections_service = None
        self.container_finder = NullContainerFinder()
        self._toolbox_lock = MockLock()
        self.tool_shed_registry = Bunch(tool_sheds={})
        self.genome_builds = GenomeBuilds(self)
        self.job_manager = NoopManager()
        self.application_stack = ApplicationStack()
        self.auth_manager = AuthManager(self.config)
        self.user_manager = UserManager(self)
        self.execution_timer_factory = Bunch(get_timer=StructuredExecutionTimer)
        self.interactivetool_manager = Bunch(create_interactivetool=lambda *args, **kwargs: None)
        self.is_job_handler = False
        self.biotools_metadata_source = None
        set_thread_app(self)

        def url_for(*args, **kwds):
            return "/mock/url"

        self.url_for = url_for

[docs]    def wait_for_toolbox_reload(self, toolbox):
        # TODO: If the tpm test case passes, does the operation really
        # need to wait.
        return True


[docs]class MockLock:
    def __enter__(self):
        pass

    def __exit__(self, type, value, traceback):
        pass


[docs]class MockAppConfig(GalaxyDataTestConfig, CommonConfigurationMixin):
[docs]    class MockSchema(Bunch):
        pass

[docs]    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.schema = self.MockSchema()
        self.use_remote_user = kwargs.get("use_remote_user", False)
        self.enable_celery_tasks = False
        self.tool_data_path = os.path.join(self.root, "tool-data")
        self.galaxy_data_manager_data_path = self.tool_data_path
        self.tool_dependency_dir = None
        self.metadata_strategy = "directory"

        self.user_activation_on = False
        self.new_user_dataset_access_role_default_private = False

        self.expose_dataset_path = True
        self.allow_user_dataset_purge = True
        self.allow_user_creation = True
        self.email_domain_allowlist_content = None
        self.email_domain_blocklist_content = None
        self.enable_old_display_applications = True
        self.redact_username_in_logs = False
        self.auth_config_file = "config/auth_conf.xml.sample"
        self.error_email_to = "admin@email.to"
        self.password_expiration_period = 0

        self.umask = 0o77
        self.flush_per_n_datasets = 0

        # Compliance related config
        self.redact_email_in_job_name = False

        # Follow two required by GenomeBuilds
        self.len_file_path = os.path.join("tool-data", "shared", "ucsc", "chrom")
        self.builds_file_path = os.path.join("tool-data", "shared", "ucsc", "builds.txt.sample")

        self.shed_tool_config_file = "config/shed_tool_conf.xml"
        self.shed_tool_config_file_set = False
        self.enable_beta_edam_toolbox = False
        self.preserve_python_environment = "always"
        self.enable_beta_gdpr = False

        self.version_major = "19.09"

        # set by MockDir
        self.enable_tool_document_cache = False
        self.tool_cache_data_dir = os.path.join(self.root, "tool_cache")
        self.delay_tool_initialization = True
        self.external_chown_script = None
        self.check_job_script_integrity = False
        self.check_job_script_integrity_count = 0
        self.check_job_script_integrity_sleep = 0

        self.default_panel_view = "default"
        self.panel_views_dir = ""
        self.panel_views = {}
        self.edam_panel_views = ""

        self.config_file = None

        self._admin_users = ""
        self.drmaa_external_runjob_script = None
        self.tool_secret = None
        self.track_jobs_in_database = False
        self.amqp_internal_connection = None
        self.tool_configs = []
        self.manage_dependency_relationships = False
        self.enable_tool_shed_check = False
        self.monitor_thread_join_timeout = 1
        self.integrated_tool_panel_config = None
        self.vault_config_file = kwargs.get("vault_config_file")
        self.max_discovered_files = 10000

    @property
    def config_dict(self):
        return self.dict()

    def __getattr__(self, name):
        # Handle the automatic [option]_set options: for tests, assume none are set
        if name == "is_set":
            return lambda x: False
        # Handle the automatic config file _set options
        if name.endswith("_file_set"):
            return False
        raise AttributeError(name)


[docs]class MockWebapp:
[docs]    def __init__(self, security: idencoding.IdEncodingHelper, **kwargs):
        self.name = kwargs.get("name", "galaxy")
        self.security = security


[docs]class MockTrans:
[docs]    def __init__(self, app=None, user=None, history=None, **kwargs):
        self.app = app or MockApp(**kwargs)
        self.model = self.app.model
        self.webapp = MockWebapp(self.app.security, **kwargs)
        self.sa_session = self.app.model.session
        self.workflow_building_mode = False
        self.error_message = None
        self.anonymous = False
        self.debug = True
        self.user_is_admin = True
        self.url_builder = None

        self.galaxy_session = None
        self.__user = user
        self.security = self.app.security
        self.history = history

        self.request: Any = Bunch(headers={}, body=None)
        self.response: Any = Bunch(headers={}, set_content_type=lambda i: None)

[docs]    def check_csrf_token(self, payload):
        pass

[docs]    def handle_user_login(self, user):
        pass

[docs]    def log_event(self, message):
        pass

[docs]    def get_user(self):
        if self.galaxy_session:
            return self.galaxy_session.user
        else:
            return self.__user

[docs]    def set_user(self, user):
        """Set the current user."""
        if self.galaxy_session:
            self.galaxy_session.user = user
            self.sa_session.add(self.galaxy_session)
            self.sa_session.flush()
        self.__user = user

    user = property(get_user, set_user)

[docs]    def get_history(self, **kwargs):
        return self.history

[docs]    def set_history(self, history):
        self.history = history

[docs]    def fill_template(self, filename, template_lookup=None, **kwargs):
        template = template_lookup.get_template(filename)
        kwargs.update(h=MockTemplateHelpers())
        return template.render(**kwargs)


[docs]class MockVisualizationsRegistry:
    BUILT_IN_VISUALIZATIONS = ["trackster"]

[docs]    def get_visualizations(self, trans, target):
        return []


[docs]class MockDir:
[docs]    def __init__(self, structure_dict, where=None):
        self.structure_dict = structure_dict
        self.create_root(structure_dict, where)

[docs]    def create_root(self, structure_dict, where=None):
        self.root_path = tempfile.mkdtemp(dir=where)
        self.create_structure(self.root_path, structure_dict)

[docs]    def create_structure(self, current_path, structure_dict):
        for k, v in structure_dict.items():
            # if value is string, create a file in the current path and write v as file contents
            if isinstance(v, str):
                self.create_file(os.path.join(current_path, k), v)
            # if it's a dict, create a dir here named k and recurse into it
            if isinstance(v, dict):
                subdir_path = os.path.join(current_path, k)
                os.mkdir(subdir_path)
                self.create_structure(subdir_path, v)

[docs]    def create_file(self, path, contents):
        with open(path, "w") as newfile:
            newfile.write(contents)

[docs]    def remove(self):
        shutil.rmtree(self.root_path)


[docs]class MockTemplateHelpers:
[docs]    def js(*js_files):
        pass

[docs]    def css(*css_files):
        pass