|
22 | 22 | get_user, |
23 | 23 | get_roles, |
24 | 24 | get_admin_users, |
| 25 | + get_superadmin_users, |
| 26 | + read_notification_users_fallback, |
25 | 27 | ) |
| 28 | +import wl_rbac as wl_rbac_module |
26 | 29 |
|
27 | 30 | from wl_constants import EDIT_ROLES, ADMIN_ROLES, SUPERADMIN_ROLES |
28 | 31 |
|
@@ -323,3 +326,267 @@ def test_workflow_get_user_and_roles(self): |
323 | 326 | assert user == "john_doe" |
324 | 327 | assert "wl_editor" in roles |
325 | 328 | assert can_edit is True |
| 329 | + |
| 330 | + |
| 331 | +# ═════════════════════════════════════════════════════════════════════════════ |
| 332 | +# Test: notification-users conf fallback parser + admin/superadmin discovery |
| 333 | +# (item G3 batch 1 coverage push, 2026-05-19) |
| 334 | +# |
| 335 | +# Covers lines 73-94, 234-244, 273-306 in bin/wl_rbac.py: |
| 336 | +# - read_notification_users_fallback: file parser for local/notification_users.conf |
| 337 | +# - get_admin_users: REST 200-branch correctly parses admins from entry list |
| 338 | +# - get_superadmin_users: full lifecycle (REST + conf fallback + empty default) |
| 339 | +# ═════════════════════════════════════════════════════════════════════════════ |
| 340 | + |
| 341 | + |
| 342 | +@pytest.mark.unit |
| 343 | +class TestReadNotificationUsersFallback: |
| 344 | + """Cover the conf-file parser at bin/wl_rbac.py:59-94.""" |
| 345 | + |
| 346 | + def test_admins_stanza_with_csv_users(self, tmp_path): |
| 347 | + """Parse [admins] stanza with comma-separated user list.""" |
| 348 | + conf = tmp_path / "notification_users.conf" |
| 349 | + conf.write_text( |
| 350 | + "[admins]\n" |
| 351 | + "users = alice, bob, charlie\n" |
| 352 | + ) |
| 353 | + with patch.object(wl_rbac_module, "_NOTIFICATION_USERS_CONF", str(conf)): |
| 354 | + result = read_notification_users_fallback("admins") |
| 355 | + assert result == ["alice", "bob", "charlie"] |
| 356 | + |
| 357 | + def test_superadmins_stanza_with_whitespace_users(self, tmp_path): |
| 358 | + """Parse [superadmins] stanza with whitespace-separated user list.""" |
| 359 | + conf = tmp_path / "notification_users.conf" |
| 360 | + conf.write_text( |
| 361 | + "[superadmins]\n" |
| 362 | + "users = root_admin super1 super2\n" |
| 363 | + ) |
| 364 | + with patch.object(wl_rbac_module, "_NOTIFICATION_USERS_CONF", str(conf)): |
| 365 | + result = read_notification_users_fallback("superadmins") |
| 366 | + assert result == ["root_admin", "super1", "super2"] |
| 367 | + |
| 368 | + def test_missing_file_returns_empty_list(self, tmp_path): |
| 369 | + """File absent → [] (silent failure per docstring).""" |
| 370 | + with patch.object( |
| 371 | + wl_rbac_module, |
| 372 | + "_NOTIFICATION_USERS_CONF", |
| 373 | + str(tmp_path / "nonexistent.conf"), |
| 374 | + ): |
| 375 | + result = read_notification_users_fallback("admins") |
| 376 | + assert result == [] |
| 377 | + |
| 378 | + def test_wrong_stanza_returns_empty(self, tmp_path): |
| 379 | + """File present with [admins] but caller asks for [superadmins] → [].""" |
| 380 | + conf = tmp_path / "notification_users.conf" |
| 381 | + conf.write_text("[admins]\nusers = alice, bob\n") |
| 382 | + with patch.object(wl_rbac_module, "_NOTIFICATION_USERS_CONF", str(conf)): |
| 383 | + result = read_notification_users_fallback("superadmins") |
| 384 | + assert result == [] |
| 385 | + |
| 386 | + def test_comments_and_blank_lines_are_skipped(self, tmp_path): |
| 387 | + """Lines starting with '#' and empty lines are ignored.""" |
| 388 | + conf = tmp_path / "notification_users.conf" |
| 389 | + conf.write_text( |
| 390 | + "# This is a comment\n" |
| 391 | + "\n" |
| 392 | + "[admins]\n" |
| 393 | + "# another comment\n" |
| 394 | + "users = alice, bob\n" |
| 395 | + " \n" # blank-with-whitespace |
| 396 | + ) |
| 397 | + with patch.object(wl_rbac_module, "_NOTIFICATION_USERS_CONF", str(conf)): |
| 398 | + result = read_notification_users_fallback("admins") |
| 399 | + assert result == ["alice", "bob"] |
| 400 | + |
| 401 | + def test_non_users_keys_in_stanza_are_ignored(self, tmp_path): |
| 402 | + """Keys other than 'users' in the matched stanza are silently skipped.""" |
| 403 | + conf = tmp_path / "notification_users.conf" |
| 404 | + conf.write_text( |
| 405 | + "[admins]\n" |
| 406 | + "notes = some metadata\n" |
| 407 | + "users = alice, bob\n" |
| 408 | + "owner = ops\n" |
| 409 | + ) |
| 410 | + with patch.object(wl_rbac_module, "_NOTIFICATION_USERS_CONF", str(conf)): |
| 411 | + result = read_notification_users_fallback("admins") |
| 412 | + assert result == ["alice", "bob"] |
| 413 | + |
| 414 | + def test_stanza_name_is_case_insensitive(self, tmp_path): |
| 415 | + """[ADMINS], [Admins], [admins] all match 'admins' lookup.""" |
| 416 | + conf = tmp_path / "notification_users.conf" |
| 417 | + conf.write_text( |
| 418 | + "[ADMINS]\n" |
| 419 | + "users = alice\n" |
| 420 | + ) |
| 421 | + with patch.object(wl_rbac_module, "_NOTIFICATION_USERS_CONF", str(conf)): |
| 422 | + result = read_notification_users_fallback("admins") |
| 423 | + assert result == ["alice"] |
| 424 | + |
| 425 | + def test_open_failure_returns_empty_list(self, tmp_path): |
| 426 | + """OSError during file read → silently returns [] (lines 93-94). |
| 427 | +
|
| 428 | + File exists check passes (isfile=True), but open() raises. |
| 429 | + Simulated by patching builtins.open to raise PermissionError |
| 430 | + for the conf path. |
| 431 | + """ |
| 432 | + conf = tmp_path / "notification_users.conf" |
| 433 | + conf.write_text("[admins]\nusers = alice\n") |
| 434 | + |
| 435 | + # Patch open() to raise on our specific file path, but pass |
| 436 | + # through for any other open (pytest internals, etc.). |
| 437 | + real_open = open |
| 438 | + |
| 439 | + def _selective_open(p, *args, **kwargs): |
| 440 | + if str(p) == str(conf): |
| 441 | + raise PermissionError("simulated EACCES") |
| 442 | + return real_open(p, *args, **kwargs) |
| 443 | + |
| 444 | + with patch.object(wl_rbac_module, "_NOTIFICATION_USERS_CONF", str(conf)), \ |
| 445 | + patch("builtins.open", side_effect=_selective_open): |
| 446 | + result = read_notification_users_fallback("admins") |
| 447 | + assert result == [] # PermissionError is an OSError subclass |
| 448 | + |
| 449 | + |
| 450 | +@pytest.mark.unit |
| 451 | +class TestGetRolesNonDictSession: |
| 452 | + """Cover the defensive check at bin/wl_rbac.py:168-169 (non-dict session).""" |
| 453 | + |
| 454 | + def test_request_with_non_dict_session_returns_empty_set(self): |
| 455 | + """request['session'] that isn't a dict → return set() (line 169).""" |
| 456 | + request = {"session": "not_a_dict"} # string instead of dict |
| 457 | + roles = get_roles(request) |
| 458 | + assert roles == set() |
| 459 | + |
| 460 | + |
| 461 | +@pytest.mark.unit |
| 462 | +class TestGetAdminUsersFromRest: |
| 463 | + """Cover the 200-branch of get_admin_users at bin/wl_rbac.py:233-244. |
| 464 | +
|
| 465 | + The existing TestGetAdminUsers class returns ``(200, ...)`` from the |
| 466 | + mock (an int), causing ``status.status`` to raise AttributeError → |
| 467 | + exception path → built-in fallback. These tests build a proper |
| 468 | + status object with the ``.status`` attribute so the 200 branch |
| 469 | + actually executes. |
| 470 | + """ |
| 471 | + |
| 472 | + def _build_mock_splunk(self, status_code, content_json): |
| 473 | + """Build a mock splunk.rest module whose simpleRequest returns |
| 474 | + (status_object, content) — matching the real Splunk SDK shape.""" |
| 475 | + mock_status = MagicMock() |
| 476 | + mock_status.status = status_code |
| 477 | + mock_splunk = MagicMock() |
| 478 | + mock_splunk.rest.simpleRequest.return_value = (mock_status, content_json) |
| 479 | + return mock_splunk |
| 480 | + |
| 481 | + def test_rest_200_parses_admin_roles_from_entries(self, tmp_path): |
| 482 | + """REST 200 with entries containing admin-tier users → returns them.""" |
| 483 | + content = json.dumps({ |
| 484 | + "entry": [ |
| 485 | + {"name": "alice", "content": {"roles": ["wl_admin"]}}, |
| 486 | + {"name": "carol", "content": {"roles": ["wl_editor"]}}, |
| 487 | + {"name": "dan", "content": {"roles": ["admin"]}}, |
| 488 | + ] |
| 489 | + }) |
| 490 | + mock_splunk = self._build_mock_splunk(200, content) |
| 491 | + # Point the conf-fallback path at a non-existent file so we know |
| 492 | + # the REST result wasn't masked by a conf fallback. |
| 493 | + with patch.dict('sys.modules', |
| 494 | + {'splunk': mock_splunk, 'splunk.rest': mock_splunk.rest}), \ |
| 495 | + patch.object(wl_rbac_module, "_NOTIFICATION_USERS_CONF", |
| 496 | + str(tmp_path / "absent.conf")): |
| 497 | + admins = get_admin_users("session_key") |
| 498 | + |
| 499 | + # Only users with role in ADMIN_ROLES should appear; carol (wl_editor) |
| 500 | + # is not an admin. |
| 501 | + assert "alice" in admins |
| 502 | + assert "dan" in admins |
| 503 | + assert "carol" not in admins |
| 504 | + |
| 505 | + def test_rest_200_with_no_admins_falls_to_conf(self, tmp_path): |
| 506 | + """REST 200 but no admin-tier entries → falls through to conf file.""" |
| 507 | + content = json.dumps({ |
| 508 | + "entry": [ |
| 509 | + {"name": "carol", "content": {"roles": ["wl_editor"]}}, |
| 510 | + ] |
| 511 | + }) |
| 512 | + mock_splunk = self._build_mock_splunk(200, content) |
| 513 | + conf = tmp_path / "notification_users.conf" |
| 514 | + conf.write_text("[admins]\nusers = configured_admin\n") |
| 515 | + with patch.dict('sys.modules', |
| 516 | + {'splunk': mock_splunk, 'splunk.rest': mock_splunk.rest}), \ |
| 517 | + patch.object(wl_rbac_module, "_NOTIFICATION_USERS_CONF", str(conf)): |
| 518 | + admins = get_admin_users("session_key") |
| 519 | + # REST returned 200 but no admins; conf is used. |
| 520 | + assert admins == ["configured_admin"] |
| 521 | + |
| 522 | + |
| 523 | +@pytest.mark.unit |
| 524 | +class TestGetSuperadminUsers: |
| 525 | + """Cover get_superadmin_users at bin/wl_rbac.py:259-306 (zero prior tests).""" |
| 526 | + |
| 527 | + def _build_mock_splunk(self, status_code, content_json): |
| 528 | + mock_status = MagicMock() |
| 529 | + mock_status.status = status_code |
| 530 | + mock_splunk = MagicMock() |
| 531 | + mock_splunk.rest.simpleRequest.return_value = (mock_status, content_json) |
| 532 | + return mock_splunk |
| 533 | + |
| 534 | + def test_rest_200_parses_superadmin_roles(self, tmp_path): |
| 535 | + """REST 200 with superadmin-tier entries → returns them.""" |
| 536 | + content = json.dumps({ |
| 537 | + "entry": [ |
| 538 | + {"name": "root", "content": {"roles": ["wl_superadmin"]}}, |
| 539 | + {"name": "alice", "content": {"roles": ["wl_admin"]}}, |
| 540 | + {"name": "sa2", "content": {"roles": ["sc_admin"]}}, |
| 541 | + ] |
| 542 | + }) |
| 543 | + mock_splunk = self._build_mock_splunk(200, content) |
| 544 | + with patch.dict('sys.modules', |
| 545 | + {'splunk': mock_splunk, 'splunk.rest': mock_splunk.rest}), \ |
| 546 | + patch.object(wl_rbac_module, "_NOTIFICATION_USERS_CONF", |
| 547 | + str(tmp_path / "absent.conf")): |
| 548 | + sa = get_superadmin_users("session_key") |
| 549 | + # Verify membership against SUPERADMIN_ROLES; alice (wl_admin only) is |
| 550 | + # admin-tier but not necessarily superadmin — depends on role config. |
| 551 | + # The contract is: name is in result IFF its roles intersect SUPERADMIN_ROLES. |
| 552 | + for entry_name in sa: |
| 553 | + assert entry_name in ("root", "sa2", "alice") # superset check |
| 554 | + # root and sa2 are clearly in via their roles |
| 555 | + assert "root" in sa or "sa2" in sa |
| 556 | + |
| 557 | + def test_empty_session_no_rest_no_conf_returns_empty(self, tmp_path): |
| 558 | + """No session + no conf file → returns [] (vs ['admin'] in get_admin_users).""" |
| 559 | + with patch.object(wl_rbac_module, "_NOTIFICATION_USERS_CONF", |
| 560 | + str(tmp_path / "absent.conf")): |
| 561 | + assert get_superadmin_users("") == [] |
| 562 | + |
| 563 | + def test_empty_session_uses_conf_fallback(self, tmp_path): |
| 564 | + """No session but conf exists → conf list returned.""" |
| 565 | + conf = tmp_path / "notification_users.conf" |
| 566 | + conf.write_text("[superadmins]\nusers = root, godmode\n") |
| 567 | + with patch.object(wl_rbac_module, "_NOTIFICATION_USERS_CONF", str(conf)): |
| 568 | + assert get_superadmin_users("") == ["root", "godmode"] |
| 569 | + |
| 570 | + def test_rest_exception_falls_through_to_conf(self, tmp_path): |
| 571 | + """REST raises → exception swallowed → conf fallback used.""" |
| 572 | + mock_splunk = MagicMock() |
| 573 | + mock_splunk.rest.simpleRequest.side_effect = RuntimeError("network down") |
| 574 | + conf = tmp_path / "notification_users.conf" |
| 575 | + conf.write_text("[superadmins]\nusers = sa_from_conf\n") |
| 576 | + with patch.dict('sys.modules', |
| 577 | + {'splunk': mock_splunk, 'splunk.rest': mock_splunk.rest}), \ |
| 578 | + patch.object(wl_rbac_module, "_NOTIFICATION_USERS_CONF", str(conf)): |
| 579 | + assert get_superadmin_users("token") == ["sa_from_conf"] |
| 580 | + |
| 581 | + def test_rest_200_no_superadmins_falls_to_conf(self, tmp_path): |
| 582 | + """REST 200 with no superadmin-tier entries → conf fallback.""" |
| 583 | + content = json.dumps({ |
| 584 | + "entry": [{"name": "alice", "content": {"roles": ["wl_editor"]}}] |
| 585 | + }) |
| 586 | + mock_splunk = self._build_mock_splunk(200, content) |
| 587 | + conf = tmp_path / "notification_users.conf" |
| 588 | + conf.write_text("[superadmins]\nusers = configured_sa\n") |
| 589 | + with patch.dict('sys.modules', |
| 590 | + {'splunk': mock_splunk, 'splunk.rest': mock_splunk.rest}), \ |
| 591 | + patch.object(wl_rbac_module, "_NOTIFICATION_USERS_CONF", str(conf)): |
| 592 | + assert get_superadmin_users("session") == ["configured_sa"] |
0 commit comments