Unit tests for osism/tasks/netbox.py

[berendt]

Background

Follow-up to #2192 (foundation) and PR #2193 (pytest + Zuul infrastructure). Part of Tier 5 (#2199). osism/tasks/netbox.py (410 LOC) defines the Celery tasks of the netbox worker: setting device custom fields (maintenance, provision_state, power_state) on the primary NetBox and on all matching NETBOX_SECONDARIES instances (Redlock-guarded, semaphore-limited), simple get_* lookup tasks, the manage task wrapping /usr/local/bin/netbox-manager via run_command, and ping. Two private helpers (_update_netbox_device_field, _matches_netbox_filter) carry most of the logic.

Scope

Add tests/unit/tasks/test_netbox.py covering osism/tasks/netbox.py. No existing coverage: tests/unit/tasks/conductor/test_netbox.py covers osism/tasks/conductor/netbox.py (get_device_oob_ip, query-list helpers) and tests/unit/commands/test_netbox.py covers osism/commands/netbox.py — both are different modules, no overlap.

All tasks are bound Celery tasks (bind=True); invoke them directly (e.g. netbox.set_maintenance("node-1", True)) — Celery binds self on direct calls, no broker needed.

The three set_* tasks are structurally identical (same lock/filter/update/skip flow, only the field name and log wording differ). Parametrize over (task, field_name) instead of triplicating every test — same approach as QUERY_LIST_VARIANTS in tests/unit/tasks/conductor/test_netbox.py.

Test targets

_update_netbox_device_field(nb, device_name, field_name, value) — netbox.py:19

Patch:

• osism.tasks.netbox.utils.create_netbox_semaphore → return a MagicMock() (supports the with protocol)

Pass a MagicMock as nb (no patching of utils.nb needed here).

• Device found → device.custom_fields.update({field_name: value}) and device.save() called, returns True
• nb.dcim.devices.get returns None → returns False, no save, no error log (the caller logs)
• requests.exceptions.ConnectTimeout raised → returns False, "Connection timeout" logged
• requests.exceptions.ReadTimeout raised → returns False, "Request timeout" logged (use ReadTimeout, not ConnectTimeout — ConnectTimeout inherits from both ConnectionError and Timeout and is caught by the first handler)
• requests.exceptions.ConnectionError raised → returns False, "Connection error" logged
• requests.exceptions.HTTPError (generic RequestException) raised → returns False, "Request error" logged
• Exception raised by device.save() (not only by get) → same handling, returns False
• Semaphore is created with nb.base_url and entered/exited around the API call (__enter__/__exit__ asserted)

_matches_netbox_filter(nb, netbox_filter, is_primary=False) — netbox.py:66

No patching; pass SimpleNamespace/MagicMock objects with base_url and optional netbox_name/netbox_site attributes.

