galaxy/test/unit/test_galaxy_mapping.py

# -*- coding: utf-8 -*-
import time
import unittest
import uuid

from six import text_type
from sqlalchemy import inspect

import galaxy.datatypes.registry
import galaxy.model
import galaxy.model.mapping as mapping

datatypes_registry = galaxy.datatypes.registry.Registry()
datatypes_registry.load_datatypes()
galaxy.model.set_datatypes_registry(datatypes_registry)


class MappingTests(unittest.TestCase):

    def test_annotations(self):
        model = self.model

        u = model.User(email="annotator@example.com", password="password")
        self.persist(u)

        def persist_and_check_annotation(annotation_class, **kwds):
            annotated_association = annotation_class()
            annotated_association.annotation = "Test Annotation"
            annotated_association.user = u
            for key, value in kwds.items():
                setattr(annotated_association, key, value)
            self.persist(annotated_association)
            self.expunge()
            stored_annotation = self.query(annotation_class).all()[0]
            assert stored_annotation.annotation == "Test Annotation"
            assert stored_annotation.user.email == "annotator@example.com"

        sw = model.StoredWorkflow()
        sw.user = u
        self.persist(sw)
        persist_and_check_annotation(model.StoredWorkflowAnnotationAssociation, stored_workflow=sw)

        workflow = model.Workflow()
        workflow.stored_workflow = sw
        self.persist(workflow)

        ws = model.WorkflowStep()
        ws.workflow = workflow
        self.persist(ws)
        persist_and_check_annotation(model.WorkflowStepAnnotationAssociation, workflow_step=ws)

        h = model.History(name="History for Annotation", user=u)
        self.persist(h)
        persist_and_check_annotation(model.HistoryAnnotationAssociation, history=h)

        d1 = model.HistoryDatasetAssociation(extension="txt", history=h, create_dataset=True, sa_session=model.session)
        self.persist(d1)
        persist_and_check_annotation(model.HistoryDatasetAssociationAnnotationAssociation, hda=d1)

        page = model.Page()
        page.user = u
        self.persist(page)
        persist_and_check_annotation(model.PageAnnotationAssociation, page=page)

        visualization = model.Visualization()
        visualization.user = u
        self.persist(visualization)
        persist_and_check_annotation(model.VisualizationAnnotationAssociation, visualization=visualization)

        dataset_collection = model.DatasetCollection(collection_type="paired")
        history_dataset_collection = model.HistoryDatasetCollectionAssociation(collection=dataset_collection)
        self.persist(history_dataset_collection)
        persist_and_check_annotation(model.HistoryDatasetCollectionAssociationAnnotationAssociation, history_dataset_collection=history_dataset_collection)

        library_dataset_collection = model.LibraryDatasetCollectionAssociation(collection=dataset_collection)
        self.persist(library_dataset_collection)
        persist_and_check_annotation(model.LibraryDatasetCollectionAnnotationAssociation, library_dataset_collection=library_dataset_collection)

    def test_ratings(self):
        model = self.model

        u = model.User(email="rater@example.com", password="password")
        self.persist(u)

        def persist_and_check_rating(rating_class, **kwds):
            rating_association = rating_class()
            rating_association.rating = 5
            rating_association.user = u
            for key, value in kwds.items():
                setattr(rating_association, key, value)
            self.persist(rating_association)
            self.expunge()
            stored_annotation = self.query(rating_class).all()[0]
            assert stored_annotation.rating == 5
            assert stored_annotation.user.email == "rater@example.com"

        sw = model.StoredWorkflow()
        sw.user = u
        self.persist(sw)
        persist_and_check_rating(model.StoredWorkflowRatingAssociation, stored_workflow=sw)

        h = model.History(name="History for Rating", user=u)
        self.persist(h)
        persist_and_check_rating(model.HistoryRatingAssociation, history=h)

        d1 = model.HistoryDatasetAssociation(extension="txt", history=h, create_dataset=True, sa_session=model.session)
        self.persist(d1)
        persist_and_check_rating(model.HistoryDatasetAssociationRatingAssociation, hda=d1)

        page = model.Page()
        page.user = u
        self.persist(page)
        persist_and_check_rating(model.PageRatingAssociation, page=page)

        visualization = model.Visualization()
        visualization.user = u
        self.persist(visualization)
        persist_and_check_rating(model.VisualizationRatingAssociation, visualization=visualization)

        dataset_collection = model.DatasetCollection(collection_type="paired")
        history_dataset_collection = model.HistoryDatasetCollectionAssociation(collection=dataset_collection)
        self.persist(history_dataset_collection)
        persist_and_check_rating(model.HistoryDatasetCollectionRatingAssociation, history_dataset_collection=history_dataset_collection)

        library_dataset_collection = model.LibraryDatasetCollectionAssociation(collection=dataset_collection)
        self.persist(library_dataset_collection)
        persist_and_check_rating(model.LibraryDatasetCollectionRatingAssociation, library_dataset_collection=library_dataset_collection)

    def test_display_name(self):

        def assert_display_name_converts_to_unicode(item, name):
            assert isinstance(item.get_display_name(), text_type)
            assert item.get_display_name() == name

        ldda = self.model.LibraryDatasetDatasetAssociation(name='ldda_name')
        assert_display_name_converts_to_unicode(ldda, 'ldda_name')

        hda = self.model.HistoryDatasetAssociation(name='hda_name')
        assert_display_name_converts_to_unicode(hda, 'hda_name')

        history = self.model.History(name='history_name')
        assert_display_name_converts_to_unicode(history, 'history_name')

        library = self.model.Library(name='library_name')
        assert_display_name_converts_to_unicode(library, 'library_name')

        library_folder = self.model.LibraryFolder(name='library_folder')
        assert_display_name_converts_to_unicode(library_folder, 'library_folder')

        history = self.model.History(
            name=u'Hello₩◎ґʟⅾ'
        )

        assert isinstance(history.name, text_type)
        assert isinstance(history.get_display_name(), text_type)
        assert history.get_display_name() == u'Hello₩◎ґʟⅾ'

    def test_tags(self):
        model = self.model

        my_tag = model.Tag(name="Test Tag")
        u = model.User(email="tagger@example.com", password="password")
        self.persist(my_tag, u)

        def tag_and_test(taggable_object, tag_association_class, backref_name):
            assert len(getattr(self.query(model.Tag).filter(model.Tag.name == "Test Tag").all()[0], backref_name)) == 0

            tag_association = tag_association_class()
            tag_association.tag = my_tag
            taggable_object.tags = [tag_association]
            self.persist(tag_association, taggable_object)

            assert len(getattr(self.query(model.Tag).filter(model.Tag.name == "Test Tag").all()[0], backref_name)) == 1

        sw = model.StoredWorkflow()
        sw.user = u
        tag_and_test(sw, model.StoredWorkflowTagAssociation, "tagged_workflows")

        h = model.History(name="History for Tagging", user=u)
        tag_and_test(h, model.HistoryTagAssociation, "tagged_histories")

        d1 = model.HistoryDatasetAssociation(extension="txt", history=h, create_dataset=True, sa_session=model.session)
        tag_and_test(d1, model.HistoryDatasetAssociationTagAssociation, "tagged_history_dataset_associations")

        page = model.Page()
        page.user = u
        tag_and_test(page, model.PageTagAssociation, "tagged_pages")

        visualization = model.Visualization()
        visualization.user = u
        tag_and_test(visualization, model.VisualizationTagAssociation, "tagged_visualizations")

        dataset_collection = model.DatasetCollection(collection_type="paired")
        history_dataset_collection = model.HistoryDatasetCollectionAssociation(collection=dataset_collection)
        tag_and_test(history_dataset_collection, model.HistoryDatasetCollectionTagAssociation, "tagged_history_dataset_collections")

        library_dataset_collection = model.LibraryDatasetCollectionAssociation(collection=dataset_collection)
        tag_and_test(library_dataset_collection, model.LibraryDatasetCollectionTagAssociation, "tagged_library_dataset_collections")

    def test_collections_in_histories(self):
        model = self.model

        u = model.User(email="mary@example.com", password="password")
        h1 = model.History(name="History 1", user=u)
        d1 = model.HistoryDatasetAssociation(extension="txt", history=h1, create_dataset=True, sa_session=model.session)
        d2 = model.HistoryDatasetAssociation(extension="txt", history=h1, create_dataset=True, sa_session=model.session)

        c1 = model.DatasetCollection(collection_type="pair")
        hc1 = model.HistoryDatasetCollectionAssociation(history=h1, collection=c1, name="HistoryCollectionTest1")

        dce1 = model.DatasetCollectionElement(collection=c1, element=d1, element_identifier="left")
        dce2 = model.DatasetCollectionElement(collection=c1, element=d2, element_identifier="right")

        self.persist(u, h1, d1, d2, c1, hc1, dce1, dce2)

        loaded_dataset_collection = self.query(model.HistoryDatasetCollectionAssociation).filter(model.HistoryDatasetCollectionAssociation.name == "HistoryCollectionTest1").first().collection
        self.assertEqual(len(loaded_dataset_collection.elements), 2)
        assert loaded_dataset_collection.collection_type == "pair"
        assert loaded_dataset_collection["left"] == dce1
        assert loaded_dataset_collection["right"] == dce2

    def test_collections_in_library_folders(self):
        model = self.model

        u = model.User(email="mary2@example.com", password="password")
        lf = model.LibraryFolder(name="RootFolder")
        l = model.Library(name="Library1", root_folder=lf)
        ld1 = model.LibraryDataset()
        ld2 = model.LibraryDataset()

        ldda1 = model.LibraryDatasetDatasetAssociation(extension="txt", library_dataset=ld1)
        ldda2 = model.LibraryDatasetDatasetAssociation(extension="txt", library_dataset=ld1)

        c1 = model.DatasetCollection(collection_type="pair")
        dce1 = model.DatasetCollectionElement(collection=c1, element=ldda1)
        dce2 = model.DatasetCollectionElement(collection=c1, element=ldda2)
        self.persist(u, l, lf, ld1, ld2, c1, ldda1, ldda2, dce1, dce2)

        # TODO:
        # loaded_dataset_collection = self.query( model.DatasetCollection ).filter( model.DatasetCollection.name == "LibraryCollectionTest1" ).first()
        # self.assertEqual(len(loaded_dataset_collection.datasets), 2)
        # assert loaded_dataset_collection.collection_type == "pair"

    def test_default_disk_usage(self):
        model = self.model

        u = model.User(email="disk_default@test.com", password="password")
        self.persist(u)
        u.adjust_total_disk_usage(1)
        u_id = u.id
        self.expunge()
        user_reload = model.session.query(model.User).get(u_id)
        assert user_reload.disk_usage == 1

    def test_basic(self):
        model = self.model

        original_user_count = len(model.session.query(model.User).all())

        # Make some changes and commit them
        u = model.User(email="james@foo.bar.baz", password="password")
        # gs = model.GalaxySession()
        h1 = model.History(name="History 1", user=u)
        # h1.queries.append( model.Query( "h1->q1" ) )
        # h1.queries.append( model.Query( "h1->q2" ) )
        h2 = model.History(name=("H" * 1024))
        self.persist(u, h1, h2)
        # q1 = model.Query( "h2->q1" )
        metadata = dict(chromCol=1, startCol=2, endCol=3)
        d1 = model.HistoryDatasetAssociation(extension="interval", metadata=metadata, history=h2, create_dataset=True, sa_session=model.session)
        # h2.queries.append( q1 )
        # h2.queries.append( model.Query( "h2->q2" ) )
        self.persist(d1)

        # Check
        users = model.session.query(model.User).all()
        assert len(users) == original_user_count + 1
        user = [user for user in users if user.email == "james@foo.bar.baz"][0]
        assert user.email == "james@foo.bar.baz"
        assert user.password == "password"
        assert len(user.histories) == 1
        assert user.histories[0].name == "History 1"
        hists = model.session.query(model.History).all()
        hist0 = [history for history in hists if history.name == "History 1"][0]
        hist1 = [history for history in hists if history.name == "H" * 255][0]
        assert hist0.name == "History 1"
        assert hist1.name == ("H" * 255)
        assert hist0.user == user
        assert hist1.user is None
        assert hist1.datasets[0].metadata.chromCol == 1
        # The filename test has moved to objectstore
        # id = hist1.datasets[0].id
        # assert hist1.datasets[0].file_name == os.path.join( "/tmp", *directory_hash_id( id ) ) + ( "/dataset_%d.dat" % id )
        # Do an update and check
        hist1.name = "History 2b"
        self.expunge()
        hists = model.session.query(model.History).all()
        hist0 = [history for history in hists if history.name == "History 1"][0]
        hist1 = [history for history in hists if history.name == "History 2b"][0]
        assert hist0.name == "History 1"
        assert hist1.name == "History 2b"
        # gvk TODO need to ad test for GalaxySessions, but not yet sure what they should look like.

    def test_jobs(self):
        model = self.model
        u = model.User(email="jobtest@foo.bar.baz", password="password")
        job = model.Job()
        job.user = u
        job.tool_id = "cat1"

        self.persist(u, job)

        loaded_job = model.session.query(model.Job).filter(model.Job.user == u).first()
        assert loaded_job.tool_id == "cat1"

    def test_job_metrics(self):
        model = self.model
        u = model.User(email="jobtest@foo.bar.baz", password="password")
        job = model.Job()
        job.user = u
        job.tool_id = "cat1"

        job.add_metric("gx", "galaxy_slots", 5)
        job.add_metric("system", "system_name", "localhost")

        self.persist(u, job)

        task = model.Task(job=job, working_directory="/tmp", prepare_files_cmd="split.sh")
        task.add_metric("gx", "galaxy_slots", 5)
        task.add_metric("system", "system_name", "localhost")

        big_value = ":".join(["%d" % i for i in range(2000)])
        task.add_metric("env", "BIG_PATH", big_value)
        self.persist(task)
        # Ensure big values truncated
        assert len(task.text_metrics[1].metric_value) <= 1023

    def test_tasks(self):
        model = self.model
        u = model.User(email="jobtest@foo.bar.baz", password="password")
        job = model.Job()
        task = model.Task(job=job, working_directory="/tmp", prepare_files_cmd="split.sh")
        job.user = u
        self.persist(u, job, task)

        loaded_task = model.session.query(model.Task).filter(model.Task.job == job).first()
        assert loaded_task.prepare_input_files_cmd == "split.sh"

    def test_history_contents(self):
        model = self.model
        u = model.User(email="contents@foo.bar.baz", password="password")
        # gs = model.GalaxySession()
        h1 = model.History(name="HistoryContentsHistory1", user=u)

        self.persist(u, h1, expunge=False)

        d1 = self.new_hda(h1, name="1")
        d2 = self.new_hda(h1, name="2", visible=False)
        d3 = self.new_hda(h1, name="3", deleted=True)
        d4 = self.new_hda(h1, name="4", visible=False, deleted=True)

        self.session().flush()

        def contents_iter_names(**kwds):
            history = model.context.query(model.History).filter(
                model.History.name == "HistoryContentsHistory1"
            ).first()
            return list(map(lambda hda: hda.name, history.contents_iter(**kwds)))

        self.assertEqual(contents_iter_names(), ["1", "2", "3", "4"])
        assert contents_iter_names(deleted=False) == ["1", "2"]
        assert contents_iter_names(visible=True) == ["1", "3"]
        assert contents_iter_names(visible=False) == ["2", "4"]
        assert contents_iter_names(deleted=True, visible=False) == ["4"]

        assert contents_iter_names(ids=[d1.id, d2.id, d3.id, d4.id]) == ["1", "2", "3", "4"]
        assert contents_iter_names(ids=[d1.id, d2.id, d3.id, d4.id], max_in_filter_length=1) == ["1", "2", "3", "4"]

        assert contents_iter_names(ids=[d1.id, d3.id]) == ["1", "3"]

    def _non_empty_flush(self):
        model = self.model
        lf = model.LibraryFolder(name="RootFolder")
        session = self.session()
        session.add(lf)
        session.flush()

    def test_flush_refreshes(self):
        # Normally I don't believe in unit testing library code, but the behaviors around attribute
        # states and flushing in SQL Alchemy is very subtle and it is good to have a executable
        # reference for how it behaves in the context of Galaxy objects.
        model = self.model
        user = model.User(
            email="testworkflows@bx.psu.edu",
            password="password"
        )
        galaxy_session = model.GalaxySession()
        galaxy_session_other = model.GalaxySession()
        galaxy_session.user = user
        galaxy_session_other.user = user
        self.persist(user, galaxy_session_other, galaxy_session)
        galaxy_session_id = galaxy_session.id

        self.expunge()
        session = self.session()
        galaxy_model_object = self.query(model.GalaxySession).get(galaxy_session_id)
        expected_id = galaxy_model_object.id

        # id loaded as part of the object query, could be any non-deferred attribute.
        assert 'id' not in inspect(galaxy_model_object).unloaded

        # Perform an empty flush, verify empty flush doesn't reload all attributes.
        session.flush()
        assert 'id' not in inspect(galaxy_model_object).unloaded

        # However, flushing anything non-empty - even unrelated object will invalidate
        # the session ID.
        self._non_empty_flush()
        assert 'id' in inspect(galaxy_model_object).unloaded

        # Fetch the ID loads the value from the database
        assert expected_id == galaxy_model_object.id
        assert 'id' not in inspect(galaxy_model_object).unloaded

        # Using cached_id instead does not exhibit this behavior.
        self._non_empty_flush()
        assert expected_id == galaxy.model.cached_id(galaxy_model_object)
        assert 'id' in inspect(galaxy_model_object).unloaded

        # Keeping the following failed experiments here for future reference,
        # I probed the internals of the attribute tracking and couldn't find an
        # alternative, generalized way to get the previously loaded value for unloaded
        # attributes.
        # print(galaxy_model_object._sa_instance_state.attrs.id)
        # print(dir(galaxy_model_object._sa_instance_state.attrs.id))
        # print(galaxy_model_object._sa_instance_state.attrs.id.loaded_value)
        # print(galaxy_model_object._sa_instance_state.attrs.id.state)
        # print(galaxy_model_object._sa_instance_state.attrs.id.load_history())
        # print(dir(galaxy_model_object._sa_instance_state.attrs.id.load_history()))
        # print(galaxy_model_object._sa_instance_state.identity)
        # print(dir(galaxy_model_object._sa_instance_state))
        # print(galaxy_model_object._sa_instance_state.expired_attributes)
        # print(galaxy_model_object._sa_instance_state.expired)
        # print(galaxy_model_object._sa_instance_state._instance_dict().keys())
        # print(dir(galaxy_model_object._sa_instance_state._instance_dict))
        # assert False

        # Verify cached_id works even immediately after an initial flush, prevents a second SELECT
        # query that would be executed if object.id was used.
        galaxy_model_object_new = model.GalaxySession()
        session.add(galaxy_model_object_new)
        session.flush()
        assert galaxy.model.cached_id(galaxy_model_object_new)
        assert 'id' in inspect(galaxy_model_object_new).unloaded

        # Verify a targeted flush prevent expiring unrelated objects.
        galaxy_model_object_new.id
        assert 'id' not in inspect(galaxy_model_object_new).unloaded
        session.flush(model.GalaxySession())
        assert 'id' not in inspect(galaxy_model_object_new).unloaded

    def test_workflows(self):
        model = self.model
        user = model.User(
            email="testworkflows@bx.psu.edu",
            password="password"
        )

        def workflow_from_steps(steps):
            stored_workflow = model.StoredWorkflow()
            stored_workflow.user = user
            workflow = model.Workflow()
            workflow.steps = steps
            workflow.stored_workflow = stored_workflow
            return workflow

        child_workflow = workflow_from_steps([])
        self.persist(child_workflow)

        workflow_step_1 = model.WorkflowStep()
        workflow_step_1.order_index = 0
        workflow_step_1.type = "data_input"
        workflow_step_2 = model.WorkflowStep()
        workflow_step_2.order_index = 1
        workflow_step_2.type = "subworkflow"
        workflow_step_2.subworkflow = child_workflow

        workflow_step_1.get_or_add_input("moo1")
        workflow_step_1.get_or_add_input("moo2")
        workflow_step_2.get_or_add_input("moo")
        workflow_step_1.add_connection("foo", "cow", workflow_step_2)

        workflow = workflow_from_steps([workflow_step_1, workflow_step_2])
        self.persist(workflow)

        assert workflow_step_1.id is not None
        h1 = model.History(name="WorkflowHistory1", user=user)

        invocation_uuid = uuid.uuid1()

        workflow_invocation = model.WorkflowInvocation()
        workflow_invocation.uuid = invocation_uuid
        workflow_invocation.history = h1

        workflow_invocation_step1 = model.WorkflowInvocationStep()
        workflow_invocation_step1.workflow_invocation = workflow_invocation
        workflow_invocation_step1.workflow_step = workflow_step_1

        subworkflow_invocation = model.WorkflowInvocation()
        workflow_invocation.attach_subworkflow_invocation_for_step(workflow_step_2, subworkflow_invocation)

        workflow_invocation_step2 = model.WorkflowInvocationStep()
        workflow_invocation_step2.workflow_invocation = workflow_invocation
        workflow_invocation_step2.workflow_step = workflow_step_2

        workflow_invocation.workflow = workflow

        d1 = self.new_hda(h1, name="1")
        workflow_request_dataset = model.WorkflowRequestToInputDatasetAssociation()
        workflow_request_dataset.workflow_invocation = workflow_invocation
        workflow_request_dataset.workflow_step = workflow_step_1
        workflow_request_dataset.dataset = d1
        self.persist(workflow_invocation)
        assert workflow_request_dataset is not None
        assert workflow_invocation.id is not None

        history_id = h1.id
        self.expunge()

        loaded_invocation = self.query(model.WorkflowInvocation).get(workflow_invocation.id)
        assert loaded_invocation.uuid == invocation_uuid, "%s != %s" % (loaded_invocation.uuid, invocation_uuid)
        assert loaded_invocation
        assert loaded_invocation.history.id == history_id

        step_1, step_2 = loaded_invocation.workflow.steps

        assert not step_1.subworkflow
        assert step_2.subworkflow
        assert len(loaded_invocation.steps) == 2

        subworkflow_invocation_assoc = loaded_invocation.get_subworkflow_invocation_association_for_step(step_2)
        assert subworkflow_invocation_assoc is not None
        assert isinstance(subworkflow_invocation_assoc.subworkflow_invocation, model.WorkflowInvocation)
        assert isinstance(subworkflow_invocation_assoc.parent_workflow_invocation, model.WorkflowInvocation)

        assert subworkflow_invocation_assoc.subworkflow_invocation.history.id == history_id

        u1 = loaded_invocation.update_time
        time.sleep(1)
        loaded_invocation.steps[0].update()
        self.expunge()
        loaded_invocation = self.query(model.WorkflowInvocation).get(workflow_invocation.id)
        u2 = loaded_invocation.update_time

        assert u1 != u2

    def new_hda(self, history, **kwds):
        return history.add_dataset(self.model.HistoryDatasetAssociation(create_dataset=True, sa_session=self.model.session, **kwds))

    @classmethod
    def setUpClass(cls):
        # Start the database and connect the mapping
        cls.model = mapping.init("/tmp", "sqlite:///:memory:", create_tables=True, object_store=MockObjectStore())
        assert cls.model.engine is not None

    @classmethod
    def query(cls, type):
        return cls.model.session.query(type)

    @classmethod
    def persist(cls, *args, **kwargs):
        session = cls.session()
        flush = kwargs.get('flush', True)
        for arg in args:
            session.add(arg)
            if flush:
                session.flush()
        if kwargs.get('expunge', not flush):
            cls.expunge()
        return arg  # Return last or only arg.

    @classmethod
    def session(cls):
        return cls.model.session

    @classmethod
    def expunge(cls):
        cls.model.session.flush()
        cls.model.session.expunge_all()


class MockObjectStore(object):

    def __init__(self):
        pass

    def size(self, dataset):
        return 42

    def exists(self, *args, **kwds):
        return True

    def get_filename(self, *args, **kwds):
        return "dataest_14.dat"


def get_suite():
    suite = unittest.TestSuite()
    suite.addTest(MappingTests("test_basic"))
    return suite