| Total Complexity | 64 |
| Total Lines | 762 |
| Duplicated Lines | 5.77 % |
| 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 | @pytest.fixture() |
||
| 120 | def proxy_getter(monkeypatch): |
||
| 121 | proxies = defaultdict(mock.Mock) |
||
| 122 | |||
| 123 | def mock_get(service_name, object_path): |
||
| 124 | initialize = not proxies |
||
| 125 | proxy = proxies[(service_name, object_path)] |
||
| 126 | |||
| 127 | if initialize: |
||
| 128 | set_dbus_defaults() |
||
| 129 | |||
| 130 | return proxy |
||
| 131 | |||
| 132 | monkeypatch.setattr("pyanaconda.core.dbus.DBus.get_proxy", mock_get) |
||
| 133 | return mock_get |
||
| 134 | |||
| 135 | |||
| 136 | def set_dbus_defaults(): |
||
| 137 | network = NETWORK.get_proxy() |
||
| 138 | network.Connected.return_value = True |
||
| 139 | |||
| 140 | firewall = NETWORK.get_proxy(FIREWALL) |
||
| 141 | firewall.EnabledServices = [] |
||
| 142 | firewall.DisabledServices = [] |
||
| 143 | firewall.EnabledPorts = [] |
||
| 144 | firewall.Trusts = [] |
||
| 145 | |||
| 146 | device_tree = STORAGE.get_proxy(DEVICE_TREE) |
||
| 147 | device_tree.GetDeviceMountOptions.return_value = "defaults" |
||
| 148 | device_tree.GetMountPoints.return_value = {} |
||
| 149 | |||
| 150 | bootloader = STORAGE.get_proxy(BOOTLOADER) |
||
| 151 | bootloader.IsPasswordSet = False |
||
| 152 | |||
| 153 | users = USERS.get_proxy() |
||
| 154 | users.IsRootPasswordSet = True |
||
| 155 | users.IsRootPasswordCrypted = False |
||
| 156 | users.RootPassword = "anaconda" |
||
| 157 | |||
| 158 | |||
| 159 | def test_evaluation_existing_part_must_exist_rules( |
||
| 160 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 161 | rules = [ |
||
| 162 | "part /tmp", |
||
| 163 | "part /", |
||
| 164 | ] |
||
| 165 | for rule in rules: |
||
| 166 | rule_data.new_rule(rule) |
||
| 167 | |||
| 168 | device_tree_mock = STORAGE.get_proxy(DEVICE_TREE) |
||
| 169 | device_tree_mock.GetDeviceMountOptions.return_value = "defaults" |
||
| 170 | device_tree_mock.GetMountPoints.return_value = { |
||
| 171 | "/tmp": "/dev/sda1", |
||
| 172 | "/": "/dev/sda2", |
||
| 173 | } |
||
| 174 | |||
| 175 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 176 | |||
| 177 | # partitions exist --> no errors, warnings or additional info |
||
| 178 | assert not messages |
||
| 179 | |||
| 180 | # no additional mount options specified |
||
| 181 | device_tree_mock.SetDeviceMountOptions.assert_has_calls([ |
||
| 182 | mock.call("/dev/sda1", "defaults"), |
||
| 183 | mock.call("/dev/sda2", "defaults"), |
||
| 184 | ]) |
||
| 185 | |||
| 186 | |||
| 187 | def test_evaluation_nonexisting_part_must_exist( |
||
| 188 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 189 | rules = [ |
||
| 190 | "part /tmp", |
||
| 191 | "part /", |
||
| 192 | ] |
||
| 193 | for rule in rules: |
||
| 194 | rule_data.new_rule(rule) |
||
| 195 | |||
| 196 | device_tree_mock = STORAGE.get_proxy(DEVICE_TREE) |
||
| 197 | device_tree_mock.GetDeviceMountOptions.return_value = "defaults" |
||
| 198 | device_tree_mock.GetMountPoints.return_value = { |
||
| 199 | "/tmp": "/dev/sda1", |
||
| 200 | } |
||
| 201 | |||
| 202 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 203 | |||
| 204 | # / mount point missing --> one error |
||
| 205 | assert len(messages) == 1 |
||
| 206 | assert messages[0].type == common.MESSAGE_TYPE_FATAL |
||
| 207 | |||
| 208 | # error has to mention the mount point |
||
| 209 | assert "/" in messages[0].text |
||
| 210 | |||
| 211 | |||
| 212 | def get_messages_for_partition_rules( |
||
| 213 | rule_data, ksdata_mock, storage_mock, |
||
| 214 | rules, |
||
| 215 | messages_evaluation_count=1, |
||
| 216 | actual_mountpoints=("/tmp", "/"), |
||
| 217 | mount_options=None, |
||
| 218 | report_only=False, |
||
| 219 | ): |
||
| 220 | |||
| 221 | assert len(rules) == 2, \ |
||
| 222 | "We need rules for temp partition and root." |
||
| 223 | |||
| 224 | if mount_options is None: |
||
| 225 | mount_options = { |
||
| 226 | "/": "defaults", |
||
| 227 | "/tmp": "defaults", |
||
| 228 | } |
||
| 229 | |||
| 230 | for rule in rules: |
||
| 231 | rule_data.new_rule(rule) |
||
| 232 | |||
| 233 | # Map mount points to device names. |
||
| 234 | mount_points = {} |
||
| 235 | |||
| 236 | if "/tmp" in actual_mountpoints: |
||
| 237 | mount_points["/tmp"] = "/dev/sda1" |
||
| 238 | |||
| 239 | if "/" in actual_mountpoints: |
||
| 240 | mount_points["/"] = "/dev/sda2" |
||
| 241 | |||
| 242 | # Map device names to mount options. |
||
| 243 | mount_options = { |
||
| 244 | mount_points[mount_point]: mount_options[mount_point] |
||
| 245 | for mount_point in actual_mountpoints |
||
| 246 | } |
||
| 247 | |||
| 248 | # Mock the device tree proxy. |
||
| 249 | def get_device_mount_options(device_name): |
||
| 250 | return mount_options[device_name] |
||
| 251 | |||
| 252 | def set_device_mount_options(device_name, options): |
||
| 253 | mount_options[device_name] = options |
||
| 254 | |||
| 255 | device_tree_mock = STORAGE.get_proxy(DEVICE_TREE) |
||
| 256 | device_tree_mock.GetMountPoints.return_value = mount_points |
||
| 257 | device_tree_mock.GetDeviceMountOptions.side_effect = get_device_mount_options |
||
| 258 | device_tree_mock.SetDeviceMountOptions.side_effect = set_device_mount_options |
||
| 259 | |||
| 260 | messages = [] |
||
| 261 | for _ in range(messages_evaluation_count): |
||
| 262 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only) |
||
| 263 | |||
| 264 | return messages |
||
| 265 | |||
| 266 | |||
| 267 | def evaluation_add_mount_options( |
||
| 268 | rule_data, ksdata_mock, storage_mock, |
||
| 269 | messages_evaluation_count): |
||
| 270 | rules = [ |
||
| 271 | "part /tmp --mountoptions=defaults,nodev", |
||
| 272 | "part / --mountoptions=noauto", |
||
| 273 | ] |
||
| 274 | |||
| 275 | messages = get_messages_for_partition_rules( |
||
| 276 | rule_data, ksdata_mock, storage_mock, |
||
| 277 | rules, messages_evaluation_count) |
||
| 278 | |||
| 279 | # two mount options added --> two info messages |
||
| 280 | assert len(messages) == 2 |
||
| 281 | assert all(message.type == common.MESSAGE_TYPE_INFO for message in messages) |
||
| 282 | |||
| 283 | # newly added mount options should be mentioned in the messages |
||
| 284 | # together with their mount points |
||
| 285 | nodev_found = False |
||
| 286 | noauto_found = False |
||
| 287 | |||
| 288 | for message in messages: |
||
| 289 | if "'nodev'" in message.text: |
||
| 290 | assert "/tmp" in message.text |
||
| 291 | nodev_found = True |
||
| 292 | elif "'noauto'" in message.text: |
||
| 293 | assert "/" in message.text |
||
| 294 | noauto_found = True |
||
| 295 | |||
| 296 | assert all([nodev_found, noauto_found]) |
||
| 297 | |||
| 298 | device_tree_mock = STORAGE.get_proxy(DEVICE_TREE) |
||
| 299 | device_tree_mock.SetDeviceMountOptions.assert_has_calls([ |
||
| 300 | mock.call("/dev/sda1", "defaults,nodev"), |
||
| 301 | mock.call("/dev/sda2", "defaults,noauto"), |
||
| 302 | ]) |
||
| 303 | |||
| 304 | |||
| 305 | def test_evaluation_add_mount_options( |
||
| 306 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 307 | evaluation_add_mount_options(rule_data, ksdata_mock, storage_mock, 1) |
||
| 308 | |||
| 309 | |||
| 310 | def test_evaluation_add_mount_options_no_duplicates( |
||
| 311 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 312 | evaluation_add_mount_options(rule_data, ksdata_mock, storage_mock, 2) |
||
| 313 | |||
| 314 | |||
| 315 | def test_evaluation_add_mount_options_report_only( |
||
| 316 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 317 | rules = [ |
||
| 318 | "part /tmp --mountoptions=nodev", |
||
| 319 | "part / --mountoptions=noauto", |
||
| 320 | ] |
||
| 321 | messages = get_messages_for_partition_rules( |
||
| 322 | rule_data, ksdata_mock, storage_mock, |
||
| 323 | rules, 1, report_only=True) |
||
| 324 | |||
| 325 | # two mount options added --> two info messages |
||
| 326 | assert len(messages) == 2 |
||
| 327 | assert messages[0].type == common.MESSAGE_TYPE_INFO |
||
| 328 | assert messages[1].type == common.MESSAGE_TYPE_INFO |
||
| 329 | |||
| 330 | # newly added mount options should be mentioned in the messages |
||
| 331 | # together with their mount points |
||
| 332 | nodev_found = False |
||
| 333 | noauto_found = False |
||
| 334 | |||
| 335 | for message in messages: |
||
| 336 | if "'nodev'" in message.text: |
||
| 337 | assert "/tmp" in message.text |
||
| 338 | nodev_found = True |
||
| 339 | elif "'noauto'" in message.text: |
||
| 340 | assert "/" in message.text |
||
| 341 | noauto_found = True |
||
| 342 | |||
| 343 | assert all([nodev_found, noauto_found]) |
||
| 344 | |||
| 345 | # no changes should be made |
||
| 346 | device_tree_mock = STORAGE.get_proxy(DEVICE_TREE) |
||
| 347 | device_tree_mock.SetDeviceMountOptions.assert_not_called() |
||
| 348 | |||
| 349 | |||
| 350 | def test_evaluation_add_mount_option_prefix( |
||
| 351 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 352 | rules = [ |
||
| 353 | "part /tmp --mountoptions=nodev", |
||
| 354 | "part / --mountoptions=noauto", |
||
| 355 | ] |
||
| 356 | mount_options = { |
||
| 357 | "/": "defaults", |
||
| 358 | "/tmp": "defaults,nodevice", |
||
| 359 | } |
||
| 360 | messages = get_messages_for_partition_rules( |
||
| 361 | rule_data, ksdata_mock, storage_mock, |
||
| 362 | rules, mount_options=mount_options) |
||
| 363 | |||
| 364 | # two mount options added (even though it is a prefix of another one) |
||
| 365 | # --> two info messages |
||
| 366 | assert len(messages) == 2 |
||
| 367 | assert messages[0].type == common.MESSAGE_TYPE_INFO |
||
| 368 | assert messages[1].type == common.MESSAGE_TYPE_INFO |
||
| 369 | |||
| 370 | # the option should be added even though it is a prefix of another one |
||
| 371 | device_tree_mock = STORAGE.get_proxy(DEVICE_TREE) |
||
| 372 | device_tree_mock.SetDeviceMountOptions.assert_has_calls([ |
||
| 373 | mock.call("/dev/sda1", "defaults,nodevice,nodev"), |
||
| 374 | ]) |
||
| 375 | |||
| 376 | |||
| 377 | def test_evaluation_add_mount_options_nonexisting_part( |
||
| 378 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 379 | rules = [ |
||
| 380 | "part /tmp --mountoptions=nodev", |
||
| 381 | "part / --mountoptions=noauto", |
||
| 382 | ] |
||
| 383 | messages = get_messages_for_partition_rules( |
||
| 384 | rule_data, ksdata_mock, storage_mock, |
||
| 385 | rules, actual_mountpoints=["/"]) |
||
| 386 | |||
| 387 | # one mount option added, one mount point missing (mount options |
||
| 388 | # cannot be added) --> one info, one error |
||
| 389 | assert len(messages) == 2 |
||
| 390 | assert any(message.type == common.MESSAGE_TYPE_INFO for message in messages) |
||
| 391 | assert any(message.type == common.MESSAGE_TYPE_FATAL for message in messages) |
||
| 392 | |||
| 393 | # the info message should report mount options added to the existing |
||
| 394 | # mount point, the error message shoud contain the missing mount point |
||
| 395 | # and not the mount option |
||
| 396 | for message in messages: |
||
| 397 | if message.type == common.MESSAGE_TYPE_INFO: |
||
| 398 | assert "/" in message.text |
||
| 399 | assert "'noauto'" in message.text |
||
| 400 | elif message.type == common.MESSAGE_TYPE_FATAL: |
||
| 401 | assert "/tmp" in message.text |
||
| 402 | assert "'nodev'" not in message.text |
||
| 403 | |||
| 404 | |||
| 405 | def test_evaluation_passwd_minlen_no_passwd( |
||
| 406 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 407 | password_proxy_mock = USERS.get_proxy() |
||
| 408 | password_proxy_mock.IsRootPasswordSet = False |
||
| 409 | evaluation_passwd_minlen_no_passwd(rule_data, ksdata_mock, storage_mock, 8, (10, 11)) |
||
| 410 | evaluation_passwd_minlen_no_passwd(rule_data, ksdata_mock, storage_mock, 10, (8, 11)) |
||
| 411 | evaluation_passwd_minlen_no_passwd(rule_data, ksdata_mock, storage_mock, 11, (8, 10)) |
||
| 412 | |||
| 413 | |||
| 414 | def evaluation_passwd_minlen_no_passwd( |
||
| 415 | rule_data, ksdata_mock, storage_mock, min_password_length, check_against=tuple()): |
||
| 416 | rule_data.new_rule("passwd --minlen={0}".format(min_password_length)) |
||
| 417 | |||
| 418 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 419 | |||
| 420 | # minimal password length required --> one warning |
||
| 421 | assert len(messages) == 1 |
||
| 422 | assert messages[0].type == common.MESSAGE_TYPE_WARNING |
||
| 423 | |||
| 424 | # warning has to mention the length |
||
| 425 | assert str(min_password_length) in messages[0].text |
||
| 426 | |||
| 427 | for not_wanted in check_against: |
||
| 428 | assert str(not_wanted) not in messages[0].text |
||
| 429 | |||
| 430 | |||
| 431 | View Code Duplication | def test_evaluation_passwd_minlen_short_passwd( |
|
|
|
|||
| 432 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 433 | password_proxy_mock = USERS.get_proxy() |
||
| 434 | password_proxy_mock.IsRootPasswordCrypted = False |
||
| 435 | password_proxy_mock.RootPassword = "aaaa" |
||
| 436 | |||
| 437 | rule_data.new_rule("passwd --minlen=8") |
||
| 438 | |||
| 439 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=False) |
||
| 440 | |||
| 441 | # minimal password length greater than actual length --> one warning |
||
| 442 | assert len(messages) == 1 |
||
| 443 | assert messages[0].type == common.MESSAGE_TYPE_FATAL |
||
| 444 | |||
| 445 | # warning has to mention the length |
||
| 446 | assert "8" in messages[0].text |
||
| 447 | |||
| 448 | # warning should mention that something is wrong with the old password |
||
| 449 | assert "is" in messages[0].text |
||
| 450 | |||
| 451 | # doing changes --> password should not be cleared |
||
| 452 | assert password_proxy_mock.RootPassword == "aaaa" |
||
| 453 | |||
| 454 | |||
| 455 | View Code Duplication | def test_evaluation_passwd_minlen_short_passwd_report_only( |
|
| 456 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 457 | password_proxy_mock = USERS.get_proxy() |
||
| 458 | password_proxy_mock.IsRootPasswordCrypted = False |
||
| 459 | password_proxy_mock.RootPassword = "aaaa" |
||
| 460 | |||
| 461 | rule_data.new_rule("passwd --minlen=8") |
||
| 462 | |||
| 463 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=True) |
||
| 464 | |||
| 465 | # minimal password length greater than actual length --> one warning |
||
| 466 | assert len(messages) == 1 |
||
| 467 | assert messages[0].type == common.MESSAGE_TYPE_FATAL |
||
| 468 | |||
| 469 | # warning has to mention the length |
||
| 470 | assert "8" in messages[0].text |
||
| 471 | |||
| 472 | # warning should mention that something is wrong with the old password |
||
| 473 | assert "is" in messages[0].text |
||
| 474 | |||
| 475 | # doing changes --> password should not be cleared |
||
| 476 | assert password_proxy_mock.RootPassword == "aaaa" |
||
| 477 | |||
| 478 | |||
| 479 | def test_evaluation_passwd_minlen_crypted_passwd( |
||
| 480 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 481 | password_proxy_mock = USERS.get_proxy() |
||
| 482 | password_proxy_mock.IsRootPasswordCrypted = True |
||
| 483 | password_proxy_mock.RootPassword = "aaaa" |
||
| 484 | |||
| 485 | rule_data.new_rule("passwd --minlen=8") |
||
| 486 | |||
| 487 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=False) |
||
| 488 | |||
| 489 | # minimal password length greater than actual length --> one warning |
||
| 490 | assert len(messages) == 1 |
||
| 491 | assert messages[0].type == common.MESSAGE_TYPE_WARNING |
||
| 492 | |||
| 493 | # warning has to mention that the password cannot be checked |
||
| 494 | assert "cannot check" in messages[0].text |
||
| 495 | |||
| 496 | |||
| 497 | def test_evaluation_passwd_minlen_good_passwd(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 498 | password_proxy_mock = USERS.get_proxy() |
||
| 499 | password_proxy_mock.IsRootPasswordCrypted = False |
||
| 500 | password_proxy_mock.RootPassword = "aaaaaaaaaaaaaaaaa" |
||
| 501 | |||
| 502 | rule_data.new_rule("passwd --minlen=8") |
||
| 503 | |||
| 504 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=False) |
||
| 505 | |||
| 506 | # minimal password length less than actual length --> no warning |
||
| 507 | assert not messages |
||
| 508 | |||
| 509 | |||
| 510 | def test_evaluation_passwd_minlen_report_only_not_ignored( |
||
| 511 | proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 512 | password_proxy_mock = USERS.get_proxy() |
||
| 513 | password_proxy_mock.IsRootPasswordCrypted = False |
||
| 514 | password_proxy_mock.RootPassword = "aaaaaaaaaaaaaaaaa" |
||
| 515 | |||
| 516 | rule_data.new_rule("passwd --minlen=8") |
||
| 517 | |||
| 518 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=False) |
||
| 519 | |||
| 520 | # Mock pw_policy returned by anaconda.pwpolicy.get_policy() |
||
| 521 | pw_policy_mock = mock.Mock() |
||
| 522 | pw_policy_mock.minlen = 6 |
||
| 523 | pw_policy_mock.strict = False |
||
| 524 | ksdata_mock.anaconda.pwpolicy.get_policy.return_value = pw_policy_mock |
||
| 525 | |||
| 526 | # call eval_rules with report_only=False |
||
| 527 | # should set password minimal length to 8 |
||
| 528 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=False) |
||
| 529 | |||
| 530 | # Password Policy changed --> no warnings |
||
| 531 | assert not messages |
||
| 532 | assert rule_data._passwd_rules._orig_minlen == 6 |
||
| 533 | assert not rule_data._passwd_rules._orig_strict |
||
| 534 | assert pw_policy_mock.minlen == 8 |
||
| 535 | assert pw_policy_mock.strict |
||
| 536 | assert rule_data._passwd_rules._minlen == 8 |
||
| 537 | |||
| 538 | # call of eval_rules with report_only=True |
||
| 539 | # should not change anything |
||
| 540 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=True) |
||
| 541 | # Password Policy stayed the same --> no warnings |
||
| 542 | assert not messages |
||
| 543 | |||
| 544 | assert rule_data._passwd_rules._orig_minlen == 6 |
||
| 545 | assert not rule_data._passwd_rules._orig_strict |
||
| 546 | assert pw_policy_mock.minlen == 8 |
||
| 547 | assert pw_policy_mock.strict |
||
| 548 | assert rule_data._passwd_rules._minlen == 8 |
||
| 549 | |||
| 550 | |||
| 551 | def _occurences_not_seen_in_strings(seeked, strings): |
||
| 552 | found = set(seeked) |
||
| 553 | for string in strings: |
||
| 554 | for might_have_seen in seeked: |
||
| 555 | if might_have_seen in string: |
||
| 556 | found.add(string) |
||
| 557 | break |
||
| 558 | return set(seeked).difference(found) |
||
| 559 | |||
| 560 | |||
| 561 | def _quoted_keywords_not_seen_in_messages(keywords, messages): |
||
| 562 | return _occurences_not_seen_in_strings( |
||
| 563 | {"'{}'".format(kw) for kw in keywords}, |
||
| 564 | [m.text for m in messages], |
||
| 565 | ) |
||
| 566 | |||
| 567 | |||
| 568 | def test_evaluation_package_rules(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 569 | rule_data.new_rule("package --add=firewalld --remove=telnet --add=iptables --add=vim") |
||
| 570 | |||
| 571 | ksdata_mock.packages.packageList = ["vim"] |
||
| 572 | ksdata_mock.packages.excludedList = [] |
||
| 573 | |||
| 574 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 575 | |||
| 576 | # one info message for each (really) added/removed package |
||
| 577 | assert len(messages) == 3 |
||
| 578 | assert all(message.type == common.MESSAGE_TYPE_INFO for message in messages) |
||
| 579 | |||
| 580 | # all packages should appear in the messages |
||
| 581 | not_seen = _quoted_keywords_not_seen_in_messages( |
||
| 582 | {"firewalld", "telnet", "iptables"}, |
||
| 583 | messages, |
||
| 584 | ) |
||
| 585 | |||
| 586 | assert not not_seen |
||
| 587 | assert set(ksdata_mock.packages.packageList) == {"firewalld", "iptables", "vim"} |
||
| 588 | assert set(ksdata_mock.packages.excludedList) == {"telnet"} |
||
| 589 | |||
| 590 | |||
| 591 | def test_evaluation_package_rules_report_only(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 592 | rule_data.new_rule("package --add=firewalld --remove=telnet --add=iptables") |
||
| 593 | |||
| 594 | ksdata_mock.packages.packageList = [] |
||
| 595 | ksdata_mock.packages.excludedList = [] |
||
| 596 | |||
| 597 | messages = rule_data.eval_rules(ksdata_mock, storage_mock, report_only=True) |
||
| 598 | |||
| 599 | # one info message for each added/removed package |
||
| 600 | assert len(messages) == 3 |
||
| 601 | assert all(message.type == common.MESSAGE_TYPE_INFO for message in messages) |
||
| 602 | |||
| 603 | not_seen = _quoted_keywords_not_seen_in_messages( |
||
| 604 | {"firewalld", "telnet", "iptables"}, |
||
| 605 | messages, |
||
| 606 | ) |
||
| 607 | |||
| 608 | assert not not_seen |
||
| 609 | |||
| 610 | # report_only --> no packages should be added or excluded |
||
| 611 | assert not ksdata_mock.packages.packageList |
||
| 612 | assert not ksdata_mock.packages.excludedList |
||
| 613 | |||
| 614 | |||
| 615 | def test_evaluation_bootloader_passwd_not_set(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 616 | bootloader_proxy_mock = STORAGE.get_proxy(BOOTLOADER) |
||
| 617 | bootloader_proxy_mock.IsPasswordSet = False |
||
| 618 | |||
| 619 | rule_data.new_rule("bootloader --passwd") |
||
| 620 | |||
| 621 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 622 | |||
| 623 | # bootloader password not set --> one warning |
||
| 624 | assert len(messages) == 1 |
||
| 625 | assert messages[0].type == common.MESSAGE_TYPE_WARNING |
||
| 626 | |||
| 627 | |||
| 628 | def test_evaluation_bootloader_passwd_set(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 629 | bootloader_proxy_mock = STORAGE.get_proxy(BOOTLOADER) |
||
| 630 | bootloader_proxy_mock.IsPasswordSet = True |
||
| 631 | |||
| 632 | rule_data.new_rule("bootloader --passwd") |
||
| 633 | |||
| 634 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 635 | |||
| 636 | # bootloader password set --> no warnings |
||
| 637 | assert messages == [] |
||
| 638 | |||
| 639 | |||
| 640 | def test_evaluation_various_rules(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 641 | for rule in ["part /tmp", "part /", "passwd --minlen=14", |
||
| 642 | "package --add=firewalld", ]: |
||
| 643 | rule_data.new_rule(rule) |
||
| 644 | |||
| 645 | storage_mock.mountpoints = dict() |
||
| 646 | ksdata_mock.packages.packageList = [] |
||
| 647 | ksdata_mock.packages.excludedList = [] |
||
| 648 | |||
| 649 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 650 | |||
| 651 | # four rules, all fail --> four messages |
||
| 652 | assert len(messages) == 4 |
||
| 653 | |||
| 654 | |||
| 655 | def test_revert_mount_options_nonexistent(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 656 | rule_data.new_rule("part /tmp --mountoptions=nodev") |
||
| 657 | storage_mock.mountpoints = dict() |
||
| 658 | |||
| 659 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 660 | |||
| 661 | # mount point doesn't exist -> one message, nothing done |
||
| 662 | assert len(messages) == 1 |
||
| 663 | assert storage_mock.mountpoints == dict() |
||
| 664 | |||
| 665 | # mount point doesn't exist -> shouldn't do anything |
||
| 666 | rule_data.revert_changes(ksdata_mock, storage_mock) |
||
| 667 | assert storage_mock.mountpoints == dict() |
||
| 668 | |||
| 669 | |||
| 670 | def test_revert_mount_options(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 671 | rule_data.new_rule("part /tmp --mountoptions=nodev") |
||
| 672 | storage_mock.mountpoints = dict() |
||
| 673 | storage_mock.mountpoints["/tmp"] = mock.Mock() |
||
| 674 | storage_mock.mountpoints["/tmp"].format.options = "defaults" |
||
| 675 | |||
| 676 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 677 | |||
| 678 | # mount option added --> one message |
||
| 679 | assert len(messages) == 1 |
||
| 680 | |||
| 681 | # "nodev" option should be added |
||
| 682 | assert storage_mock.mountpoints["/tmp"].format.options, "defaults == nodev" |
||
| 683 | |||
| 684 | rule_data.revert_changes(ksdata_mock, storage_mock) |
||
| 685 | |||
| 686 | # should be reverted to the original value |
||
| 687 | assert storage_mock.mountpoints["/tmp"].format.options == "defaults" |
||
| 688 | |||
| 689 | # another cycle of the same # |
||
| 690 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 691 | |||
| 692 | # mount option added --> one message |
||
| 693 | assert len(messages) == 1 |
||
| 694 | |||
| 695 | # "nodev" option should be added |
||
| 696 | assert storage_mock.mountpoints["/tmp"].format.options, "defaults == nodev" |
||
| 697 | |||
| 698 | rule_data.revert_changes(ksdata_mock, storage_mock) |
||
| 699 | |||
| 700 | # should be reverted to the original value |
||
| 701 | assert storage_mock.mountpoints["/tmp"].format.options == "defaults" |
||
| 702 | |||
| 703 | |||
| 704 | def test_revert_password_policy_changes(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 705 | password_proxy_mock = USERS.get_proxy() |
||
| 706 | password_proxy_mock.IsRootPasswordCrypted = False |
||
| 707 | password_proxy_mock.RootPassword = "aaaa" |
||
| 708 | |||
| 709 | # FIXME: Add password policy changes to this test. It only checks |
||
| 710 | # password length right now outside of policy changes. |
||
| 711 | rule_data.new_rule("passwd --minlen=8") |
||
| 712 | |||
| 713 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 714 | |||
| 715 | # password error --> one message |
||
| 716 | assert len(messages) == 1 |
||
| 717 | |||
| 718 | rule_data.revert_changes(ksdata_mock, storage_mock) |
||
| 719 | |||
| 720 | # with long enough password this time # |
||
| 721 | password_proxy_mock.RootPassword = "aaaaaaaaaaaaa" |
||
| 722 | |||
| 723 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 724 | |||
| 725 | # long enough password |
||
| 726 | # entered --> no message |
||
| 727 | assert messages == [] |
||
| 728 | |||
| 729 | |||
| 730 | def test_revert_package_rules(proxy_getter, rule_data, ksdata_mock, storage_mock): |
||
| 731 | rule_data.new_rule("package --add=firewalld --remove=telnet --add=iptables --add=vim") |
||
| 732 | |||
| 733 | ksdata_mock.packages.packageList = ["vim"] |
||
| 734 | ksdata_mock.packages.excludedList = [] |
||
| 735 | |||
| 736 | # run twice --> nothing should be different in the second run |
||
| 737 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 738 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 739 | |||
| 740 | # one info message for each added/removed package |
||
| 741 | assert len(messages) == 3 |
||
| 742 | |||
| 743 | rule_data.revert_changes(ksdata_mock, storage_mock) |
||
| 744 | |||
| 745 | # (only) added and excluded packages should have been removed from the |
||
| 746 | # list |
||
| 747 | assert ksdata_mock.packages.packageList == ["vim"] |
||
| 748 | assert ksdata_mock.packages.excludedList == [] |
||
| 749 | |||
| 750 | # now do the same again # |
||
| 751 | messages = rule_data.eval_rules(ksdata_mock, storage_mock) |
||
| 752 | |||
| 753 | # one info message for each added/removed package |
||
| 754 | assert len(messages) == 3 |
||
| 755 | |||
| 756 | rule_data.revert_changes(ksdata_mock, storage_mock) |
||
| 757 | |||
| 758 | # (only) added and excluded packages should have been removed from the |
||
| 759 | # list |
||
| 760 | assert ksdata_mock.packages.packageList == ["vim"] |
||
| 761 | assert ksdata_mock.packages.excludedList == [] |
||
| 762 |