Merge pull request #17976 from mvdbeek/allow_compressed_text_preview_…

…in_get_content_as_text

[24.0] Fix get_content_as_text for compressed text datatypes

lib/galaxy/managers/hdas.py

@@ -68,6 +68,7 @@
     MinimalManagerApp,
     StructuredApp,
 )
+from galaxy.util.compression_utils import get_fileobj
 
 log = logging.getLogger(__name__)
 
@@ -303,11 +304,13 @@ def text_data(self, hda, preview=True):
         # For now, cannot get data from non-text datasets.
         if not isinstance(hda.datatype, datatypes.data.Text):
             return truncated, hda_data
-        if not os.path.exists(hda.get_file_name()):
+        file_path = hda.get_file_name()
+        if not os.path.exists(file_path):
             return truncated, hda_data
 
-        truncated = preview and os.stat(hda.get_file_name()).st_size > MAX_PEEK_SIZE
-        hda_data = open(hda.get_file_name()).read(MAX_PEEK_SIZE)
+        truncated = preview and os.stat(file_path).st_size > MAX_PEEK_SIZE
+        with get_fileobj(file_path) as fh:
+            hda_data = fh.read(MAX_PEEK_SIZE)
         return truncated, hda_data
 
     # .... annotatable

lib/galaxy_test/api/test_datasets.py

@@ -12,6 +12,7 @@
     one_hda_model_store_dict,
     TEST_SOURCE_URI,
 )
+from galaxy.tool_util.verify.test_data import TestDataResolver
 from galaxy.util.unittest_utils import skip_if_github_down
 from galaxy_test.base.api_asserts import assert_has_keys
 from galaxy_test.base.decorators import (
@@ -356,6 +357,15 @@ def test_get_content_as_text(self, history_id):
         self._assert_has_key(get_content_as_text_response.json(), "item_data")
         assert get_content_as_text_response.json().get("item_data") == contents
 
+    def test_get_content_as_text_with_compressed_text_data(self, history_id):
+        test_data_resolver = TestDataResolver()
+        with open(test_data_resolver.get_filename("1.fasta.gz"), mode="rb") as fh:
+            hda1 = self.dataset_populator.new_dataset(history_id, content=fh, ftype="fasta.gz", wait=True)
+        get_content_as_text_response = self._get(f"datasets/{hda1['id']}/get_content_as_text")
+        self._assert_status_code_is(get_content_as_text_response, 200)
+        self._assert_has_key(get_content_as_text_response.json(), "item_data")
+        assert ">hg17" in get_content_as_text_response.json().get("item_data")
+
     def test_anon_get_content_as_text(self, history_id):
         contents = "accessible data"
         hda1 = self.dataset_populator.new_dataset(history_id, content=contents, wait=True)