File: 062-sudo-g_flag_controller_only_systemd-v2.py

package info (click to toggle)
libcgroup 3.2.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 4,840 kB
  • sloc: ansic: 15,527; python: 8,219; cpp: 5,652; sh: 5,209; yacc: 470; makefile: 419; lex: 38
file content (171 lines) | stat: -rwxr-xr-x 5,666 bytes parent folder | download
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
#!/usr/bin/env python3
# SPDX-License-Identifier: LGPL-2.1-only
#
# Advanced cgget functionality test - '-b' '-g' <controller> (cgroup v2)
#
# Copyright (c) 2023 Oracle and/or its affiliates.
# Author: Kamalesh Babulal <kamalesh.babulal@oracle.com>
#

from cgroup import Cgroup, CgroupVersion
from systemd import Systemd
from run import RunError
import consts
import ftests
import sys
import os

CONTROLLER = 'cpu'
SYSTEMD_CGNAME = 'cg_in_scope'
OTHER_CGNAME = 'cg_not_in_scope'

SLICE = 'libcgtests.slice'
SCOPE = 'test062.scope'

CONFIG_FILE_NAME = os.path.join(os.getcwd(), '062cgconfig.conf')


def prereqs(config):
    result = consts.TEST_PASSED
    cause = None

    if CgroupVersion.get_version('cpu') != CgroupVersion.CGROUP_V2:
        result = consts.TEST_SKIPPED
        cause = 'This test requires the cgroup v2 cpu controller'
        return result, cause

    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):
    result = consts.TEST_PASSED
    cause = None

    pid = Systemd.write_config_with_pid(config, CONFIG_FILE_NAME, SLICE, SCOPE)

    Cgroup.configparser(config, load_file=CONFIG_FILE_NAME)

    # create and check if the cgroup was created under the systemd default path
    if not Cgroup.create_and_validate(config, None, SYSTEMD_CGNAME):
        result = consts.TEST_FAILED
        cause = (
                    'Failed to create systemd delegated cgroup {} under '
                    '/sys/fs/cgroup/{}/{}/'.format(SYSTEMD_CGNAME, SLICE, SCOPE)
                )
        return result, cause

    # With cgroup v2, we can't enable controller for the child cgroup, while
    # a task is attached to test062.scope. Attach the task from test062.scope
    # to child cgroup SYSTEMD_CGNAME and then enable cpu controller in the parent,
    # so that the cgroup.get() works
    Cgroup.set(config, cgname=SYSTEMD_CGNAME, setting='cgroup.procs', value=pid)

    Cgroup.set(
                config, cgname=(os.path.join(SLICE, SCOPE)), setting='cgroup.subtree_control',
                value='+cpu', ignore_systemd=True
              )

    # create and check if the cgroup was created under the controller root
    if not Cgroup.create_and_validate(config, CONTROLLER, OTHER_CGNAME, ignore_systemd=True):
        result = consts.TEST_FAILED
        cause = (
                    'Failed to create cgroup {} under '
                    '/sys/fs/cgroup/{}/'.format(OTHER_CGNAME, CONTROLLER)
                )

    return result, cause


def test(config):
    result = consts.TEST_PASSED
    cause = None

    out = Cgroup.get(config, controller=CONTROLLER, cgname=SYSTEMD_CGNAME)
    if len(out.splitlines()) < 10:
        # This cgget command gets all of the settings/values within the cgroup.
        # We don't care about the exact data, but there should be at least 10
        # lines of settings/values
        result = consts.TEST_FAILED
        cause = (
                    'cgget failed to read at least 10 lines from '
                    'cgroup {}: {}'.format(SYSTEMD_CGNAME, out)
                )

    out = Cgroup.get(config, controller=CONTROLLER, cgname=OTHER_CGNAME, ignore_systemd=True)
    if len(out.splitlines()) < 10:
        result = consts.TEST_FAILED
        tmp_cause = (
                        'cgget failed to read at least 10 lines from '
                        'cgroup {}: {}'.format(OTHER_CGNAME, out)
                    )
        cause = '\n'.join(filter(None, [cause, tmp_cause]))

    # This should fail because the wrong path should be built up
    out = Cgroup.get(config, controller=CONTROLLER, cgname=SYSTEMD_CGNAME,
                     ignore_systemd=True, print_headers=False)
    if len(out) > 0:
        result = consts.TEST_FAILED
        tmp_cause = (
                        'cgget erroneously read cgroup {} at the wrong '
                        'path: {}'.format(SYSTEMD_CGNAME, out)
                    )
        cause = '\n'.join(filter(None, [cause, tmp_cause]))

    # This should fail because the wrong path should be built up
    out = Cgroup.get(config, controller=CONTROLLER, cgname=OTHER_CGNAME, print_headers=False)
    if len(out) > 0:
        result = consts.TEST_FAILED
        cause = (
                    'cgget erroneously read cgroup {} at the wrong '
                    'path: {}'.format(OTHER_CGNAME, out)
                )
        cause = '\n'.join(filter(None, [cause, tmp_cause]))

    return result, cause


def teardown(config):
    Systemd.remove_scope_slice_conf(config, SLICE, SCOPE, CONTROLLER, CONFIG_FILE_NAME)

    # Incase the error occurs before the creation of OTHER_CGNAME,
    # let's ignore the exception
    try:
        Cgroup.delete(config, CONTROLLER, OTHER_CGNAME, ignore_systemd=True)
    except RunError as re:
        if 'No such file or directory' in re.stderr:
            raise re


def main(config):
    [result, cause] = prereqs(config)
    if result != consts.TEST_PASSED:
        return [result, cause]

    [result, cause] = setup(config)
    if result != consts.TEST_PASSED:
        teardown(config)
        return [result, cause]

    try:
        [result, cause] = test(config)
    finally:
        teardown(config)

    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: