mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-28 02:58:18 +08:00
Merge pull request #10222 from ic4f/dev_state5
Store job_id on Dataset table.
This commit is contained in:
@@ -0,0 +1,42 @@
|
||||
"""
|
||||
A Dataset should have a reference to the Job that created it.
|
||||
Here we test the creation of this reference in various contexts:
|
||||
- job creates one dataset
|
||||
- job creates multiple datasets
|
||||
- job uploads a dataset
|
||||
- job creates dynamically discovered datasets (after the tool is run)
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from galaxy import model
|
||||
from galaxy_test.driver.driver_util import GalaxyTestDriver
|
||||
|
||||
|
||||
class TestConfig:
|
||||
framework_tool_and_types = True
|
||||
|
||||
|
||||
TEST_TOOL_IDS = [
|
||||
'boolean_conditional', # 1 job >> 1 dataset
|
||||
'color_param', # 1 job >> multiple datasets
|
||||
'multi_output', # 1 job >> 1 dataset + 1 discovered dataset
|
||||
# test upload? # not implemented
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize('tool_id', TEST_TOOL_IDS)
|
||||
def test_tool_datasets(tool_id):
|
||||
driver = GalaxyTestDriver()
|
||||
driver.setup(TestConfig)
|
||||
driver.run_tool_test(tool_id)
|
||||
session = driver.app.model.context.current
|
||||
job = session.query(model.Job).order_by(model.Job.id.desc()).first()
|
||||
datasets = session.query(model.Dataset).filter(model.Dataset.job_id == job.id).all()
|
||||
|
||||
if tool_id == 'boolean_conditional':
|
||||
assert len(datasets) == 1
|
||||
elif tool_id in ('color_param', 'multi_output'):
|
||||
assert len(datasets) == 2
|
||||
|
||||
driver.tear_down()
|
||||
@@ -386,6 +386,15 @@ class MappingTests(BaseModelTestCase):
|
||||
assert d.metadata.anyAttribute is None
|
||||
assert 'items' not in d.metadata
|
||||
|
||||
def test_dataset_job_relationship(self):
|
||||
model = self.model
|
||||
dataset = model.Dataset()
|
||||
job = model.Job()
|
||||
dataset.job = job
|
||||
self.persist(job, dataset)
|
||||
loaded_dataset = model.session.query(model.Dataset).filter(model.Dataset.id == dataset.id).one()
|
||||
assert loaded_dataset.job_id == job.id
|
||||
|
||||
def test_jobs(self):
|
||||
model = self.model
|
||||
u = model.User(email="jobtest@foo.bar.baz", password="password")
|
||||
|
||||
Reference in New Issue
Block a user