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
|
# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
#
# 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/>.
"""Test the cylc.flow.remote module."""
import os
import pytest
from cylc.flow.remote import (
run_cmd, construct_rsync_over_ssh_cmd, construct_ssh_cmd
)
import cylc.flow
def test_run_cmd_stdin_str():
"""Test passing stdin as a string."""
proc = run_cmd(
['sed', 's/foo/bar/'],
stdin_str='1foo2',
capture_process=True
)
assert [s.strip() for s in proc.communicate()] == [
'1bar2',
''
]
def test_run_cmd_stdin_file(tmp_path):
"""Test passing stdin as a file."""
tmp_path = tmp_path / 'stdin'
with tmp_path.open('w+') as tmp_file:
tmp_file.write('1foo2')
tmp_file = tmp_path.open('rb')
proc = run_cmd(
['sed', 's/foo/bar/'],
stdin=tmp_file,
capture_process=True
)
assert [s.strip() for s in proc.communicate()] == [
'1bar2',
''
]
def test_construct_rsync_over_ssh_cmd():
"""Function against known good output.
"""
cmd, host = construct_rsync_over_ssh_cmd(
'/foo',
'/bar',
{
'rsync command': 'rsync command',
'hosts': ['miklegard'],
'ssh command': 'strange_ssh',
'selection': {'method': 'definition order'},
'name': 'testplat'
}
)
assert host == 'miklegard'
assert cmd == [
'rsync',
'command',
'--delete',
'--rsh=strange_ssh',
'--include=/.service/',
'--include=/.service/server.key',
'-a',
'--checksum',
'--out-format=%o %n%L',
'--no-t',
'--exclude=log',
'--exclude=share',
'--exclude=work',
'--include=/ana/***',
'--include=/app/***',
'--include=/bin/***',
'--include=/etc/***',
'--include=/lib/***',
'--exclude=*',
'/foo/',
'miklegard:/bar/',
]
def test_construct_ssh_cmd_forward_env(monkeypatch: pytest.MonkeyPatch):
""" Test for 'ssh forward environment variables'
"""
# Clear CYLC_* env vars as these will show up in the command
for env_var in os.environ:
if env_var.startswith('CYLC'):
monkeypatch.delenv(env_var)
host = 'example.com'
config = {
'ssh command': 'ssh',
'use login shell': None,
'cylc path': None,
'ssh forward environment variables': ['FOO', 'BAZ'],
}
# Variable isn't set, no change to command
expect = [
'ssh',
host,
'env',
f'CYLC_VERSION={cylc.flow.__version__}',
'cylc',
'play',
]
cmd = construct_ssh_cmd(['play'], config, host)
assert cmd == expect
# Variable is set, appears in `env` list
monkeypatch.setenv('FOO', 'BAR')
expect = [
'ssh',
host,
'env',
f'CYLC_VERSION={cylc.flow.__version__}',
'FOO=BAR',
'cylc',
'play',
]
cmd = construct_ssh_cmd(['play'], config, host)
assert cmd == expect
|
