From 63c2ffd3bae58fc4d37edf6149b79cf51de8a37c Mon Sep 17 00:00:00 2001 From: John Chilton Date: Mon, 23 Jul 2018 13:52:34 -0400 Subject: [PATCH] Implement simple history copy unit tests. --- test/unit/test_model_copy.py | 134 +++++++++++++++++++++++++++++++++++ 1 file changed, 134 insertions(+) create mode 100644 test/unit/test_model_copy.py diff --git a/test/unit/test_model_copy.py b/test/unit/test_model_copy.py new file mode 100644 index 00000000000..5e52b2c4c11 --- /dev/null +++ b/test/unit/test_model_copy.py @@ -0,0 +1,134 @@ +import contextlib +import os +import threading + +import galaxy.datatypes.registry +import galaxy.model +import galaxy.model.mapping as mapping +from galaxy.model.metadata import MetadataTempFile +from galaxy.util import ExecutionTimer +from .test_objectstore import DISK_TEST_CONFIG, TestConfig + + +datatypes_registry = galaxy.datatypes.registry.Registry() +datatypes_registry.load_datatypes() +galaxy.model.set_datatypes_registry(datatypes_registry) + +NUM_DATASETS = 3 +NUM_COLLECTIONS = 1 +SLOW_QUERY_LOG_THRESHOLD = 1000 +INCLUDE_METADATA_FILE = True +THREAD_LOCAL_LOG = threading.local() + + +def test_history_dataset_copy(num_datasets=NUM_DATASETS, include_metadata_file=INCLUDE_METADATA_FILE): + with _setup_mapping_and_user() as (test_config, object_store, model, old_history): + for i in range(num_datasets): + hda_path = test_config.write("moo", "test_metadata_original_%d" % i) + _create_hda(model, object_store, old_history, hda_path, include_metadata_file=include_metadata_file) + + model.context.flush() + + history_copy_timer = ExecutionTimer() + new_history = old_history.copy(target_user=old_history.user) + print("history copied %s" % history_copy_timer) + assert new_history.name == "HistoryCopyHistory1" + assert new_history.user == old_history.user + for i, hda in enumerate(new_history.active_datasets): + assert hda.get_size() == 3 + if include_metadata_file: + _check_metadata_file(hda) + annotation_str = hda.get_item_annotation_str(model.context, old_history.user, hda) + assert annotation_str == "annotation #%d" % hda.hid, annotation_str + + +def test_history_collection_copy(list_size=NUM_DATASETS): + with _setup_mapping_and_user() as (test_config, object_store, model, old_history): + for i in range(NUM_COLLECTIONS): + hdas = [] + for i in range(list_size * 2): + hda_path = test_config.write("moo", "test_metadata_original_%d" % i) + hda = _create_hda(model, object_store, old_history, hda_path, visible=False, include_metadata_file=False) + hdas.append(hda) + + list_elements = [] + list_collection = model.DatasetCollection(collection_type="list:paired") + for j in range(list_size): + paired_collection = model.DatasetCollection(collection_type="paired") + forward_dce = model.DatasetCollectionElement(collection=paired_collection, element=hdas[j * 2]) + reverse_dce = model.DatasetCollectionElement(collection=paired_collection, element=hdas[j * 2 + 1]) + paired_collection.elements = [forward_dce, reverse_dce] + paired_collection_element = model.DatasetCollectionElement(collection=list_collection, element=paired_collection) + list_elements.append(paired_collection_element) + model.context.add_all([forward_dce, reverse_dce, paired_collection_element]) + list_collection.elements = list_elements + history_dataset_collection = model.HistoryDatasetCollectionAssociation(collection=list_collection) + history_dataset_collection.user = old_history.user + model.context.add(history_dataset_collection) + + model.context.flush() + old_history.add_dataset_collection(history_dataset_collection) + history_dataset_collection.add_item_annotation(model.context, old_history.user, history_dataset_collection, "annotation #%d" % history_dataset_collection.hid) + + model.context.flush() + annotation_str = history_dataset_collection.get_item_annotation_str(model.context, old_history.user, history_dataset_collection) + + # Saving magic SA invocations for detecting full flushes that may harm performance. + # from sqlalchemy import event + # @event.listens_for(model.context, "before_flush") + # def track_instances_before_flush(session, context, instances): + # if not instances: + # print("FULL FLUSH...") + # else: + # print("Flushing just %s" % instances) + + history_copy_timer = ExecutionTimer() + new_history = old_history.copy(target_user=old_history.user) + print("history copied %s" % history_copy_timer) + + for i, hda in enumerate(new_history.active_datasets): + assert hda.get_size() == 3 + annotation_str = hda.get_item_annotation_str(model.context, old_history.user, hda) + assert annotation_str == "annotation #%d" % hda.hid, annotation_str + + assert len(new_history.active_dataset_collections) == NUM_COLLECTIONS + for hdca in new_history.active_dataset_collections: + annotation_str = hdca.get_item_annotation_str(model.context, old_history.user, hdca) + assert annotation_str == "annotation #%d" % hdca.hid, annotation_str + + +@contextlib.contextmanager +def _setup_mapping_and_user(): + with TestConfig(DISK_TEST_CONFIG) as (test_config, object_store): + # Start the database and connect the mapping + model = mapping.init("/tmp", "sqlite:///:memory:", create_tables=True, object_store=object_store, slow_query_log_threshold=SLOW_QUERY_LOG_THRESHOLD, thread_local_log=THREAD_LOCAL_LOG) + + u = model.User(email="historycopy@example.com", password="password") + h1 = model.History(name="HistoryCopyHistory1", user=u) + model.context.add_all([u, h1]) + model.context.flush() + yield test_config, object_store, model, h1 + + +def _create_hda(model, object_store, history, path, visible=True, include_metadata_file=False): + hda = model.HistoryDatasetAssociation(extension="bam", create_dataset=True, sa_session=model.context) + hda.visible = visible + model.context.add(hda) + model.context.flush([hda]) + object_store.update_from_file(hda, file_name=path, create=True) + if include_metadata_file: + hda.metadata.from_JSON_dict(json_dict={"bam_index": MetadataTempFile.from_JSON({"kwds": {}, "filename": path})}) + _check_metadata_file(hda) + hda.set_size() + history.add_dataset(hda) + hda.add_item_annotation(model.context, history.user, hda, "annotation #%d" % hda.hid) + return hda + + +def _check_metadata_file(hda): + assert hda.metadata.bam_index.id + copied_index = hda.metadata.bam_index.file_name + assert os.path.exists(copied_index) + with open(copied_index, "r") as f: + assert f.read() == "moo" + assert copied_index.endswith("metadata_%d.dat" % hda.id)