Mercurial > repos > shellac > sam_consensus_v3
comparison 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 |
comparison
equal
deleted
inserted
replaced
-1:000000000000 | 0:4f3585e2f14b |
---|---|
1 import shutil | |
2 import tempfile | |
3 from io import StringIO | |
4 from typing import Callable, Dict, List, Tuple, Union | |
5 | |
6 import pytest | |
7 | |
8 from cwltool.main import main | |
9 from cwltool.secrets import SecretStore | |
10 from cwltool.utils import CWLObjectType | |
11 | |
12 from .util import get_data, needs_docker, needs_singularity | |
13 | |
14 | |
15 @pytest.fixture | |
16 def secrets() -> Tuple[SecretStore, CWLObjectType]: | |
17 """Fixture to return a secret store.""" | |
18 sec_store = SecretStore() | |
19 job: CWLObjectType = {"foo": "bar", "baz": "quux"} | |
20 | |
21 sec_store.store(["foo"], job) | |
22 return sec_store, job | |
23 | |
24 | |
25 def test_obscuring(secrets: Tuple[SecretStore, CWLObjectType]) -> None: | |
26 """Basic test of secret store.""" | |
27 storage, obscured = secrets | |
28 assert obscured["foo"] != "bar" | |
29 assert obscured["baz"] == "quux" | |
30 result = storage.retrieve(obscured) | |
31 assert isinstance(result, dict) and result["foo"] == "bar" | |
32 | |
33 | |
34 obscured_factories_expected = [ | |
35 ((lambda x: "hello %s" % x), "hello bar"), | |
36 ((lambda x: ["echo", "hello %s" % x]), ["echo", "hello bar"]), | |
37 ((lambda x: {"foo": x}), {"foo": "bar"}), | |
38 ] | |
39 | |
40 | |
41 @pytest.mark.parametrize("factory,expected", obscured_factories_expected) | |
42 def test_secrets( | |
43 factory: Callable[[str], CWLObjectType], | |
44 expected: Union[str, List[str], Dict[str, str]], | |
45 secrets: Tuple[SecretStore, CWLObjectType], | |
46 ) -> None: | |
47 storage, obscured = secrets | |
48 obs = obscured["foo"] | |
49 assert isinstance(obs, str) | |
50 pattern = factory(obs) | |
51 assert pattern != expected | |
52 | |
53 assert storage.has_secret(pattern) | |
54 assert storage.retrieve(pattern) == expected | |
55 | |
56 assert obscured["foo"] != "bar" | |
57 assert obscured["baz"] == "quux" | |
58 | |
59 | |
60 @needs_docker | |
61 def test_secret_workflow_log() -> None: | |
62 stream = StringIO() | |
63 tmpdir = tempfile.mkdtemp() | |
64 main( | |
65 [ | |
66 "--debug", | |
67 "--enable-ext", | |
68 "--outdir", | |
69 tmpdir, | |
70 get_data("tests/wf/secret_wf.cwl"), | |
71 "--pw", | |
72 "Hoopla!", | |
73 ], | |
74 stderr=stream, | |
75 ) | |
76 | |
77 shutil.rmtree(tmpdir) | |
78 assert "Hoopla!" not in stream.getvalue() | |
79 | |
80 | |
81 @needs_singularity | |
82 def test_secret_workflow_log_singularity() -> None: | |
83 stream = StringIO() | |
84 tmpdir = tempfile.mkdtemp() | |
85 main( | |
86 [ | |
87 "--debug", | |
88 "--outdir", | |
89 tmpdir, | |
90 "--singularity", | |
91 get_data("tests/wf/secret_wf.cwl"), | |
92 "--pw", | |
93 "Hoopla!", | |
94 ], | |
95 stderr=stream, | |
96 ) | |
97 | |
98 shutil.rmtree(tmpdir) | |
99 assert "Hoopla!" not in stream.getvalue() | |
100 | |
101 | |
102 @needs_docker | |
103 def test_secret_workflow_log_override() -> None: | |
104 stream = StringIO() | |
105 tmpdir = tempfile.mkdtemp() | |
106 main( | |
107 [ | |
108 "--debug", | |
109 "--outdir", | |
110 tmpdir, | |
111 "--enable-ext", | |
112 "--overrides", | |
113 get_data("tests/wf/override-no-secrets.yml"), | |
114 get_data("tests/wf/secret_wf.cwl"), | |
115 "--pw", | |
116 "Hoopla!", | |
117 ], | |
118 stderr=stream, | |
119 ) | |
120 shutil.rmtree(tmpdir) | |
121 | |
122 assert "Hoopla!" in stream.getvalue() |