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
|
#!/usr/bin/env python3
# SPDX-License-Identifier: LGPL-2.1-only
#
# set_default_systemd_cgroup functionality test using the python bindings
#
# Copyright (c) 2023 Oracle and/or its affiliates.
# Author: Kamalesh Babulal <kamalesh.babulal@oracle.com>
# Author: Tom Hromatka <tom.hromatka@oracle.com>
#
from libcgroup import Version, Cgroup, Mode
from cgroup import Cgroup as CgroupCli
from process import Process
from systemd import Systemd
import consts
import ftests
import time
import sys
import os
OTHER_CGNAME = '071_cg_not_in_scope'
SLICE = 'libcgtests.slice'
SCOPE = 'test071.scope'
CONTROLLER = 'cpu'
def prereqs(config):
result = consts.TEST_PASSED
cause = None
if config.args.container:
result = consts.TEST_SKIPPED
cause = 'This test cannot be run within a container'
if not Systemd.is_systemd_enabled():
result = consts.TEST_SKIPPED
cause = 'Systemd support not compiled in'
return result, cause
def setup(config):
pass
def test(config):
result = consts.TEST_PASSED
cause = None
#
# Test 1 - Ensure _set_default_systemd_cgroup() throws an exception if
# libcgroup doesn't set a default (slice/scope) cgroup path
#
try:
Cgroup._set_default_systemd_cgroup()
except RuntimeError as re:
if 'Failed to set' not in str(re):
result = consts.TEST_FAILED
cause = "Expected 'Failed to set' to be in the exception, " \
'received {}'.format(str(re))
else:
result = consts.TEST_FAILED
cause = '_set_default_systemd_cgroup() erroneously passed'
#
# Test 2 - write_default_systemd_scope() should succeed if the slice/scope
# are invalid and we're not setting it as the default. If we
# create a cgroup at this point, it should be created at the root
# cgroup level, and the default slice/scope should have no bearing
#
Cgroup.write_default_systemd_scope(SLICE, SCOPE, False)
pid = config.process.create_process(config)
cg = Cgroup(OTHER_CGNAME, Version.CGROUP_V2)
cg.add_controller(CONTROLLER)
cg.create()
Cgroup.move_process(pid, OTHER_CGNAME, CONTROLLER)
cg_pid = cg.get_processes()[0]
if pid != cg_pid:
result = consts.TEST_FAILED
tmp_cause = 'Expected pid {} to be in {} cgroup, but received pid {} ' \
'via python bindings instead'.format(pid, OTHER_CGNAME, cg_pid)
cause = '\n'.join(filter(None, [cause, tmp_cause]))
cli_pid = CgroupCli.get_pids_in_cgroup(config, OTHER_CGNAME, CONTROLLER)[0]
if pid != cli_pid:
result = consts.TEST_FAILED
tmp_cause = 'Expected pid {} to be in {} cgroup, but received pid {} ' \
'via CLI instead'.format(pid, OTHER_CGNAME, cli_pid)
cause = '\n'.join(filter(None, [cause, tmp_cause]))
Process.kill(config, pid)
# Allow the cgroup sync, on killed pid
time.sleep(0.5)
cg.delete()
#
# Test 3 - Write the slice/scope and attempt to set them as the default.
# This should fail because they haven't been created yet, and thus
# it's an invalid path
#
try:
Cgroup.write_default_systemd_scope(SLICE, SCOPE, True)
except RuntimeError as re:
if 'Failed to set' not in str(re):
result = consts.TEST_FAILED
tmp_cause = "Expected 'Failed to set' to be in the exception, " \
'received {}'.format(str(re))
cause = '\n'.join(filter(None, [cause, tmp_cause]))
else:
result = consts.TEST_FAILED
tmp_cause = 'write_default_systemd_scope() erroneously passed'
cause = '\n'.join(filter(None, [cause, tmp_cause]))
#
# Test 4 - Create a systemd scope and set it as the default. Everything
# should work properly in this case
#
pid = None
if Cgroup.cgroup_mode() != Mode.CGROUP_MODE_LEGACY:
pid = config.process.create_process(config)
Cgroup.create_scope(SCOPE, SLICE, pid=pid)
Cgroup.write_default_systemd_scope(SLICE, SCOPE)
cg = Cgroup('/', Version.CGROUP_V2)
cg_pid = cg.get_processes()[0]
if pid != cg_pid:
result = consts.TEST_FAILED
tmp_cause = "Expected pid {} to be in '/' cgroup, but received pid {} " \
'instead'.format(pid, cg_pid)
cause = '\n'.join(filter(None, [cause, tmp_cause]))
path = Cgroup.get_current_controller_path(pid)
if path != os.path.join('/', SLICE, SCOPE):
result = consts.TEST_FAILED
tmp_cause = 'Expected pid path to be: {}, but received path {} ' \
'instead'.format(os.path.join('/', SLICE, SCOPE), path)
cause = '\n'.join(filter(None, [cause, tmp_cause]))
cli_pid = CgroupCli.get_pids_in_cgroup(config, os.path.join(SLICE, SCOPE), CONTROLLER)[0]
if pid != cli_pid:
result = consts.TEST_FAILED
tmp_cause = 'Expected pid {} to be in {} cgroup, but received pid {} ' \
'via CLI instead'.format(pid, os.path.join(SLICE, SCOPE), cli_pid)
cause = '\n'.join(filter(None, [cause, tmp_cause]))
return result, cause, pid
def teardown(config, pid):
if pid:
Process.kill(config, pid)
# Give systemd a chance to remove the scope
time.sleep(0.5)
Cgroup.clear_default_systemd_scope()
try:
cg = Cgroup(SLICE, Version.CGROUP_V2)
cg.delete()
except RuntimeError:
pass
def main(config):
[result, cause] = prereqs(config)
if result != consts.TEST_PASSED:
return [result, cause]
setup(config)
try:
pid = None
[result, cause, pid] = test(config)
finally:
teardown(config, pid)
return [result, cause]
if __name__ == '__main__':
config = ftests.parse_args()
# this test was invoked directly. run only it
config.args.num = int(os.path.basename(__file__).split('-')[0])
sys.exit(ftests.main(config))
# vim: set et ts=4 sw=4:
|
