Merge pull request #18 from instituteofcancerresearch/update-tests

Update app utils tests

.github/workflows/dev_tests.yml

@@ -28,7 +28,7 @@ jobs:
           pip install ruff black
       - name: Lint check with ruff
         run: |
-          ruff --format=github --target-version=py311 --line-length 79 .
+          ruff --target-version=py311 --line-length 79 .
       - name: Style check with black
         run: |
           black ./ --check --line-length 79

src/SOPRANO/utils/app_utils.py

@@ -233,6 +233,9 @@ def release(release: str):
 
     @staticmethod
     def type(type_selection: str):
+        if type_selection not in ("toplevel", "primary_assembly"):
+            raise ValueError(type_selection)
+
         st.text(f"Selected: {type_selection}")
         return type_selection
 
@@ -286,8 +289,8 @@ def transcript_ids():
 
         with open(hla_binders_path, "r") as f:
             transcript_options = f.read()
-
-        return transcript_options.split("\n")
+        # eol marker generates empty so excluded
+        return transcript_options.split("\n")[:-1]
 
     @staticmethod
     def subset_method():

tests/test_units/_app_utils_fixtures.py

@@ -0,0 +1,11 @@
+import pytest
+
+from SOPRANO.utils.path_utils import Directories
+
+
+@pytest.fixture
+def mock_genome_dir():
+    assembly, release = "FOO", 999
+    mock_dir = Directories.data() / "homo_sapiens" / f"{release}_{assembly}"
+
+    return assembly, release, mock_dir

tests/test_units/conftest.py

@@ -1,3 +1,4 @@
+from _app_utils_fixtures import mock_genome_dir  # noqa: F401
 from _step2_fixtures import step_2_defs  # noqa: F401
 from _step3_fixtures import step_3_defs  # noqa: F401
 from _step6_fixtures import tcga_05_4396_ssb192_cfg  # noqa: F401

tests/test_units/test_app_utils.py

@@ -1,45 +1,268 @@
-from SOPRANO.utils.app_utils import PipelineUIOptions
-from SOPRANO.utils.path_utils import Directories
+import os
+from pathlib import Path
 
+import pytest
 
-def _check_detected(dir_method, options_method, ext):
-    tmp_name = f"pytest.{ext}"
-    other_name = "pytest.other"
+from SOPRANO.core.objects import GenomePaths
+from SOPRANO.utils.app_utils import (
+    DownloaderUIOptions,
+    DownloaderUIProcessing,
+    ImmunopeptidomesUIOptions,
+    ImmunopeptidomeUIProcessing,
+    LinkVEPUIProcessing,
+    PipelineUIOptions,
+    PipelineUIProcessing,
+    _select_from_dict,
+)
+from SOPRANO.utils.path_utils import _SOPRANO_DEFAULT_CACHE, Directories
 
-    tmp_path = dir_method(tmp_name)
-    other_path = dir_method(other_name)
-    tmp_path.touch(exist_ok=False)
-    other_path.touch(exist_ok=False)
 
