mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-16 15:45:04 +08:00
John Chilton
45019c3fbd
Split test/base into lib/galaxy_test/base and lib/galaxy_test/driver. Dividing things on whether they import galaxy-app code or not - with the goal of a galaxy-test-base package that doesn't depend on galaxy-app and a galaxy-test-driver package that does. I've been refactoring around this division for a long time. Certain tests (namely the integration tests) always need to bootstrap Galaxy - but the API tests and the Selenium tests should be installable and runnable without Galaxy in the Python environment. So future galaxy-test-api and galaxy-test-selenium packages should have only an optional dependency on galaxy-test-driver. Properly sturcturing these namespaces is a precursor for real packages on PyPI, clean separation of the tool shed testing code, and cleans up all the flake8 noqa hacks in test/.
42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
import tempfile
|
|
|
|
from galaxy_test.base.populators import DatasetPopulator
|
|
from galaxy_test.driver import integration_util
|
|
|
|
job_conf_yaml = """
|
|
runners:
|
|
local:
|
|
load: galaxy.jobs.runners.pulsar:PulsarRESTJobRunner
|
|
workers: 1
|
|
execution:
|
|
default: local
|
|
environments:
|
|
local:
|
|
runner: local
|
|
"""
|
|
|
|
|
|
class WorkQueuePutFailureTestCase(integration_util.IntegrationTestCase):
|
|
|
|
def setUp(self):
|
|
super(WorkQueuePutFailureTestCase, self).setUp()
|
|
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
|
|
self.history_id = self.dataset_populator.new_history()
|
|
|
|
@classmethod
|
|
def handle_galaxy_config_kwds(cls, config, ):
|
|
# config["jobs_directory"] = cls.jobs_directory
|
|
fd, path = tempfile.mkstemp(suffix='job_conf.yml')
|
|
with open(path, 'w') as job_conf:
|
|
job_conf.write(job_conf_yaml)
|
|
config["job_config_file"] = path
|
|
# Disable tool dependency resolution.
|
|
config["tool_dependency_dir"] = "none"
|
|
|
|
def test_job_fails(self):
|
|
self.dataset_populator.new_dataset(self.history_id, content="1 2 3")
|
|
self.dataset_populator.wait_for_history(self.history_id, assert_ok=False)
|
|
state_details = self.galaxy_interactor.get('histories/%s' % self.history_id).json()['state_details']
|
|
assert state_details['running'] == 0
|
|
assert state_details['error'] == 1
|