• netbox_filter=None and netbox_filter="" → True (empty filter matches everything)
• is_primary=True, filter "primary" → True; also case-insensitive ("PRIMARY") and substring ("primary-dc" contains "primary")
• is_primary=False, filter "primary", URL/name/site without "primary" → False
• Filter substring of base_url, case-insensitive ("NETBOX.example" vs https://netbox.example.com) → True
• Object without netbox_name/netbox_site attributes (primary instances) → no AttributeError, falls through to False when URL does not match
• netbox_name set and matching (case-insensitive substring) → True
• netbox_site set and matching → True
• Nothing matches → False

run(action, arguments) — netbox.py:104

• Patch osism.tasks.netbox.utils.check_task_lock_and_exit; direct call returns None, lock check invoked once
• Lock check raises SystemExit (locked) → propagates

set_maintenance / set_provision_state / set_power_state — netbox.py:117 / netbox.py:190 / netbox.py:263

Patch (parametrized over the three tasks and their field names maintenance / provision_state / power_state):

• osism.tasks.netbox.utils.check_task_lock_and_exit
• osism.tasks.netbox.utils.create_redlock → MagicMock whose .acquire(timeout=120) returns True/False
• osism.tasks.netbox._update_netbox_device_field
• osism.utils.nb — lazy attribute, patch with mocker.patch("osism.utils.nb", new=MagicMock(), create=True) (same convention as tests/unit/tasks/conductor/sonic/conftest.py)

For secondaries, prefer passing the secondary_nb_list=... parameter over patching osism.utils.secondary_nb_list.

• lock.acquire returns False → returns False, "Could not acquire lock" logged, _update_netbox_device_field never called, lock.release() not called
• create_redlock called with key=f"lock_osism_tasks_netbox_{device_name}", auto_release_time=300
• Happy path, netbox_filter=None, secondary_nb_list=[] → update called once with (utils.nb, device_name, field, state), returns True, lock.release() called
• _update_netbox_device_field returns False → "Could not set ..." error logged, task still returns True
• netbox_filter not matching the primary (e.g. filter that matches nothing) → primary skipped (update not called for utils.nb), still returns True
• Two secondaries passed via secondary_nb_list, one matching the filter, one not → update called only for the matching one (plus/minus primary depending on filter)
• secondary_nb_list=None → falls back to utils.secondary_nb_list (patch the lazy attribute with create=True to a one-element list and assert it is processed)
• _update_netbox_device_field raises → exception propagates but lock.release() is still called (finally)
• set_power_state only: state=None → converted to "n/a" (update called with "n/a"); conversion happens before the lock-check
• One representative test that check_task_lock_and_exit raising SystemExit aborts before create_redlock is called

get_location_id / get_rack_id — netbox.py:340 / netbox.py:352

Patch osism.utils.nb (lazy, create=True). Parametrize over (task, endpoint) — dcim.locations vs dcim.racks:

• Object found → returns .id
• get returns None → returns None
• get raises ValueError (pynetbox raises this for ambiguous matches) → returns None

get_devices / get_device_by_name / get_interfaces_by_device / get_addresses_by_device_and_interface — netbox.py:364 / netbox.py:369 / netbox.py:374 / netbox.py:379

Thin pass-throughs; patch osism.utils.nb and assert delegation:

• get_devices(role="server", site="x") → nb.dcim.devices.filter(role="server", site="x"), return value passed through
• get_device_by_name("n1") → nb.dcim.devices.get(name="n1")
• get_interfaces_by_device("n1") → nb.dcim.interfaces.filter(device="n1")
• get_addresses_by_device_and_interface("n1", "eth0") → nb.dcim.addresses.filter(device="n1", interface="eth0") (test as written — the module uses dcim.addresses)

manage(*arguments, publish=True, locking=False, auto_release_time=3600) — netbox.py:384

Patch:

• osism.tasks.netbox.utils.check_task_lock_and_exit
• osism.tasks.netbox.run_command (imported into the module namespace via from osism.tasks import ... run_command)
• osism.tasks.netbox.settings.NETBOX_URL / NETBOX_TOKEN / IGNORE_SSL_ERRORS

Cases:

• Env dict passed to run_command contains exactly NETBOX_MANAGER_URL, NETBOX_MANAGER_TOKEN, NETBOX_MANAGER_IGNORE_SSL_ERRORS, NETBOX_MANAGER_VERBOSE="true"; all values str()-coerced (e.g. IGNORE_SSL_ERRORS=True → "True")
• Command is /usr/local/bin/netbox-manager; positional *arguments forwarded; publish/locking/auto_release_time forwarded (defaults and explicit values)
• Returns run_command's return value
• First positional argument is self.request.id — on a direct call this is None; that is fine since run_command is mocked (alternatively use manage.apply(args=[...]) for a real task id)
• Lock check raising SystemExit → run_command never called

ping() — netbox.py:407

• Patch osism.utils.nb; returns utils.nb.status() result unchanged

setup_periodic_tasks(sender, **kwargs) — netbox.py:15

• No-op signal handler; call once with sender=None for line coverage

Mocking hints

• utils.nb and utils.secondary_nb_list are lazy module attributes resolved via osism.utils.__getattr__ — always patch with create=True and target osism.utils.nb (never trigger the real __getattr__, it would open a NetBox connection). See the mock_nb fixtures in tests/unit/tasks/conductor/sonic/conftest.py and tests/unit/tasks/conductor/test_netbox.py:59.
• utils.check_task_lock_and_exit calls exit(1) when locked; in tests, set side_effect=SystemExit(1) on the patched mock for the "locked" cases and assert with pytest.raises(SystemExit).
• The Redlock mock needs only acquire(timeout=120) -> bool and release(); assert the exact timeout=120 kwarg.
• Build secondary instances as MagicMock(base_url="https://nb2.example.com") and set netbox_name / netbox_site attributes explicitly where a test needs them; use spec/SimpleNamespace for the "attribute absent" cases so getattr(..., None) actually returns None.
• For the set_* tasks, patching osism.tasks.netbox._update_netbox_device_field keeps the lock/filter tests independent of the helper, which is tested separately.
• Importing osism.tasks.netbox instantiates Celery("netbox") with config_from_object(Config) — config only, no broker connection; safe in unit tests (same pattern as the other osism.tasks modules already imported by existing tests).

Definition of Done

• tests/unit/tasks/test_netbox.py created
• All listed cases covered
• pytest --cov=osism.tasks.netbox shows ≥ 90 %
• pipenv run pytest tests/unit/tasks/test_netbox.py passes locally
• flake8, mypy, python-black remain green
• Zuul job python-osism-unit-tests passes

Dependencies

• Part of Meta: Unit test coverage for osism/ #2199 (meta). Blocked by Introduce foundation for unit tests (pytest + Zuul CI integration) #2192 / PR Introduce foundation for unit tests (pytest + Zuul CI) #2193.