/
pbs_passing_environment_variable.py
211 lines (195 loc) · 8.5 KB
/
pbs_passing_environment_variable.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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
# coding: utf-8
# Copyright (C) 1994-2021 Altair Engineering, Inc.
# For more information, contact Altair at www.altair.com.
#
# This file is part of both the OpenPBS software ("OpenPBS")
# and the PBS Professional ("PBS Pro") software.
#
# Open Source License Information:
#
# OpenPBS is free software. You can redistribute it and/or modify it under
# the terms of the GNU Affero General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# OpenPBS 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 Affero General Public
# License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# Commercial License Information:
#
# PBS Pro is commercially licensed software that shares a common core with
# the OpenPBS software. For a copy of the commercial license terms and
# conditions, go to: (http://www.pbspro.com/agreement.html) or contact the
# Altair Legal Department.
#
# Altair's dual-license business model allows companies, individuals, and
# organizations to create proprietary derivative works of OpenPBS and
# distribute them - whether embedded or bundled with other software -
# under a commercial license agreement.
#
# Use of Altair's trademarks, including but not limited to "PBS™",
# "OpenPBS®", "PBS Professional®", and "PBS Pro™" and Altair's logos is
# subject to Altair's trademark licensing policies.
from tests.functional import *
class Test_passing_environment_variable_via_qsub(TestFunctional):
"""
Test to check passing environment variables via qsub
"""
def create_and_submit_job(self, user=None, attribs=None, content=None,
content_interactive=None, preserve_env=False):
"""
create the job object and submit it to the server as 'user',
attributes list 'attribs' script 'content' or 'content_interactive',
and to 'preserve_env' if interactive job.
"""
# A user=None value means job will be executed by current user
# where the environment is set up
if attribs is None:
use_attribs = {}
else:
use_attribs = attribs
retjob = Job(username=user, attrs=use_attribs)
if content is not None:
retjob.create_script(body=content)
elif content_interactive is not None:
retjob.interactive_script = content_interactive
retjob.preserve_env = preserve_env
return self.server.submit(retjob)
def test_commas_in_custom_variable(self):
"""
Submit a job with -v "var1='A,B,C,D'" and check that the value
is passed correctly
"""
a = {'Resource_List.select': '1:ncpus=1',
'Resource_List.walltime': 10}
script = ['#PBS -v "var1=\'A,B,C,D\'"']
script += ['env | grep var1']
jid = self.create_and_submit_job(user=TEST_USER, content=script,
attribs={ATTR_S: "/bin/bash"})
qstat = self.server.status(JOB, ATTR_o, id=jid)
job_outfile = qstat[0][ATTR_o].split(':')[1]
self.server.expect(JOB, 'queue', op=UNSET, id=jid, offset=10)
job_output = ""
ret = self.du.cat(self.server.client, filename=job_outfile,
runas=TEST_USER, logerr=False)
job_output = (' '.join(ret['out'])).strip()
self.assertEqual(job_output, "var1=A,B,C,D")
def test_passing_shell_function(self):
"""
Define a shell function with new line characters and check that
the function is passed correctly
"""
# Check if ShellShock fix for exporting shell function in bash exists
# on this system and what "BASH_FUNC_" format to use
foo_scr = """#!/bin/bash
foo() { a=B; echo $a; }
export -f foo
env | grep foo
unset -f foo
exit 0
"""
fn = self.du.create_temp_file(hostname=self.mom.hostname, body=foo_scr)
self.du.chmod(hostname=self.mom.hostname, path=fn, mode=0o755)
foo_msg = 'Failed to run foo_scr'
ret = self.du.run_cmd(self.mom.hostname, cmd=fn)
self.assertEqual(ret['rc'], 0, foo_msg)
msg = 'BASH_FUNC_'
n = 'foo'
for m in ret['out']:
if m.find(msg) != -1:
n = m.split('=')[0]
break
# Adjustments in bash due to ShellShock malware fix in various OS
script = """#!/bin/bash
foo() { if [ /bin/true ]; then\necho hello;\nfi\n}
export -f foo
#PBS -V
env | grep -A 3 foo\n
foo\n
"""
# Submit a job without hooks in the system
jid = self.create_and_submit_job(user=TEST_USER, content=script,
attribs={ATTR_S: "/bin/bash"})
qstat = self.server.status(JOB, ATTR_o, id=jid)
job_outfile = qstat[0][ATTR_o].split(':')[1]
self.server.expect(JOB, 'queue', op=UNSET, id=jid, offset=2)
job_output = ""
ret = self.du.cat(self.server.client, filename=job_outfile,
runas=TEST_USER, logerr=False)
job_output = ('\n'.join(ret['out'])).strip()
match = n + \
'=() { if [ /bin/true ]; then\n echo hello;\n fi\n}\nhello'
self.assertEqual(job_output, match,
msg="Environment variable foo content does "
"not match original")
def test_option_V_dfltqsubargs(self):
"""
Test exporting environment variable when -V is enabled
in default_qsub_arguments.
"""
os.environ["SET_IN_SUBMISSION"] = "true"
self.server.manager(MGR_CMD_SET, SERVER,
{'default_qsub_arguments': '-V'})
j = Job(self.du.get_current_user())
jid = self.server.submit(j)
self.server.expect(JOB, {'Variable_List': (MATCH_RE,
'SET_IN_SUBMISSION=true')},
id=jid)
def test_option_V_cmdline(self):
"""
Test exporting environment variable when -V is passed
through command line.
"""
os.environ["SET_IN_SUBMISSION"] = "true"
self.ATTR_V = 'Full_Variable_List'
api_to_cli.setdefault(self.ATTR_V, 'V')
a = {self.ATTR_V: None}
j = Job(self.du.get_current_user(), attrs=a)
jid = self.server.submit(j)
self.server.expect(JOB, {'Variable_List': (MATCH_RE,
'SET_IN_SUBMISSION=true')},
id=jid)
def test_option_V_dfltqsubargs_qsub_daemon(self):
"""
Test whether the changed value of the exported
environment variable is reflected if the submitted job
goes to qsub daemon.
"""
self.server.manager(MGR_CMD_SET, SERVER,
{'default_qsub_arguments': '-V'})
os.environ["SET_IN_SUBMISSION"] = "true"
j = Job(self.du.get_current_user())
jid = self.server.submit(j)
os.environ["SET_IN_SUBMISSION"] = "false"
j1 = Job(self.du.get_current_user())
jid1 = self.server.submit(j1)
self.server.expect(JOB, {'Variable_List': (MATCH_RE,
'SET_IN_SUBMISSION=true')},
id=jid)
self.server.expect(JOB, {'Variable_List': (MATCH_RE,
'SET_IN_SUBMISSION=false')},
id=jid1)
def test_passing_env_special_char_via_qsub(self):
"""
Submit a job with -v ENV_TEST=N:\\aa\\bb\\cc\\dd\\ee\\ff\\gg\\hh\\ii
and check that the value is passed correctly
NOTE: As per the Guide 5.2.4.7 Special Characters
in Variable_List Job Attribute
Python requires that double quotes
and backslashes also be escaped with a backslash
"""
a = {ATTR_v: 'ENV_TEST="N:\\aa\\bb\\cc\\dd\\ee\\ff\\gg\\hh\\ii"'}
j2 = Job(TEST_USER, attrs=a)
jid2 = self.server.submit(j2)
self.server.expect(JOB, {'job_state': 'R'}, id=jid2)
qstat = self.server.status(JOB, ATTR_v, id=jid2)
job_outfile = qstat[0]['Variable_List']
var_list = job_outfile.split(",")
exp_string = 'ENV_TEST=N:\\\\\\\\aa\\\\\\\\bb\\\\\\\\cc\\\\\\\\dd'
exp_string += '\\\\\\\\ee\\\\\\\\ff\\\\\\\\gg\\\\\\\\hh\\\\\\\\ii'
self.assertIn(exp_string, var_list)