Mercurial > repos > shellac > sam_consensus_v3
diff env/lib/python3.9/site-packages/cwltool/tests/test_secrets.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/env/lib/python3.9/site-packages/cwltool/tests/test_secrets.py Mon Mar 22 18:12:50 2021 +0000 @@ -0,0 +1,122 @@ +import shutil +import tempfile +from io import StringIO +from typing import Callable, Dict, List, Tuple, Union + +import pytest + +from cwltool.main import main +from cwltool.secrets import SecretStore +from cwltool.utils import CWLObjectType + +from .util import get_data, needs_docker, needs_singularity + + +@pytest.fixture +def secrets() -> Tuple[SecretStore, CWLObjectType]: + """Fixture to return a secret store.""" + sec_store = SecretStore() + job: CWLObjectType = {"foo": "bar", "baz": "quux"} + + sec_store.store(["foo"], job) + return sec_store, job + + +def test_obscuring(secrets: Tuple[SecretStore, CWLObjectType]) -> None: + """Basic test of secret store.""" + storage, obscured = secrets + assert obscured["foo"] != "bar" + assert obscured["baz"] == "quux" + result = storage.retrieve(obscured) + assert isinstance(result, dict) and result["foo"] == "bar" + + +obscured_factories_expected = [ + ((lambda x: "hello %s" % x), "hello bar"), + ((lambda x: ["echo", "hello %s" % x]), ["echo", "hello bar"]), + ((lambda x: {"foo": x}), {"foo": "bar"}), +] + + +@pytest.mark.parametrize("factory,expected", obscured_factories_expected) +def test_secrets( + factory: Callable[[str], CWLObjectType], + expected: Union[str, List[str], Dict[str, str]], + secrets: Tuple[SecretStore, CWLObjectType], +) -> None: + storage, obscured = secrets + obs = obscured["foo"] + assert isinstance(obs, str) + pattern = factory(obs) + assert pattern != expected + + assert storage.has_secret(pattern) + assert storage.retrieve(pattern) == expected + + assert obscured["foo"] != "bar" + assert obscured["baz"] == "quux" + + +@needs_docker +def test_secret_workflow_log() -> None: + stream = StringIO() + tmpdir = tempfile.mkdtemp() + main( + [ + "--debug", + "--enable-ext", + "--outdir", + tmpdir, + get_data("tests/wf/secret_wf.cwl"), + "--pw", + "Hoopla!", + ], + stderr=stream, + ) + + shutil.rmtree(tmpdir) + assert "Hoopla!" not in stream.getvalue() + + +@needs_singularity +def test_secret_workflow_log_singularity() -> None: + stream = StringIO() + tmpdir = tempfile.mkdtemp() + main( + [ + "--debug", + "--outdir", + tmpdir, + "--singularity", + get_data("tests/wf/secret_wf.cwl"), + "--pw", + "Hoopla!", + ], + stderr=stream, + ) + + shutil.rmtree(tmpdir) + assert "Hoopla!" not in stream.getvalue() + + +@needs_docker +def test_secret_workflow_log_override() -> None: + stream = StringIO() + tmpdir = tempfile.mkdtemp() + main( + [ + "--debug", + "--outdir", + tmpdir, + "--enable-ext", + "--overrides", + get_data("tests/wf/override-no-secrets.yml"), + get_data("tests/wf/secret_wf.cwl"), + "--pw", + "Hoopla!", + ], + stderr=stream, + ) + shutil.rmtree(tmpdir) + + assert "Hoopla!" in stream.getvalue()