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
|
# Copyright 2014 Codethink Limited
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# =*= License: GPL-3+ =*=
import itertools
import unittest
import yarnlib
class ScenarioRunnerEnvironmentTests(unittest.TestCase):
def setUp(self):
self.cmdlog = []
self.scenario = yarnlib.Scenario('foo')
step = yarnlib.ScenarioStep('THEN', 'foo bar')
step.implementation = yarnlib.Implementation('THEN', r'foo (\S+)',
'echo foo $MATCH_1')
self.scenario.steps = [step]
def fake_cmdrunner(self, cmd, env=None, cwd=None):
self.cmdlog.append((cmd, env, cwd))
return 0, '', ''
def test_command_env(self):
sr = yarnlib.ScenarioRunner('', '/tmp', {'FOO': 'bar'},
cmdrunner=self.fake_cmdrunner)
sr.run_scenario(self.scenario, '/tmp/datadir', '/tmp/home')
cmd, cmd_env, cmd_cwd = self.cmdlog[0]
self.assertIn('FOO', cmd_env)
self.assertEqual(cmd_env['FOO'], 'bar')
self.assertEqual(cmd_env['SRCDIR'], '/tmp')
self.assertEqual(cmd_env['DATADIR'], '/tmp/datadir')
self.assertEqual(cmd_env['HOME'], '/tmp/home')
self.assertEqual(cmd_env['MATCH_1'], 'bar')
self.assertEqual('/tmp', cmd_cwd)
def test_command_args(self):
prelude = 'TESTVAR=foo\n'
impl_shell = self.scenario.steps[0].implementation.shell
sr = yarnlib.ScenarioRunner(prelude, '/tmp',
cmdrunner=self.fake_cmdrunner)
sr.run_scenario(self.scenario, '/tmp/datadir', '/tmp/home')
shell, opts, command = self.cmdlog[0][0]
self.assertTrue(shell.endswith('sh'))
self.assertTrue(command.startswith(prelude))
self.assertIn(impl_shell, command)
class ScenarioRunnerFlowTests(unittest.TestCase):
def setUp(self):
self.cmdlog = []
self.scenario_runner = (
yarnlib.ScenarioRunner('', 'srcdir',
cmdrunner=self.fake_cmdrunner))
for verb, cmd in itertools.product(('ASSUMING', 'THEN', 'FINALLY'),
('true', 'false')):
step = yarnlib.ScenarioStep(verb, cmd)
step.implementation = yarnlib.Implementation(verb, cmd, cmd)
setattr(self, '%s_%s' % (verb.lower(), cmd), step)
def fake_cmdrunner(self, cmd, *args, **kwargs):
self.cmdlog.append(cmd)
if 'false' in cmd[-1]:
return 1, '', ''
return 0, '', ''
def run_scenario(self, *steps):
scenario = yarnlib.Scenario('foo')
scenario.steps = steps
return self.scenario_runner.run_scenario(scenario, 'data', 'home')
def test_assuming_skips_remaining(self):
ok = self.run_scenario(
self.assuming_false,
self.then_true,
self.finally_true,
)
self.assertTrue(ok)
self.assertEqual(len(self.cmdlog), 1)
def test_cleanup_run_on_success(self):
ok = self.run_scenario(
self.assuming_true,
self.then_true,
self.finally_true,
)
self.assertTrue(ok)
self.assertEqual(len(self.cmdlog), 3)
def test_cleanup_run_on_failure(self):
ok = self.run_scenario(
self.then_false,
self.finally_true,
)
self.assertFalse(ok)
self.assertEqual(len(self.cmdlog), 2)
def test_skip_steps_after_failure(self):
ok = self.run_scenario(
self.then_false,
self.then_true,
)
self.assertFalse(ok)
self.assertEqual(len(self.cmdlog), 1)
def test_cleanup_failure(self):
ok = self.run_scenario(
self.then_true,
self.finally_false,
self.finally_true,
)
self.assertFalse(ok)
self.assertEqual(len(self.cmdlog), 2)
|
