diff --git a/lib/galaxy_test/api/test_history_contents.py b/lib/galaxy_test/api/test_history_contents.py index afa8ee74c038..25610246ce90 100644 --- a/lib/galaxy_test/api/test_history_contents.py +++ b/lib/galaxy_test/api/test_history_contents.py @@ -1,4 +1,6 @@ import json +import time +from datetime import datetime from requests import delete, put @@ -563,3 +565,59 @@ def test_job_state_summary_field(self): assert isinstance(c, dict) assert 'job_state_summary' in c assert isinstance(c['job_state_summary'], dict) + + def _get_content(self, history_id, update_time): + return self._get(f"/api/histories/{history_id}/contents/near/100/100?update_time-ge={update_time}").json() + + def test_history_contents_near_with_update_time(self): + with self.dataset_populator.test_history() as history_id: + first_time = datetime.utcnow().isoformat() + assert len(self._get_content(history_id, update_time=first_time)) == 0 + self.dataset_collection_populator.create_list_in_history(history_id=history_id) + assert len(self._get_content(history_id, update_time=first_time)) == 4 # 3 datasets + self.dataset_populator.wait_for_history(history_id) + all_datasets_finished = first_time = datetime.utcnow().isoformat() + assert len(self._get_content(history_id, update_time=all_datasets_finished)) == 0 + + @skip_without_tool('cat_data_and_sleep') + def test_history_contents_near_with_update_time_implicit_collection(self): + with self.dataset_populator.test_history() as history_id: + hdca_id = self.dataset_collection_populator.create_list_in_history(history_id=history_id).json()['id'] + self.dataset_populator.wait_for_history(history_id) + inputs = { + "input1": {'batch': True, 'values': [{"src": "hdca", "id": hdca_id}]}, + "sleep_time": 2, + } + response = self.dataset_populator.run_tool( + "cat_data_and_sleep", + inputs, + history_id, + assert_ok=False, + ).json() + collection_id = response['implicit_collections'][0]['id'] + for _ in range(20): + update_time = datetime.utcnow().isoformat() + time.sleep(1) + update = self._get_content(history_id, update_time=update_time) + if any((c for c in update if c['history_content_type'] == 'dataset_collection' and c['job_state_summary']['ok'] == 3)): + return + raise Exception(f"History content update time query did not include final update for implicit collection {collection_id}") + + @skip_without_tool('collection_creates_dynamic_nested') + def test_history_contents_near_with_update_time_explicit_collection(self): + with self.dataset_populator.test_history() as history_id: + inputs = {'foo': 'bar', 'sleep_time': 2} + response = self.dataset_populator.run_tool( + "collection_creates_dynamic_nested", + inputs, + history_id, + assert_ok=False, + ).json() + collection_id = response['output_collections'][0]['id'] + for _ in range(20): + update_time = datetime.utcnow().isoformat() + time.sleep(1) + update = self._get_content(history_id, update_time=update_time) + if any((c for c in update if c['history_content_type'] == 'dataset_collection' and c['populated_state'] == 'ok')): + return + raise Exception(f"History content update time query did not include populated_state update for dynamic nested collection {collection_id}")