| Total Complexity | 64 |
| Total Lines | 765 |
| Duplicated Lines | 5.75 % |
| Changes | 0 | ||
Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.
Common duplication problems, and corresponding solutions are:
Complex classes like test_rule_handling often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
| 1 | import pytest |
||
| 2 | import mock |
||
| 3 | from collections import defaultdict |
||
| 4 | |||
| 5 | from pyanaconda.modules.common.constants.objects import FIREWALL, DEVICE_TREE, BOOTLOADER |
||
| 6 | from pyanaconda.modules.common.constants.services import NETWORK, STORAGE, USERS |
||
| 7 | |||
| 8 | try: |
||
| 9 | from org_fedora_oscap import rule_handling, common |
||
| 10 | except ImportError as exc: |
||
| 11 | pytestmark = pytest.mark.skip( |
||
| 12 | "Unable to import modules, possibly due to bad version of Anaconda: {error}" |
||
| 13 | .format(error=str(exc))) |
||
| 14 | |||
| 15 | |||
| 16 | @pytest.fixture() |
||
| 17 | def part_rules(): |
||
| 18 | rules = rule_handling.PartRules() |
||
| 19 | rules.ensure_mount_point("/tmp") |
||
| 20 | return rules |
||
| 21 | |||
| 22 | |||
| 23 | # simple tests, shouldn't raise exceptions |
||
| 24 | def test_part_rules_getitem(part_rules): |
||
| 25 | part_rules["/tmp"] |
||
| 26 | |||
| 27 | |||
| 28 | def test_part_rules_setitem(part_rules): |
||
| 29 | rule = rule_handling.PartRule("/var/log") |
||
| 30 | part_rules["/var/log"] = rule |
||
| 31 | |||
| 32 | |||
| 33 | def test_part_rules_len(part_rules): |
||
| 34 | assert len(part_rules) == 1 |
||
| 35 | |||
| 36 | |||
| 37 | def test_part_rules_contains(part_rules): |
||
| 38 | assert "/tmp" in part_rules |
||
| 39 | |||
| 40 | |||
| 41 | def test_part_rules_delitem(part_rules): |
||
| 42 | del(part_rules["/tmp"]) |
||
| 43 | assert "/tmp" not in part_rules |
||
| 44 | |||
| 45 | |||
| 46 | @pytest.fixture() |
||
| 47 | def rule_data(): |
||
| 48 | return rule_handling.RuleData() |
||
| 49 | |||
| 50 | |||
| 51 | def test_rule_data_artificial(rule_data): |
||
| 52 | rule_data.new_rule(" part /tmp --mountoptions=nodev,noauto") |
||
| 53 | rule_data.new_rule("part /var/log ") |
||
| 54 | rule_data.new_rule(" passwd --minlen=14 ") |
||
| 55 | rule_data.new_rule("package --add=iptables") |
||
| 56 | rule_data.new_rule(" package --add=firewalld --remove=telnet") |
||
| 57 | rule_data.new_rule("package --remove=rlogin --remove=sshd") |
||
| 58 | rule_data.new_rule("bootloader --passwd") |
||
| 59 | |||
| 60 | # both partitions should appear in rule_data._part_rules |
||
| 61 | assert "/tmp" in rule_data._part_rules |
||
| 62 | assert "/var/log" in rule_data._part_rules |
||
| 63 | |||
| 64 | # mount options should be parsed |
||
| 65 | assert "nodev" in rule_data._part_rules["/tmp"]._mount_options |
||
| 66 | assert "noauto" in rule_data._part_rules["/tmp"]._mount_options |
||
| 67 | |||
| 68 | # no mount options for /var/log |
||
| 69 | assert not rule_data._part_rules["/var/log"]._mount_options |
||
| 70 | |||
| 71 | # minimal password length should be parsed and stored correctly |
||
| 72 | assert rule_data._passwd_rules._minlen == 14 |
||
| 73 | |||
| 74 | # packages should be parsed correctly |
||
| 75 | assert "iptables" in rule_data._package_rules._add_pkgs |
||
| 76 | assert "firewalld" in rule_data._package_rules._add_pkgs |
||
| 77 | assert "telnet" in rule_data._package_rules._remove_pkgs |
||
| 78 | assert "rlogin" in rule_data._package_rules._remove_pkgs |
||
| 79 | assert "sshd" in rule_data._package_rules._remove_pkgs |
||
| 80 | |||
| 81 | # bootloader should require password |
||
| 82 | assert rule_data._bootloader_rules._require_password |
||
| 83 | |||
| 84 | |||
| 85 | def test_rule_data_quoted_opt_values(rule_data): |
||
| 86 | rule_data.new_rule('part /tmp --mountoptions="nodev,noauto"') |
||
| 87 | |||
| 88 | assert "nodev" in rule_data._part_rules["/tmp"]._mount_options |
||
| 89 | assert "noauto" in rule_data._part_rules["/tmp"]._mount_options |
||
| 90 | assert '"' not in rule_data._part_rules["/tmp"]._mount_options |
||
| 91 | |||
| 92 | |||
| 93 | def test_rule_data_real_output(rule_data): |
||
| 94 | output = """ |
||
| 95 | part /tmp |
||
| 96 | |||
| 97 | part /tmp --mountoptions=nodev |
||
| 98 | """ |
||
| 99 | for line in output.splitlines(): |
||
| 100 | rule_data.new_rule(line) |
||
| 101 | |||
| 102 | assert "/tmp" in rule_data._part_rules |
||
| 103 | assert "nodev" in rule_data._part_rules["/tmp"]._mount_options |
||
| 104 | |||
| 105 | # should be stripped and merged |
||
| 106 | assert str(rule_data._part_rules) == "part /tmp --mountoptions=nodev" |
||
| 107 | |||
| 108 | |||
| 109 | @pytest.fixture() |
||
| 110 | def ksdata_mock(): |
||
| 111 | return mock.Mock() |
||
| 112 | |||
| 113 | |||
| 114 | @pytest.fixture() |
||
| 115 | def storage_mock(): |
||
| 116 | return mock.Mock() |
||
| 117 | |||
| 118 | |||
| 119 | # monkeypatch is a predefined fixture that is used to perform per-test case changes in the test env. |
||
| 120 | # Here, it mocks the Anaconda interface on the module level. For more info, see: |
||
| 121 | # https://docs.pytest.org/en/latest/monkeypatch.html#monkeypatching-mocking-modules-and-environments |
||
| 122 | @pytest.fixture() |
||
| 123 | def proxy_getter(monkeypatch): |
||
| 124 | proxies = defaultdict(mock.Mock) |
||
| 125 | |||
| 126 | def mock_get(service_name, object_path): |
||
| 127 | initialize = not proxies |
||
| 128 | proxy = proxies[(service_name, object_path)] |
||
| 129 | |||
| 130 | if initialize: |
||
| 131 | set_dbus_defaults() |
||
| 132 | |||
| 133 | return proxy |
||
| 134 | |||
| 135 | monkeypatch.setattr("pyanaconda.core.dbus.DBus.get_proxy", mock_get) |
||
| 136 | return mock_get |
||
| 137 | |||
| 138 | |||
| 139 | def set_dbus_defaults(): |
||
| 140 | network = NETWORK.get_proxy() |
||
| 141 | network.Connected.return_value = True |
||
| 142 | |||
| 143 | firewall = NETWORK.get_proxy(FIREWALL) |
||
| 144 | firewall.EnabledServices = [] |
||
| 145 | firewall.DisabledServices = [] |
||
| 146 | firewall.EnabledPorts = [] |
||
| 147 | firewall.Trusts = [] |
||
| 148 | |||
| 149 | device_tree = STORAGE.get_proxy(DEVICE_TREE) |
||
| 150 | device_tree.GetDeviceMountOptions.return_value = "defaults" |
||
| 151 | device_tree.GetMountPoints.return_value = {} |
||
| 152 | |||
| 153 | bootloader = STORAGE.get_proxy(BOOTLOADER) |
||
| 154 | bootloader.IsPasswordSet = False |
||
| 155 | |||
| 156 | users = USERS.get_proxy() |
||
| 157 | users.IsRootPasswordSet = True |
||
| 158 | users.IsRootPasswordCrypted = False |
||
| 159 | users.RootPassword = "anaconda" |
||
| 160 | |||
| 161 | |||
| 162 | def test_evaluation_existing_part_must_exist_rules( |
||
| 163 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 164 | rules = [ |
||
| 165 | "part /tmp", |
||
| 166 | "part /", |
||
| 167 | ] |
||
| 168 | for rule in rules: |
||
| 169 | rule_data.new_rule(rule) |
||
| 170 | |||
| 171 | device_tree_mock = STORAGE.get_proxy(DEVICE_TREE) |
||
| 172 | device_tree_mock.GetDeviceMountOptions.return_value = "defaults" |
||
| 173 | device_tree_mock.GetMountPoints.return_value = { |
||
| 174 | "/tmp": "/dev/sda1", |
||
| 175 | "/": "/dev/sda2", |
||
| 176 | } |
||
| 177 | |||
| 178 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 179 | |||
| 180 | # partitions exist --> no errors, warnings or additional info |
||
| 181 | assert not messages |
||
| 182 | |||
| 183 | # no additional mount options specified |
||
| 184 | device_tree_mock.SetDeviceMountOptions.assert_has_calls([ |
||
| 185 | mock.call("/dev/sda1", "defaults"), |
||
| 186 | mock.call("/dev/sda2", "defaults"), |
||
| 187 | ]) |
||
| 188 | |||
| 189 | |||
| 190 | def test_evaluation_nonexisting_part_must_exist( |
||
| 191 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 192 | rules = [ |
||
| 193 | "part /tmp", |
||
| 194 | "part /", |
||
| 195 | ] |
||
| 196 | for rule in rules: |
||
| 197 | rule_data.new_rule(rule) |
||
| 198 | |||
| 199 | device_tree_mock = STORAGE.get_proxy(DEVICE_TREE) |
||
| 200 | device_tree_mock.GetDeviceMountOptions.return_value = "defaults" |
||
| 201 | device_tree_mock.GetMountPoints.return_value = { |
||
| 202 | "/tmp": "/dev/sda1", |
||
| 203 | } |
||
| 204 | |||
| 205 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 206 | |||
| 207 | # / mount point missing --> one error |
||
| 208 | assert len(messages) == 1 |
||
| 209 | assert messages[0].type == common.MESSAGE_TYPE_FATAL |
||
| 210 | |||
| 211 | # error has to mention the mount point |
||
| 212 | assert "/" in messages[0].text |
||
| 213 | |||
| 214 | |||
| 215 | def get_messages_for_partition_rules( |
||
| 216 | rule_data, ksdata_mock, storage_mock, |
||
| 217 | rules, |
||
| 218 | messages_evaluation_count=1, |
||
| 219 | actual_mountpoints=("/tmp", "/"), |
||
| 220 | mount_options=None, |
||
| 221 | report_only=False, |
||
| 222 | ): |
||
| 223 | |||
| 224 | assert len(rules) == 2, \ |
||
| 225 | "We need rules for temp partition and root." |
||
| 226 | |||
| 227 | if mount_options is None: |
||
| 228 | mount_options = { |
||
| 229 | "/": "defaults", |
||
| 230 | "/tmp": "defaults", |
||
| 231 | } |
||
| 232 | |||
| 233 | for rule in rules: |
||
| 234 | rule_data.new_rule(rule) |
||
| 235 | |||
| 236 | # Map mount points to device names. |
||
| 237 | mount_points = {} |
||
| 238 | |||
| 239 | if "/tmp" in actual_mountpoints: |
||
| 240 | mount_points["/tmp"] = "/dev/sda1" |
||
| 241 | |||
| 242 | if "/" in actual_mountpoints: |
||
| 243 | mount_points["/"] = "/dev/sda2" |
||
| 244 | |||
| 245 | # Map device names to mount options. |
||
| 246 | mount_options = { |
||
| 247 | mount_points[mount_point]: mount_options[mount_point] |
||
| 248 | for mount_point in actual_mountpoints |
||
| 249 | } |
||
| 250 | |||
| 251 | # Mock the device tree proxy. |
||
| 252 | def get_device_mount_options(device_name): |
||
| 253 | return mount_options[device_name] |
||
| 254 | |||
| 255 | def set_device_mount_options(device_name, options): |
||
| 256 | mount_options[device_name] = options |
||
| 257 | |||
| 258 | device_tree_mock = STORAGE.get_proxy(DEVICE_TREE) |
||
| 259 | device_tree_mock.GetMountPoints.return_value = mount_points |
||
| 260 | device_tree_mock.GetDeviceMountOptions.side_effect = get_device_mount_options |
||
| 261 | device_tree_mock.SetDeviceMountOptions.side_effect = set_device_mount_options |
||
| 262 | |||
| 263 | messages = [] |
||
| 264 | for _ in range(messages_evaluation_count): |
||
| 265 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only) |
||
| 266 | |||
| 267 | return messages |
||
| 268 | |||
| 269 | |||
| 270 | def evaluation_add_mount_options( |
||
| 271 | rule_data, ksdata_mock, storage_mock, |
||
| 272 | messages_evaluation_count): |
||
| 273 | rules = [ |
||
| 274 | "part /tmp --mountoptions=defaults,nodev", |
||
| 275 | "part / --mountoptions=noauto", |
||
| 276 | ] |
||
| 277 | |||
| 278 | messages = get_messages_for_partition_rules( |
||
| 279 | rule_data, ksdata_mock, storage_mock, |
||
| 280 | rules, messages_evaluation_count) |
||
| 281 | |||
| 282 | # two mount options added --> two info messages |
||
| 283 | assert len(messages) == 2 |
||
| 284 | assert all(message.type == common.MESSAGE_TYPE_INFO for message in messages) |
||
| 285 | |||
| 286 | # newly added mount options should be mentioned in the messages |
||
| 287 | # together with their mount points |
||
| 288 | nodev_found = False |
||
| 289 | noauto_found = False |
||
| 290 | |||
| 291 | for message in messages: |
||
| 292 | if "'nodev'" in message.text: |
||
| 293 | assert "/tmp" in message.text |
||
| 294 | nodev_found = True |
||
| 295 | elif "'noauto'" in message.text: |
||
| 296 | assert "/" in message.text |
||
| 297 | noauto_found = True |
||
| 298 | |||
| 299 | assert all([nodev_found, noauto_found]) |
||
| 300 | |||
| 301 | device_tree_mock = STORAGE.get_proxy(DEVICE_TREE) |
||
| 302 | device_tree_mock.SetDeviceMountOptions.assert_has_calls([ |
||
| 303 | mock.call("/dev/sda1", "defaults,nodev"), |
||
| 304 | mock.call("/dev/sda2", "defaults,noauto"), |
||
| 305 | ]) |
||
| 306 | |||
| 307 | |||
| 308 | def test_evaluation_add_mount_options( |
||
| 309 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 310 | evaluation_add_mount_options(rule_data, ksdata_mock, storage_mock, 1) |
||
| 311 | |||
| 312 | |||
| 313 | def test_evaluation_add_mount_options_no_duplicates( |
||
| 314 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 315 | evaluation_add_mount_options(rule_data, ksdata_mock, storage_mock, 2) |
||
| 316 | |||
| 317 | |||
| 318 | def test_evaluation_add_mount_options_report_only( |
||
| 319 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 320 | rules = [ |
||
| 321 | "part /tmp --mountoptions=nodev", |
||
| 322 | "part / --mountoptions=noauto", |
||
| 323 | ] |
||
| 324 | messages = get_messages_for_partition_rules( |
||
| 325 | rule_data, ksdata_mock, storage_mock, |
||
| 326 | rules, 1, report_only=True) |
||
| 327 | |||
| 328 | # two mount options added --> two info messages |
||
| 329 | assert len(messages) == 2 |
||
| 330 | assert messages[0].type == common.MESSAGE_TYPE_INFO |
||
| 331 | assert messages[1].type == common.MESSAGE_TYPE_INFO |
||
| 332 | |||
| 333 | # newly added mount options should be mentioned in the messages |
||
| 334 | # together with their mount points |
||
| 335 | nodev_found = False |
||
| 336 | noauto_found = False |
||
| 337 | |||
| 338 | for message in messages: |
||
| 339 | if "'nodev'" in message.text: |
||
| 340 | assert "/tmp" in message.text |
||
| 341 | nodev_found = True |
||
| 342 | elif "'noauto'" in message.text: |
||
| 343 | assert "/" in message.text |
||
| 344 | noauto_found = True |
||
| 345 | |||
| 346 | assert all([nodev_found, noauto_found]) |
||
| 347 | |||
| 348 | # no changes should be made |
||
| 349 | device_tree_mock = STORAGE.get_proxy(DEVICE_TREE) |
||
| 350 | device_tree_mock.SetDeviceMountOptions.assert_not_called() |
||
| 351 | |||
| 352 | |||
| 353 | def test_evaluation_add_mount_option_prefix( |
||
| 354 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 355 | rules = [ |
||
| 356 | "part /tmp --mountoptions=nodev", |
||
| 357 | "part / --mountoptions=noauto", |
||
| 358 | ] |
||
| 359 | mount_options = { |
||
| 360 | "/": "defaults", |
||
| 361 | "/tmp": "defaults,nodevice", |
||
| 362 | } |
||
| 363 | messages = get_messages_for_partition_rules( |
||
| 364 | rule_data, ksdata_mock, storage_mock, |
||
| 365 | rules, mount_options=mount_options) |
||
| 366 | |||
| 367 | # two mount options added (even though it is a prefix of another one) |
||
| 368 | # --> two info messages |
||
| 369 | assert len(messages) == 2 |
||
| 370 | assert messages[0].type == common.MESSAGE_TYPE_INFO |
||
| 371 | assert messages[1].type == common.MESSAGE_TYPE_INFO |
||
| 372 | |||
| 373 | # the option should be added even though it is a prefix of another one |
||
| 374 | device_tree_mock = STORAGE.get_proxy(DEVICE_TREE) |
||
| 375 | device_tree_mock.SetDeviceMountOptions.assert_has_calls([ |
||
| 376 | mock.call("/dev/sda1", "defaults,nodevice,nodev"), |
||
| 377 | ]) |
||
| 378 | |||
| 379 | |||
| 380 | def test_evaluation_add_mount_options_nonexisting_part( |
||
| 381 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 382 | rules = [ |
||
| 383 | "part /tmp --mountoptions=nodev", |
||
| 384 | "part / --mountoptions=noauto", |
||
| 385 | ] |
||
| 386 | messages = get_messages_for_partition_rules( |
||
| 387 | rule_data, ksdata_mock, storage_mock, |
||
| 388 | rules, actual_mountpoints=["/"]) |
||
| 389 | |||
| 390 | # one mount option added, one mount point missing (mount options |
||
| 391 | # cannot be added) --> one info, one error |
||
| 392 | assert len(messages) == 2 |
||
| 393 | assert any(message.type == common.MESSAGE_TYPE_INFO for message in messages) |
||
| 394 | assert any(message.type == common.MESSAGE_TYPE_FATAL for message in messages) |
||
| 395 | |||
| 396 | # the info message should report mount options added to the existing |
||
| 397 | # mount point, the error message shoud contain the missing mount point |
||
| 398 | # and not the mount option |
||
| 399 | for message in messages: |
||
| 400 | if message.type == common.MESSAGE_TYPE_INFO: |
||
| 401 | assert "/" in message.text |
||
| 402 | assert "'noauto'" in message.text |
||
| 403 | elif message.type == common.MESSAGE_TYPE_FATAL: |
||
| 404 | assert "/tmp" in message.text |
||
| 405 | assert "'nodev'" not in message.text |
||
| 406 | |||
| 407 | |||
| 408 | def test_evaluation_passwd_minlen_no_passwd( |
||
| 409 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 410 | password_proxy_mock = USERS.get_proxy() |
||
| 411 | password_proxy_mock.IsRootPasswordSet = False |
||
| 412 | evaluation_passwd_minlen_no_passwd(rule_data, ksdata_mock, storage_mock, 8, (10, 11)) |
||
| 413 | evaluation_passwd_minlen_no_passwd(rule_data, ksdata_mock, storage_mock, 10, (8, 11)) |
||
| 414 | evaluation_passwd_minlen_no_passwd(rule_data, ksdata_mock, storage_mock, 11, (8, 10)) |
||
| 415 | |||
| 416 | |||
| 417 | def evaluation_passwd_minlen_no_passwd( |
||
| 418 | rule_data, ksdata_mock, storage_mock, min_password_length, check_against=tuple()): |
||
| 419 | rule_data.new_rule("passwd --minlen={0}".format(min_password_length)) |
||
| 420 | |||
| 421 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 422 | |||
| 423 | # minimal password length required --> one warning |
||
| 424 | assert len(messages) == 1 |
||
| 425 | assert messages[0].type == common.MESSAGE_TYPE_WARNING |
||
| 426 | |||
| 427 | # warning has to mention the length |
||
| 428 | assert str(min_password_length) in messages[0].text |
||
| 429 | |||
| 430 | for not_wanted in check_against: |
||
| 431 | assert str(not_wanted) not in messages[0].text |
||
| 432 | |||
| 433 | |||
| 434 | View Code Duplication | def test_evaluation_passwd_minlen_short_passwd( |
|
|
|
|||
| 435 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 436 | password_proxy_mock = USERS.get_proxy() |
||
| 437 | password_proxy_mock.IsRootPasswordCrypted = False |
||
| 438 | password_proxy_mock.RootPassword = "aaaa" |
||
| 439 | |||
| 440 | rule_data.new_rule("passwd --minlen=8") |
||
| 441 | |||
| 442 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=False) |
||
| 443 | |||
| 444 | # minimal password length greater than actual length --> one warning |
||
| 445 | assert len(messages) == 1 |
||
| 446 | assert messages[0].type == common.MESSAGE_TYPE_FATAL |
||
| 447 | |||
| 448 | # warning has to mention the length |
||
| 449 | assert "8" in messages[0].text |
||
| 450 | |||
| 451 | # warning should mention that something is wrong with the old password |
||
| 452 | assert "is" in messages[0].text |
||
| 453 | |||
| 454 | # doing changes --> password should not be cleared |
||
| 455 | assert password_proxy_mock.RootPassword == "aaaa" |
||
| 456 | |||
| 457 | |||
| 458 | View Code Duplication | def test_evaluation_passwd_minlen_short_passwd_report_only( |
|
| 459 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 460 | password_proxy_mock = USERS.get_proxy() |
||
| 461 | password_proxy_mock.IsRootPasswordCrypted = False |
||
| 462 | password_proxy_mock.RootPassword = "aaaa" |
||
| 463 | |||
| 464 | rule_data.new_rule("passwd --minlen=8") |
||
| 465 | |||
| 466 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=True) |
||
| 467 | |||
| 468 | # minimal password length greater than actual length --> one warning |
||
| 469 | assert len(messages) == 1 |
||
| 470 | assert messages[0].type == common.MESSAGE_TYPE_FATAL |
||
| 471 | |||
| 472 | # warning has to mention the length |
||
| 473 | assert "8" in messages[0].text |
||
| 474 | |||
| 475 | # warning should mention that something is wrong with the old password |
||
| 476 | assert "is" in messages[0].text |
||
| 477 | |||
| 478 | # doing changes --> password should not be cleared |
||
| 479 | assert password_proxy_mock.RootPassword == "aaaa" |
||
| 480 | |||
| 481 | |||
| 482 | def test_evaluation_passwd_minlen_crypted_passwd( |
||
| 483 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 484 | password_proxy_mock = USERS.get_proxy() |
||
| 485 | password_proxy_mock.IsRootPasswordCrypted = True |
||
| 486 | password_proxy_mock.RootPassword = "aaaa" |
||
| 487 | |||
| 488 | rule_data.new_rule("passwd --minlen=8") |
||
| 489 | |||
| 490 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=False) |
||
| 491 | |||
| 492 | # minimal password length greater than actual length --> one warning |
||
| 493 | assert len(messages) == 1 |
||
| 494 | assert messages[0].type == common.MESSAGE_TYPE_WARNING |
||
| 495 | |||
| 496 | # warning has to mention that the password cannot be checked |
||
| 497 | assert "cannot check" in messages[0].text |
||
| 498 | |||
| 499 | |||
| 500 | def test_evaluation_passwd_minlen_good_passwd(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 501 | password_proxy_mock = USERS.get_proxy() |
||
| 502 | password_proxy_mock.IsRootPasswordCrypted = False |
||
| 503 | password_proxy_mock.RootPassword = "aaaaaaaaaaaaaaaaa" |
||
| 504 | |||
| 505 | rule_data.new_rule("passwd --minlen=8") |
||
| 506 | |||
| 507 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=False) |
||
| 508 | |||
| 509 | # minimal password length less than actual length --> no warning |
||
| 510 | assert not messages |
||
| 511 | |||
| 512 | |||
| 513 | def test_evaluation_passwd_minlen_report_only_not_ignored( |
||
| 514 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 515 | password_proxy_mock = USERS.get_proxy() |
||
| 516 | password_proxy_mock.IsRootPasswordCrypted = False |
||
| 517 | password_proxy_mock.RootPassword = "aaaaaaaaaaaaaaaaa" |
||
| 518 | |||
| 519 | rule_data.new_rule("passwd --minlen=8") |
||
| 520 | |||
| 521 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=False) |
||
| 522 | |||
| 523 | # Mock pw_policy returned by anaconda.pwpolicy.get_policy() |
||
| 524 | pw_policy_mock = mock.Mock() |
||
| 525 | pw_policy_mock.minlen = 6 |
||
| 526 | pw_policy_mock.strict = False |
||
| 527 | ksdata_mock.anaconda.pwpolicy.get_policy.return_value = pw_policy_mock |
||
| 528 | |||
| 529 | # call eval_rules with report_only=False |
||
| 530 | # should set password minimal length to 8 |
||
| 531 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=False) |
||
| 532 | |||
| 533 | # Password Policy changed --> no warnings |
||
| 534 | assert not messages |
||
| 535 | assert rule_data._passwd_rules._orig_minlen == 6 |
||
| 536 | assert not rule_data._passwd_rules._orig_strict |
||
| 537 | assert pw_policy_mock.minlen == 8 |
||
| 538 | assert pw_policy_mock.strict |
||
| 539 | assert rule_data._passwd_rules._minlen == 8 |
||
| 540 | |||
| 541 | # call of eval_rules with report_only=True |
||
| 542 | # should not change anything |
||
| 543 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=True) |
||
| 544 | # Password Policy stayed the same --> no warnings |
||
| 545 | assert not messages |
||
| 546 | |||
| 547 | assert rule_data._passwd_rules._orig_minlen == 6 |
||
| 548 | assert not rule_data._passwd_rules._orig_strict |
||
| 549 | assert pw_policy_mock.minlen == 8 |
||
| 550 | assert pw_policy_mock.strict |
||
| 551 | assert rule_data._passwd_rules._minlen == 8 |
||
| 552 | |||
| 553 | |||
| 554 | def _occurences_not_seen_in_strings(seeked, strings): |
||
| 555 | found = set(seeked) |
||
| 556 | for string in strings: |
||
| 557 | for might_have_seen in seeked: |
||
| 558 | if might_have_seen in string: |
||
| 559 | found.add(string) |
||
| 560 | break |
||
| 561 | return set(seeked).difference(found) |
||
| 562 | |||
| 563 | |||
| 564 | def _quoted_keywords_not_seen_in_messages(keywords, messages): |
||
| 565 | return _occurences_not_seen_in_strings( |
||
| 566 | {"'{}'".format(kw) for kw in keywords}, |
||
| 567 | [m.text for m in messages], |
||
| 568 | ) |
||
| 569 | |||
| 570 | |||
| 571 | def test_evaluation_package_rules(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 572 | rule_data.new_rule("package --add=firewalld --remove=telnet --add=iptables --add=vim") |
||
| 573 | |||
| 574 | ksdata_mock.packages.packageList = ["vim"] |
||
| 575 | ksdata_mock.packages.excludedList = [] |
||
| 576 | |||
| 577 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 578 | |||
| 579 | # one info message for each (really) added/removed package |
||
| 580 | assert len(messages) == 3 |
||
| 581 | assert all(message.type == common.MESSAGE_TYPE_INFO for message in messages) |
||
| 582 | |||
| 583 | # all packages should appear in the messages |
||
| 584 | not_seen = _quoted_keywords_not_seen_in_messages( |
||
| 585 | {"firewalld", "telnet", "iptables"}, |
||
| 586 | messages, |
||
| 587 | ) |
||
| 588 | |||
| 589 | assert not not_seen |
||
| 590 | assert set(ksdata_mock.packages.packageList) == {"firewalld", "iptables", "vim"} |
||
| 591 | assert set(ksdata_mock.packages.excludedList) == {"telnet"} |
||
| 592 | |||
| 593 | |||
| 594 | def test_evaluation_package_rules_report_only(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 595 | rule_data.new_rule("package --add=firewalld --remove=telnet --add=iptables") |
||
| 596 | |||
| 597 | ksdata_mock.packages.packageList = [] |
||
| 598 | ksdata_mock.packages.excludedList = [] |
||
| 599 | |||
| 600 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=True) |
||
| 601 | |||
| 602 | # one info message for each added/removed package |
||
| 603 | assert len(messages) == 3 |
||
| 604 | assert all(message.type == common.MESSAGE_TYPE_INFO for message in messages) |
||
| 605 | |||
| 606 | not_seen = _quoted_keywords_not_seen_in_messages( |
||
| 607 | {"firewalld", "telnet", "iptables"}, |
||
| 608 | messages, |
||
| 609 | ) |
||
| 610 | |||
| 611 | assert not not_seen |
||
| 612 | |||
| 613 | # report_only --> no packages should be added or excluded |
||
| 614 | assert not ksdata_mock.packages.packageList |
||
| 615 | assert not ksdata_mock.packages.excludedList |
||
| 616 | |||
| 617 | |||
| 618 | def test_evaluation_bootloader_passwd_not_set(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 619 | bootloader_proxy_mock = STORAGE.get_proxy(BOOTLOADER) |
||
| 620 | bootloader_proxy_mock.IsPasswordSet = False |
||
| 621 | |||
| 622 | rule_data.new_rule("bootloader --passwd") |
||
| 623 | |||
| 624 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 625 | |||
| 626 | # bootloader password not set --> one warning |
||
| 627 | assert len(messages) == 1 |
||
| 628 | assert messages[0].type == common.MESSAGE_TYPE_WARNING |
||
| 629 | |||
| 630 | |||
| 631 | def test_evaluation_bootloader_passwd_set(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 632 | bootloader_proxy_mock = STORAGE.get_proxy(BOOTLOADER) |
||
| 633 | bootloader_proxy_mock.IsPasswordSet = True |
||
| 634 | |||
| 635 | rule_data.new_rule("bootloader --passwd") |
||
| 636 | |||
| 637 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 638 | |||
| 639 | # bootloader password set --> no warnings |
||
| 640 | assert messages == [] |
||
| 641 | |||
| 642 | |||
| 643 | def test_evaluation_various_rules(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 644 | for rule in ["part /tmp", "part /", "passwd --minlen=14", |
||
| 645 | "package --add=firewalld", ]: |
||
| 646 | rule_data.new_rule(rule) |
||
| 647 | |||
| 648 | storage_mock.mountpoints = dict() |
||
| 649 | ksdata_mock.packages.packageList = [] |
||
| 650 | ksdata_mock.packages.excludedList = [] |
||
| 651 | |||
| 652 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 653 | |||
| 654 | # four rules, all fail --> four messages |
||
| 655 | assert len(messages) == 4 |
||
| 656 | |||
| 657 | |||
| 658 | def test_revert_mount_options_nonexistent(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 659 | rule_data.new_rule("part /tmp --mountoptions=nodev") |
||
| 660 | storage_mock.mountpoints = dict() |
||
| 661 | |||
| 662 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 663 | |||
| 664 | # mount point doesn't exist -> one message, nothing done |
||
| 665 | assert len(messages) == 1 |
||
| 666 | assert storage_mock.mountpoints == dict() |
||
| 667 | |||
| 668 | # mount point doesn't exist -> shouldn't do anything |
||
| 669 | rule_data.revert_changes(ksdata_mock, storage_mock) |
||
| 670 | assert storage_mock.mountpoints == dict() |
||
| 671 | |||
| 672 | |||
| 673 | def test_revert_mount_options(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 674 | rule_data.new_rule("part /tmp --mountoptions=nodev") |
||
| 675 | storage_mock.mountpoints = dict() |
||
| 676 | storage_mock.mountpoints["/tmp"] = mock.Mock() |
||
| 677 | storage_mock.mountpoints["/tmp"].format.options = "defaults" |
||
| 678 | |||
| 679 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 680 | |||
| 681 | # mount option added --> one message |
||
| 682 | assert len(messages) == 1 |
||
| 683 | |||
| 684 | # "nodev" option should be added |
||
| 685 | assert storage_mock.mountpoints["/tmp"].format.options, "defaults == nodev" |
||
| 686 | |||
| 687 | rule_data.revert_changes(ksdata_mock, storage_mock) |
||
| 688 | |||
| 689 | # should be reverted to the original value |
||
| 690 | assert storage_mock.mountpoints["/tmp"].format.options == "defaults" |
||
| 691 | |||
| 692 | # another cycle of the same # |
||
| 693 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 694 | |||
| 695 | # mount option added --> one message |
||
| 696 | assert len(messages) == 1 |
||
| 697 | |||
| 698 | # "nodev" option should be added |
||
| 699 | assert storage_mock.mountpoints["/tmp"].format.options, "defaults == nodev" |
||
| 700 | |||
| 701 | rule_data.revert_changes(ksdata_mock, storage_mock) |
||
| 702 | |||
| 703 | # should be reverted to the original value |
||
| 704 | assert storage_mock.mountpoints["/tmp"].format.options == "defaults" |
||
| 705 | |||
| 706 | |||
| 707 | def test_revert_password_policy_changes(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 708 | password_proxy_mock = USERS.get_proxy() |
||
| 709 | password_proxy_mock.IsRootPasswordCrypted = False |
||
| 710 | password_proxy_mock.RootPassword = "aaaa" |
||
| 711 | |||
| 712 | # FIXME: Add password policy changes to this test. It only checks |
||
| 713 | # password length right now outside of policy changes. |
||
| 714 | rule_data.new_rule("passwd --minlen=8") |
||
| 715 | |||
| 716 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 717 | |||
| 718 | # password error --> one message |
||
| 719 | assert len(messages) == 1 |
||
| 720 | |||
| 721 | rule_data.revert_changes(ksdata_mock, storage_mock) |
||
| 722 | |||
| 723 | # with long enough password this time # |
||
| 724 | password_proxy_mock.RootPassword = "aaaaaaaaaaaaa" |
||
| 725 | |||
| 726 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 727 | |||
| 728 | # long enough password |
||
| 729 | # entered --> no message |
||
| 730 | assert messages == [] |
||
| 731 | |||
| 732 | |||
| 733 | def test_revert_package_rules(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 734 | rule_data.new_rule("package --add=firewalld --remove=telnet --add=iptables --add=vim") |
||
| 735 | |||
| 736 | ksdata_mock.packages.packageList = ["vim"] |
||
| 737 | ksdata_mock.packages.excludedList = [] |
||
| 738 | |||
| 739 | # run twice --> nothing should be different in the second run |
||
| 740 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 741 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 742 | |||
| 743 | # one info message for each added/removed package |
||
| 744 | assert len(messages) == 3 |
||
| 745 | |||
| 746 | rule_data.revert_changes(ksdata_mock, storage_mock) |
||
| 747 | |||
| 748 | # (only) added and excluded packages should have been removed from the |
||
| 749 | # list |
||
| 750 | assert ksdata_mock.packages.packageList == ["vim"] |
||
| 751 | assert ksdata_mock.packages.excludedList == [] |
||
| 752 | |||
| 753 | # now do the same again # |
||
| 754 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 755 | |||
| 756 | # one info message for each added/removed package |
||
| 757 | assert len(messages) == 3 |
||
| 758 | |||
| 759 | rule_data.revert_changes(ksdata_mock, storage_mock) |
||
| 760 | |||
| 761 | # (only) added and excluded packages should have been removed from the |
||
| 762 | # list |
||
| 763 | assert ksdata_mock.packages.packageList == ["vim"] |
||
| 764 | assert ksdata_mock.packages.excludedList == [] |
||
| 765 |