-    try:
-        options = options_method()
-        assert tmp_name in options.keys()
-        assert options[tmp_name] == tmp_path
-        assert other_name not in options.keys()
-    finally:
-        tmp_path.unlink(missing_ok=False)
-        other_path.unlink(missing_ok=False)
+class TemporaryFiles:
+    def __init__(self, pass_ext, fail_ext, target_dir, mkdir, files):
+        self._pass_ext = pass_ext
+        self._fail_ext = fail_ext
+        self._target_dir = target_dir
+        self._mkdir = mkdir
+        self._files = files
+
+    @classmethod
+    def tmp_in_dir(cls, target_dir: Path, ext: str):
+        files = target_dir / f"pytest.{ext}", target_dir / "pytest.fail"
+        return cls(
+            pass_ext=files[0],
+            fail_ext=files[1],
+            target_dir=target_dir,
+            mkdir=False,
+            files=files,
+        )
+
+    @classmethod
+    def tmp_files(cls, target_dir: Path, *names: str, mkdir=False):
+        files = [target_dir / n for n in names]
+        return cls(
+            pass_ext=None,
+            fail_ext=None,
+            target_dir=target_dir,
+            mkdir=mkdir,
+            files=files,
+        )
+
+    def __enter__(self):
+        if self._mkdir:
+            self._target_dir.mkdir(exist_ok=False)
+
+        for f in self._files:
+            f.touch(exist_ok=False)
+
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        for f in self._files:
+            f.unlink(missing_ok=False)
+
+        if self._mkdir:
+            self._target_dir.rmdir()
 
 
-def test_get_annotated_input_options():
-    _check_detected(
-        Directories.app_annotated_inputs,
+def _check_options_generator(method, directory: Path, extension: str):
+    with TemporaryFiles.tmp_in_dir(directory, extension) as c:
+        options = method()
+        assert c._pass_ext.name in options
+        assert options[c._pass_ext.name] == c._pass_ext
+        assert c._fail_ext.name not in options
+        with pytest.raises(KeyError):
+            assert options[c._fail_ext.name] == c._fail_ext
+
+
+def test__select_from_dict():
+    k, v = "foo", "bar"
+    assert _select_from_dict(k, {k: v}) == v
+
+
+def test_pipeline_options_genome_reference(mock_genome_dir):
+    assembly, release, mock_dir = mock_genome_dir
+
+    with TemporaryFiles.tmp_files(
+        mock_dir,
+        "something.dna.toplevel.fa",
+        "something.dna.toplevel.chrom",
+        mkdir=True,
+    ):
+        options = PipelineUIOptions.genome_reference()
+        assert "{} - Ensembl release {}".format(assembly, release) in options
+
+
+def test_pipeline_options_annotated_mutations():
+    _check_options_generator(
         PipelineUIOptions.annotated_mutations,
+        Directories.app_annotated_inputs(),
         "anno",
     )
 
 
-def test_get_immunopeptidome_options():
-    _check_detected(
-        Directories.app_immunopeptidomes,
+def test_pipeline_options_immunopeptidome():
+    _check_options_generator(
         PipelineUIOptions.immunopeptidome,
+        Directories.app_immunopeptidomes(),
         "bed",
     )
 
 
-def test_get_coordinate_options():
-    _check_detected(
-        Directories.app_coordinate_files,
+def test_pipeline_options_substitution_method():
+    for v in (192, 7):
+        assert v in PipelineUIOptions.substitution_method().values()
+
+
+def test_pipeline_options_coordinates():
+    _check_options_generator(
         PipelineUIOptions.coordinates,
+        Directories.app_coordinate_files(),
         "bed",
     )
+
+
+def test_pipeline_processing_genome_reference(mock_genome_dir):
+    assembly, release, mock_dir = mock_genome_dir
+
+    selection_string = "{} - Ensembl release {}".format(assembly, release)
+
+    output = PipelineUIProcessing.genome_reference(selection_string)
+
+    assert isinstance(output, GenomePaths)
+    assert output.fasta.name.endswith(".fa")
+    assert output.sizes.name.endswith(".chrom")
+
+
+def test_pipeline_processing_annotated_mutations():
+    with TemporaryFiles.tmp_in_dir(
+        Directories.app_annotated_inputs(), "anno"
+    ) as c:
+        assert (
+            PipelineUIProcessing.annotated_mutations(c._pass_ext.name)
+            == c._pass_ext
+        )
+
+        with pytest.raises(KeyError):
+            PipelineUIProcessing.annotated_mutations(c._fail_ext.name)
+
+
+def test_pipeline_processing_immunopeptidome():
+    with TemporaryFiles.tmp_in_dir(
+        Directories.app_immunopeptidomes(), "bed"
+    ) as c:
+        assert (
+            PipelineUIProcessing.immunopeptidome(c._pass_ext.name)
+            == c._pass_ext
+        )
+
+        with pytest.raises(KeyError):
+            PipelineUIProcessing.immunopeptidome(c._fail_ext.name)
+
+
+def test_pipeline_processing_substitution_method():
+    assert PipelineUIProcessing.substitution_method("SSB192") == 192
+    assert PipelineUIProcessing.substitution_method("SSB7") == 7
+
+    with pytest.raises(KeyError):
+        PipelineUIProcessing.substitution_method("SSB000")
+
+
+def test_pipeline_processing_job_name():
+    cache_key = "SOPRANO_CACHE"
+    if cache_key in os.environ:
+        _soprano_cache = os.environ[cache_key]
+        del os.environ[cache_key]
+    else:
+        _soprano_cache = _SOPRANO_DEFAULT_CACHE.as_posix()
+    try:
+        mock_dir = Path("/just/for/pytest")
+        os.environ[cache_key] = mock_dir.as_posix()
+        job_name = "soprano"
+        assert PipelineUIProcessing.job_name(job_name) == mock_dir / job_name
+    finally:
+        os.environ[cache_key] = _soprano_cache
+
+
+def test_vep_processing_cache_location():
+    test_loc = Path.cwd()
+    assert LinkVEPUIProcessing.cache_location(test_loc.as_posix()) == test_loc
+
+
+def test_downloader_options_type():
+    options = DownloaderUIOptions.type()
+    assert "toplevel" in options
+    assert "primary_assembly" in options
+
+
+def test_downloader_processing_species():
+    assert DownloaderUIProcessing.species("Big Dog") == "big_dog"
+
+
+def test_downloader_processing_assembly():
+    assert DownloaderUIProcessing.assembly("cat") == "cat"
+
+
+def test_downloader_processing_release():
+    assert DownloaderUIProcessing.release("123") == 123
+
+
+def test_downloader_processing_type():
+    with pytest.raises(ValueError):
+        DownloaderUIProcessing.type("cat")
+
+    for permitted in ("primary_assembly", "toplevel"):
+        assert DownloaderUIProcessing.type(permitted) == permitted
+
+
+def test_immunopeptidome_options_hla_alleles():
+    options = ImmunopeptidomesUIOptions.hla_alleles()
+
+    assert isinstance(options, list)
+    assert len(options) == 354  # NOTE: Based on hard coded file... 13 Oct 2023
+
+
+def test_immunopeptidome_options_transcript_ids():
+    options = ImmunopeptidomesUIOptions.transcript_ids()
+
+    assert isinstance(options, list)
+    assert (
+        len(options) == 9754
+    )  # NOTE: Based on hard coded file... 13 Oct 2023
+
+
+def test_immunopeptidome_options_subset_method():
+    expected = {"None", "Retention", "Exclusion"}
+
+    assert set(ImmunopeptidomesUIOptions.subset_method()) == expected
+
+
+def test_immunopeptidome_processing_hla_alleles():
+    assert ImmunopeptidomeUIProcessing.hla_alleles([1, 2, 3]) == [1, 2, 3]
+
+
+def test_immunopeptidome_processing_transcript_ids():
+    assert ImmunopeptidomeUIProcessing.transcript_ids([1, 2, 3]) == [1, 2, 3]
+
+
+def test_immunopeptidome_processing_subset_method():
+    assert ImmunopeptidomeUIProcessing.subset_method([], "foo") == ([], [])
+    assert ImmunopeptidomeUIProcessing.subset_method([1, 2, 3], "None") == (
+        [],
+        [],
+    )
+
+    assert ImmunopeptidomeUIProcessing.subset_method([1, 2], "Retention") == (
+        [1, 2],
+        [],
+    )
+    assert ImmunopeptidomeUIProcessing.subset_method([1, 2], "Exclusion") == (
+        [],
+        [1, 2],
+    )
+    with pytest.raises(ValueError):
+        ImmunopeptidomeUIProcessing.subset_method([1, 2], "bar")
+
+
+def test_immunopeptidome_processing_name():
+    assert ImmunopeptidomeUIProcessing.name("x") == "x.bed"
+    assert ImmunopeptidomeUIProcessing.name("x.bed") == "x.bed"