Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

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.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_stack import ApplicationStack
from .celery_helper import rebind_container_to_task


# =============================================================================
[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 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={}, use_messaging=False, 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.file_sources = Bunch(to_dict=lambda *args, **kwargs: {}) self.interactivetool_manager = Bunch(create_interactivetool=lambda *args, **kwargs: None) self.is_job_handler = False self.biotools_metadata_source = None rebind_container_to_task(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.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