import collections
import os
import unittest
import uuid

import pytest
from sqlalchemy import inspect

import galaxy.datatypes.registry
import galaxy.model
import galaxy.model.mapping as mapping
from galaxy.model.database_utils import create_database
from galaxy.model.security import GalaxyRBACAgent

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

DB_URI = "sqlite:///:memory:"
# docker run -e POSTGRES_USER=galaxy -p 5432:5432 -d postgres
# GALAXY_TEST_UNIT_MAPPING_URI_POSTGRES_BASE='postgresql://galaxy@localhost:5432/' pytest test/unit/data/test_galaxy_mapping.py
skip_if_not_postgres_base = pytest.mark.skipif(
    not os.environ.get('GALAXY_TEST_UNIT_MAPPING_URI_POSTGRES_BASE'),
    reason="GALAXY_TEST_UNIT_MAPPING_URI_POSTGRES_BASE not set"
)


class BaseModelTestCase(unittest.TestCase):

    @classmethod
    def _db_uri(cls):
        return DB_URI

    @classmethod
    def setUpClass(cls):
        # Start the database and connect the mapping
        cls.model = mapping.init("/tmp", cls._db_uri(), 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 MappingTests(BaseModelTestCase):

    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(), str)
            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='Hello₩◎ґʟⅾ'
        )

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

    def test_hda_to_library_dataset_dataset_association(self):
        u = self.model.User(email="mary@example.com", password="password")
        hda = self.model.HistoryDatasetAssociation(name='hda_name')
        self.persist(hda)
        trans = collections.namedtuple('trans', 'user')
        target_folder = self.model.LibraryFolder(name='library_folder')
        ldda = hda.to_library_dataset_dataset_association(
            trans=trans(user=u),
            target_folder=target_folder,
        )
        assert target_folder.item_count == 1
        assert ldda.id
        assert ldda.library_dataset.id
        assert ldda.library_dataset_id
        assert ldda.library_dataset.library_dataset_dataset_association
        assert ldda.library_dataset.library_dataset_dataset_association_id
        library_dataset_id = ldda.library_dataset_id
        replace_dataset = ldda.library_dataset
        new_ldda = hda.to_library_dataset_dataset_association(
            trans=trans(user=u),
            target_folder=target_folder,
            replace_dataset=replace_dataset
        )
        assert new_ldda.id != ldda.id
        assert new_ldda.library_dataset_id == library_dataset_id
        assert new_ldda.library_dataset.library_dataset_dataset_association_id == new_ldda.id
        assert len(new_ldda.library_dataset.expired_datasets) == 1
        assert new_ldda.library_dataset.expired_datasets[0] == ldda
        assert target_folder.item_count == 1

    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")
        library = 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, library, 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_nested_collection_attributes(self):
        model = self.model
        u = model.User(email="mary2@example.com", password="password")
        h1 = model.History(name="History 1", user=u)
        d1 = model.HistoryDatasetAssociation(extension="bam", 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='paired')
        dce1 = model.DatasetCollectionElement(collection=c1, element=d1, element_identifier="forward", element_index=0)
        dce2 = model.DatasetCollectionElement(collection=c1, element=d2, element_identifier="reverse", element_index=1)

        c2 = model.DatasetCollection(collection_type="list:paired")
        dce3 = model.DatasetCollectionElement(collection=c2, element=c1, element_identifier="inner_list", element_index=0)
        c3 = model.DatasetCollection(collection_type="list:list")
        c4 = model.DatasetCollection(collection_type="list:list:paired")
        dce4 = model.DatasetCollectionElement(collection=c4, element=c2, element_identifier="outer_list", element_index=0)
        model.session.add_all([d1, d2, c1, dce1, dce2, c2, dce3, c3, c4, dce4])
        model.session.flush()
        q = c2._get_nested_collection_attributes(element_attributes=('element_identifier',), hda_attributes=('extension',), dataset_attributes=('state',))
        assert [(r.keys()) for r in q] == [['element_identifier_0', 'element_identifier_1', 'extension', 'state'], ['element_identifier_0', 'element_identifier_1', 'extension', 'state']]
        assert q.all() == [('inner_list', 'forward', 'bam', 'new'), ('inner_list', 'reverse', 'txt', 'new')]
        q = c2._get_nested_collection_attributes(return_entities=(model.HistoryDatasetAssociation,))
        assert q.all() == [d1, d2]
        q = c2._get_nested_collection_attributes(return_entities=(model.HistoryDatasetAssociation, model.Dataset))
        assert q.all() == [(d1, d1.dataset), (d2, d2.dataset)]
        # Assert properties that use _get_nested_collection_attributes return correct content
        assert c2.dataset_instances == [d1, d2]
        assert c2.dataset_elements == [dce1, dce2]
        assert c2.dataset_action_tuples == []
        assert c2.populated_optimized
        assert c2.dataset_states_and_extensions_summary == ({'new'}, {'txt', 'bam'})
        assert c3.dataset_instances == []
        assert c3.dataset_elements == []
        assert c3.dataset_states_and_extensions_summary == (set(), set())
        q = c4._get_nested_collection_attributes(element_attributes=('element_identifier',))
        assert q.all() == [('outer_list', 'inner_list', 'forward'), ('outer_list', 'inner_list', 'reverse')]
        assert c4.dataset_elements == [dce1, dce2]

    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_metadata_spec(self):
        metadata = dict(chromCol=1, startCol=2, endCol=3)
        d = self.model.HistoryDatasetAssociation(extension="interval", metadata=metadata, sa_session=self.model.session)
        assert d.metadata.chromCol == 1
        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")
        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 test_history_audit(self):
        model = self.model
        u = model.User(email="contents@foo.bar.baz", password="password")
        h1 = model.History(name="HistoryAuditHistory", user=u)
        h2 = model.History(name="HistoryAuditHistory", user=u)

        def get_audit_table_entries(history):
            return self.session().query(model.HistoryAudit.table).filter(
                model.HistoryAudit.table.c.history_id == history.id).all()

        def get_latest_entry(entries):
            # key ensures result is correct if new columns are added
            return max(entries, key=lambda x: x.update_time)

        self.persist(u, h1, h2, expunge=False)
        assert len(get_audit_table_entries(h1)) == 1
        assert len(get_audit_table_entries(h2)) == 1

        self.new_hda(h1, name="1")
        self.new_hda(h2, name="2")
        self.session().flush()
        # db_next_hid modifies history, plus trigger on HDA means 2 additional audit rows per history

        h1_audits = get_audit_table_entries(h1)
        h2_audits = get_audit_table_entries(h2)
        assert len(h1_audits) == 3
        assert len(h2_audits) == 3

        h1_latest = get_latest_entry(h1_audits)
        h2_latest = get_latest_entry(h2_audits)

        model.HistoryAudit.prune(self.session())

        h1_audits = get_audit_table_entries(h1)
        h2_audits = get_audit_table_entries(h2)
        assert len(h1_audits) == 1
        assert len(h2_audits) == 1
        assert h1_audits[0] == h1_latest
        assert h2_audits[0] == h2_latest

    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)
        workflow_id = workflow.id

        annotation = model.WorkflowStepAnnotationAssociation()
        annotation.annotation = "Test Step Annotation"
        annotation.user = user
        annotation.workflow_step = workflow_step_1
        self.persist(annotation)

        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, f"{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

        loaded_workflow = self.query(model.Workflow).get(workflow_id)
        assert len(loaded_workflow.steps[0].annotations) == 1
        copied_workflow = loaded_workflow.copy(user=user)
        annotations = copied_workflow.steps[0].annotations
        assert len(annotations) == 1

    def test_role_creation(self):
        security_agent = GalaxyRBACAgent(self.model)

        def check_private_role(private_role, email):
            assert private_role.type == self.model.Role.types.PRIVATE
            assert len(private_role.users) == 1
            assert private_role.name == email
            assert private_role.description == "Private Role for " + email

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

        role = security_agent.get_private_user_role(u)
        assert role is None
        role = security_agent.create_private_user_role(u)
        assert role is not None
        check_private_role(role, email)

        email = "rule_user_2@example.com"
        u = self.model.User(email=email, password="password")
        self.persist(u)
        role = security_agent.get_private_user_role(u)
        assert role is None
        role = security_agent.get_private_user_role(u, auto_create=True)
        assert role is not None
        check_private_role(role, email)

        # make sure re-running auto_create doesn't break things
        role = security_agent.get_private_user_role(u, auto_create=True)
        assert role is not None
        check_private_role(role, email)

    def test_private_share_role(self):
        security_agent = GalaxyRBACAgent(self.model)

        u_from, u_to, u_other = self._three_users("private_share_role")

        h = self.model.History(name="History for Annotation", user=u_from)
        d1 = self.model.HistoryDatasetAssociation(extension="txt", history=h, create_dataset=True, sa_session=self.model.session)
        self.persist(h, d1)

        security_agent.privately_share_dataset(d1.dataset, [u_to])
        assert security_agent.can_access_dataset(u_to.all_roles(), d1.dataset)
        assert not security_agent.can_access_dataset(u_other.all_roles(), d1.dataset)

    def test_make_dataset_public(self):
        security_agent = GalaxyRBACAgent(self.model)
        u_from, u_to, u_other = self._three_users("make_dataset_public")

        h = self.model.History(name="History for Annotation", user=u_from)
        d1 = self.model.HistoryDatasetAssociation(extension="txt", history=h, create_dataset=True, sa_session=self.model.session)
        self.persist(h, d1)

        security_agent.privately_share_dataset(d1.dataset, [u_to])

        security_agent.make_dataset_public(d1.dataset)
        assert security_agent.can_access_dataset(u_to.all_roles(), d1.dataset)
        assert security_agent.can_access_dataset(u_other.all_roles(), d1.dataset)

    def test_set_all_dataset_permissions(self):
        security_agent = GalaxyRBACAgent(self.model)
        u_from, _, u_other = self._three_users("set_all_perms")

        h = self.model.History(name="History for Annotation", user=u_from)
        d1 = self.model.HistoryDatasetAssociation(extension="txt", history=h, create_dataset=True, sa_session=self.model.session)
        self.persist(h, d1)

        role = security_agent.get_private_user_role(u_from, auto_create=True)
        access_action = security_agent.permitted_actions.DATASET_ACCESS.action
        manage_action = security_agent.permitted_actions.DATASET_MANAGE_PERMISSIONS.action
        permissions = {access_action: [role], manage_action: [role]}
        assert security_agent.can_access_dataset(u_other.all_roles(), d1.dataset)
        security_agent.set_all_dataset_permissions(d1.dataset, permissions)
        assert not security_agent.allow_action(u_other.all_roles(), security_agent.permitted_actions.DATASET_ACCESS, d1.dataset)
        assert not security_agent.can_access_dataset(u_other.all_roles(), d1.dataset)

    def test_can_manage_privately_shared_dataset(self):
        security_agent = GalaxyRBACAgent(self.model)
        u_from, u_to, u_other = self._three_users("can_manage_dataset")

        h = self.model.History(name="History for Prevent Sharing", user=u_from)
        d1 = self.model.HistoryDatasetAssociation(extension="txt", history=h, create_dataset=True, sa_session=self.model.session)
        self.persist(h, d1)

        self._make_owned(security_agent, u_from, d1)
        assert security_agent.can_manage_dataset(u_from.all_roles(), d1.dataset)
        security_agent.privately_share_dataset(d1.dataset, [u_to])
        assert not security_agent.can_manage_dataset(u_to.all_roles(), d1.dataset)

    def test_can_manage_private_dataset(self):
        security_agent = GalaxyRBACAgent(self.model)
        u_from, _, u_other = self._three_users("can_manage_dataset_ps")

        h = self.model.History(name="History for Prevent Sharing", user=u_from)
        d1 = self.model.HistoryDatasetAssociation(extension="txt", history=h, create_dataset=True, sa_session=self.model.session)
        self.persist(h, d1)

        self._make_private(security_agent, u_from, d1)
        assert security_agent.can_manage_dataset(u_from.all_roles(), d1.dataset)
        assert not security_agent.can_manage_dataset(u_other.all_roles(), d1.dataset)

    def _three_users(self, suffix):
        email_from = f"user_{suffix}e1@example.com"
        email_to = f"user_{suffix}e2@example.com"
        email_other = f"user_{suffix}e3@example.com"

        u_from = self.model.User(email=email_from, password="password")
        u_to = self.model.User(email=email_to, password="password")
        u_other = self.model.User(email=email_other, password="password")
        self.persist(u_from, u_to, u_other)
        return u_from, u_to, u_other

    def _make_private(self, security_agent, user, hda):
        role = security_agent.get_private_user_role(user, auto_create=True)
        access_action = security_agent.permitted_actions.DATASET_ACCESS.action
        manage_action = security_agent.permitted_actions.DATASET_MANAGE_PERMISSIONS.action
        permissions = {access_action: [role], manage_action: [role]}
        security_agent.set_all_dataset_permissions(hda.dataset, permissions)

    def _make_owned(self, security_agent, user, hda):
        role = security_agent.get_private_user_role(user, auto_create=True)
        manage_action = security_agent.permitted_actions.DATASET_MANAGE_PERMISSIONS.action
        permissions = {manage_action: [role]}
        security_agent.set_all_dataset_permissions(hda.dataset, permissions)

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


@skip_if_not_postgres_base
class PostgresMappingTests(MappingTests):

    @classmethod
    def _db_uri(cls):
        base = os.environ.get("GALAXY_TEST_UNIT_MAPPING_URI_POSTGRES_BASE")
        dbname = "gxtest" + str(uuid.uuid4())
        postgres_url = base + dbname
        create_database(postgres_url)
        return postgres_url


class MockObjectStore:

    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_store_by(self, *args, **kwds):
        return 'id'


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