Initial unit test for copying test data.

lib/galaxy/tool_util/unittest_utils/__init__.py

@@ -1,3 +1,9 @@
+from typing import (
+    Callable,
+    Dict,
+    Optional,
+    Union,
+)
 from unittest.mock import Mock
 
 
@@ -9,3 +15,14 @@ def mock_trans(has_user=True, is_admin=False):
     else:
         trans.user = None
     return trans
+
+
+def t_data_downloader_for(content: Union[Dict[Optional[str], bytes], bytes]) -> Callable[[str], bytes]:
+    def get_content(filename: Optional[str]) -> bytes:
+        if isinstance(content, dict):
+            assert filename in content, f"failed to find {filename} in {content}"
+            return content[filename]
+        else:
+            return content
+
+    return get_content

lib/galaxy/tool_util/verify/interactor.py

@@ -127,6 +127,7 @@ def stage_data_in_history(
 
 class GalaxyInteractorApi:
     api_key: Optional[str]
+    keep_outputs_dir: Optional[str]
 
     def __init__(self, **kwds):
         self.api_url = f"{kwds['galaxy_url'].rstrip('/')}/api"
@@ -950,7 +951,12 @@ class RunToolException(Exception):
 
 # Galaxy specific methods - rest of this can be used with arbitrary files and such.
 def verify_hid(
-    filename, hda_id, attributes, test_data_downloader, hid="", dataset_fetcher=None, keep_outputs_dir=False
+    filename: Optional[str],
+    hda_id: str,
+    attributes: Dict[str, Any],
+    test_data_downloader,
+    dataset_fetcher=None,
+    keep_outputs_dir: Optional[str] = None,
 ):
     assert dataset_fetcher is not None
 
@@ -1050,7 +1056,7 @@ def _verify_composite_datatype_file_content(
     attributes=None,
     dataset_fetcher=None,
     test_data_downloader=None,
-    keep_outputs_dir=False,
+    keep_outputs_dir: Optional[str] = None,
     mode="file",
 ):
     assert dataset_fetcher is not None
@@ -1073,7 +1079,9 @@ def _verify_composite_datatype_file_content(
         raise AssertionError(errmsg)
 
 
-def _verify_extra_files_content(extra_files, hda_id, dataset_fetcher, test_data_downloader, keep_outputs_dir):
+def _verify_extra_files_content(
+    extra_files: List[Dict[str, Any]], hda_id: str, dataset_fetcher, test_data_downloader, keep_outputs_dir
+):
     files_list = []
     cleanup_directories = []
     for extra_file_dict in extra_files:

test/unit/tool_util/test_verify_function.py

@@ -0,0 +1,95 @@
+from galaxy.tool_util.unittest_utils import t_data_downloader_for
+from galaxy.tool_util.verify import verify
+
+
+def test_writes_output_to_target_directory(tmp_path):
+    item_label = "my test case"
+    output_content = b"expected"
+    filename = "my_output.txt"
+
+    assert not (tmp_path / filename).exists()
+    verify(
+        item_label,
+        output_content,
+        attributes={},
+        filename=filename,
+        keep_outputs_dir=str(tmp_path),
+        get_filecontent=t_data_downloader_for(output_content),
+    )
+    assert (tmp_path / filename).exists()
+    assert (tmp_path / filename).open("rb").read() == output_content
+
+
+def test_no_writes_output_to_target_directory_on_contains(tmp_path):
+    item_label = "my test case"
+    output_content = b"expected"
+    attributes = {
+        "compare": "contains",
+    }
+    filename = "my_output.txt"
+
+    get_filecontent = t_data_downloader_for(b"xpecte")
+    assert not (tmp_path / filename).exists()
+    verify(
+        item_label,
+        output_content,
+        attributes=attributes,
+        filename=filename,
+        keep_outputs_dir=str(tmp_path),
+        get_filecontent=get_filecontent,
+    )
+    # Consider switching this to not update - see discussion on
+    # https://github.com/galaxyproject/galaxy/pull/14661
+    assert (tmp_path / filename).exists()
+
+
+def test_sim_size(tmp_path):
+    item_label = "my test case"
+    output_content = b"expected"
+    attributes = {
+        "compare": "sim_size",
+        "delta": 2,
+    }
+    filename = "my_output.txt"
+
+    get_filecontent = t_data_downloader_for(b"xpected")
+    assert not (tmp_path / filename).exists()
+    verify(
+        item_label,
+        output_content,
+        attributes=attributes,
+        filename=filename,
+        keep_outputs_dir=str(tmp_path),
+        get_filecontent=get_filecontent,
+    )
+    assert (tmp_path / filename).exists()
+    assert (tmp_path / filename).open("rb").read() == b"expected"
+
+
+def test_sim_size_failure_still_updates(tmp_path):
+    item_label = "my test case"
+    output_content = b"expected"
+    attributes = {
+        "compare": "sim_size",
+        "delta": 2,
+    }
+    filename = "my_output.txt"
+
+    get_filecontent = t_data_downloader_for(b"ected")
+    assert not (tmp_path / filename).exists()
+    assertion_error = None
+    try:
+        verify(
+            item_label,
+            output_content,
+            attributes=attributes,
+            filename=filename,
+            keep_outputs_dir=str(tmp_path),
+            get_filecontent=get_filecontent,
+        )
+    except AssertionError as ae:
+        assertion_error = ae
+
+    assert assertion_error
+    assert (tmp_path / filename).exists()
+    assert (tmp_path / filename).open("rb").read() == b"expected"

test/unit/tool_util/test_verify_hid.py

@@ -0,0 +1,110 @@
+from typing import (
+    Callable,
+    Dict,
+    Optional,
+)
+
+from galaxy.tool_util.unittest_utils import t_data_downloader_for
+from galaxy.tool_util.verify.interactor import verify_hid
+
+
+def dataset_fetcher_for(
+    expected_hda_id: str, content: Dict[Optional[str], bytes]
+) -> Callable[[str, Optional[str]], bytes]:
+    def get_content(hda_id, filename: Optional[str] = None) -> bytes:
+        assert expected_hda_id == hda_id
+        return content[filename]
+
+    return get_content
+
+
+TEST_ID = "mydatasetid"
+
+
+def test_verify_hid_simple_pass():
+    test_data_downloader = t_data_downloader_for(b"expected")
+    dataset_fetcher = dataset_fetcher_for(TEST_ID, {None: b"expected"})
+    verify_hid(
+        "",
+        TEST_ID,
+        {},
+        dataset_fetcher=dataset_fetcher,
+        test_data_downloader=test_data_downloader,
+    )
+
+
+def test_verify_hid_simple_fail():
+    test_data_downloader = t_data_downloader_for(b"expected")
+    dataset_fetcher = dataset_fetcher_for(TEST_ID, {None: b"unexpected"})
+    assertion_error = None
+    try:
+        verify_hid(
+            "",
+            TEST_ID,
+            {},
+            dataset_fetcher=dataset_fetcher,
+            test_data_downloader=test_data_downloader,
+        )
+    except AssertionError as ae:
+        assertion_error = ae
+    assert assertion_error
+
+
+def test_verify_extra_files_success(tmp_path):
+    test_data_downloader = t_data_downloader_for(
+        {
+            "": b"expected",
+            "test_extras/expected.txt": b"extra_expected",
+        }
+    )
+    found_data = {
+        None: b"expected",
+        "extra_1": b"extra_expected",
+    }
+    extra_files = [{"type": "file", "name": "extra_1", "attributes": {}, "value": "test_extras/expected.txt"}]
+    dataset_fetcher = dataset_fetcher_for(TEST_ID, found_data)
+    assert not (tmp_path / "expected.txt").exists()
+    verify_hid(
+        "",
+        TEST_ID,
+        {"extra_files": extra_files},
+        dataset_fetcher=dataset_fetcher,
+        test_data_downloader=test_data_downloader,
+        keep_outputs_dir=tmp_path,
+    )
+    extra_path = tmp_path / "test_extras" / "expected.txt"
+    assert extra_path.exists()
+    assert extra_path.open("rb").read() == b"extra_expected"
+
+
+def test_verify_extra_files_failed(tmp_path):
+    test_data_downloader = t_data_downloader_for(
+        {
+            "": b"expected",
+            "test_extras/expected.txt": b"extra_expected",
+        }
+    )
+    found_data = {
+        None: b"expected",
+        "extra_1": b"extra_unexpected",
+    }
+    extra_files = [{"type": "file", "name": "extra_1", "attributes": {}, "value": "test_extras/expected.txt"}]
+    dataset_fetcher = dataset_fetcher_for(TEST_ID, found_data)
+    assert not (tmp_path / "expected.txt").exists()
+    assert_error = None
+    try:
+        verify_hid(
+            "",
+            TEST_ID,
+            {"extra_files": extra_files},
+            dataset_fetcher=dataset_fetcher,
+            test_data_downloader=test_data_downloader,
+            keep_outputs_dir=tmp_path,
+        )
+    except AssertionError as ae:
+        assert_error = ae
+    assert assert_error
+    assert "Composite file (extra_1) of History item mydatasetid different than expected" in str(assert_error)
+    extra_path = tmp_path / "test_extras" / "expected.txt"
+    assert extra_path.exists()
+    assert extra_path.open("rb").read() == b"extra_unexpected"