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 MockAppConfig(GalaxyDataTestConfig, CommonConfigurationMixin):
[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 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 fill_template(self, filename, template_lookup=None, **kwargs):
template = template_lookup.get_template(filename)
kwargs.update(h=MockTemplateHelpers())
return template.render(**kwargs)
[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)