forked from reanahub/reana
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_cli.py
150 lines (126 loc) · 5.41 KB
/
test_cli.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
# -*- coding: utf-8 -*-
#
# This file is part of REANA.
# Copyright (C) 2018, 2019, 2020 CERN.
#
# REANA is free software; you can redistribute it and/or modify it
# under the terms of the MIT License; see LICENSE file for more details.
"""REANA CLI tests."""
from __future__ import absolute_import, print_function
import os
def test_shorten_component_name():
"""Tests for shorten_component_name()."""
from reana.cli import shorten_component_name
for (name_long, name_short) in (
('', ''),
('reana', 'reana'),
('reana-job-controller', 'r-j-controller'),
):
assert name_short == shorten_component_name(name_long)
def test_get_expected_output_filenames_for_example():
"""Tests for get_expected_output_filenames_for_example()."""
from reana.cli import get_expected_output_filenames_for_example
for (example, output) in (
('', ('plot.png',)),
('reana-demo-helloworld', ('greetings.txt',)),
('reana-demo-root6-roofit', ('plot.png',)),
('reana-demo-alice-lego-train-test-run', ('plot.pdf', )),
):
assert output == get_expected_output_filenames_for_example(example)
def test_get_expected_log_message_for_example():
"""Tests for get_expected_log_messages_for_example()."""
from reana.cli import get_expected_log_messages_for_example
for (example, output) in (
('', ('job:',)),
('reana-demo-helloworld', ('Parameters: inputfile=',)),
):
assert output == get_expected_log_messages_for_example(example)
def test_is_component_dockerised():
"""Tests for is_component_dockerised()."""
from reana.cli import is_component_dockerised
assert is_component_dockerised(
'reana') is False
def test_is_component_runnable_example():
"""Tests for is_component_runnable_example()."""
from reana.cli import is_component_runnable_example
assert is_component_runnable_example(
'reana') is False
def test_select_components():
"""Tests for select_components()."""
from reana.cli import select_components, \
REPO_LIST_ALL, REPO_LIST_CLIENT, REPO_LIST_CLUSTER
for (input_value, output_expected) in (
# regular operation:
(['reana-job-controller', ],
['reana-job-controller', ]),
(['reana-job-controller', 'reana', ],
['reana-job-controller', 'reana, ']),
# special value: '.'
(['.', ], [os.path.basename(os.getcwd()), ]),
# special value: 'CLUSTER'
(['CLUSTER', ], REPO_LIST_CLUSTER),
# special value: 'CLIENT'
(['CLIENT', ], REPO_LIST_CLIENT),
# special value: 'ALL'
(['ALL', ], REPO_LIST_ALL),
# bad values:
(['nonsense', ], []),
(['nonsense', 'reana', ], ['reana', ]),
# output uniqueness:
(['ALL', 'reana', ], REPO_LIST_ALL),
(['CLUSTER', 'reana', ], REPO_LIST_CLUSTER),
(['ALL', 'CLUSTER', 'reana'], REPO_LIST_ALL),
):
output_obtained = select_components(input_value)
assert output_obtained.sort() == output_expected.sort()
num_excluded = 2
exclude_components = REPO_LIST_CLUSTER[:num_excluded]
output_obtained = select_components(REPO_LIST_CLUSTER, exclude_components)
assert len(output_obtained) == (len(REPO_LIST_CLUSTER) - num_excluded)
assert not set(exclude_components).intersection(output_obtained)
def test_select_workflow_engines():
"""Tests for select_workflow_engines()."""
from reana.cli import select_workflow_engines
for (input_value, output_expected) in (
# regular workflow engines:
(['cwl', ], ['cwl', ]),
(['serial', ], ['serial', ]),
(['cwl', 'yadage', ], ['cwl', 'yadage, ']),
# bad values:
(['nonsense', ], []),
(['nonsense', 'cwl', ], ['cwl', ]),
# output uniqueness:
(['cwl', 'cwl', ], ['cwl', ]),
):
output_obtained = select_workflow_engines(input_value)
assert output_obtained.sort() == output_expected.sort()
def test_find_standard_component_name():
"""Tests for find_standard_component_name()."""
from reana.cli import find_standard_component_name
for (input_value, output_expected) in (
('reana', 'reana'),
('r-server', 'reana-server'),
('r-j-controller', 'reana-job-controller'),
('reana-ui', 'reana-ui')
):
output_obtained = find_standard_component_name(input_value)
assert output_obtained == output_expected
def test_uniqueness_of_short_names():
"""Test whether all shortened component names are unique."""
from reana.cli import shorten_component_name, REPO_LIST_ALL
short_names = []
for repo in REPO_LIST_ALL:
short_name = shorten_component_name(repo)
if short_name in short_names:
raise Exception('Found ')
short_names.append(short_name)
def test_construct_workflow_name():
"""Tests for construct_workflow_name()."""
from reana.cli import construct_workflow_name
for (input_value, output_expected) in (
(('reana', 'cwl'), 'reana-cwl'),
(('reana-demo-root6-roofit', 'yadage'), 'root6-roofit-yadage'),
):
output_obtained = construct_workflow_name(input_value[0],
input_value[1])
assert output_obtained == output_expected