Unit tests for osism/utils/__init__.py — connection initialization

[berendt]

Background

Follow-up to #2192 (foundation) and PR #2193 (pytest + Zuul infrastructure). Part of Tier 4 (#2199). osism/utils/__init__.py is 672 LOC and bundles several distinct concerns. Splitting it across three sub-issues keeps each test module manageable:

• This issue: connection / client initialization and the lazy __getattr__ indirection.
• Companion issue: task output streaming + ansible/redis helpers.
• Companion issue: concurrency primitives (RedisSemaphore, redlock) and task locks.

Scope

Add tests/unit/utils/test_init_connections.py covering the connection-init helpers and lazy attribute resolution in osism/utils/__init__.py.

Test targets

_init_redis() — __init__.py:22

Patch osism.utils.settings for REDIS_HOST / REDIS_PORT / REDIS_DB, and patch the imported redis.Redis class via mocker.patch("redis.Redis") so no real connection is made. Reset module global _redis between tests via autouse fixture.

• First call → instantiates Redis(host, port, db, socket_keepalive=True) and calls .ping(); caches the instance in _redis
• Second call → returns the cached instance (no second Redis(...) instantiation)
• ping() raises → exception propagated (no swallowing)

_init_nb() — __init__.py:37

Patch get_netbox_connection. Reset _nb/_nb_initialized between tests.

• First call → delegates to get_netbox_connection(NETBOX_URL, NETBOX_TOKEN, IGNORE_SSL_ERRORS)
• Sets _nb_initialized=True even if get_netbox_connection returns None (e.g. URL/token missing)
• Subsequent calls → return cached _nb (no second call to get_netbox_connection)

_init_secondary_nb_list() — __init__.py:47

Patch osism.utils.settings.NETBOX_SECONDARIES and get_netbox_connection. Reset _secondary_nb_list and _secondary_nb_initialized between tests.

• Two valid entries → returns list with two NetBox API objects (verify two get_netbox_connection calls with the right args including IGNORE_SSL_ERRORS=True default)
• Entry with NETBOX_NAME and NETBOX_SITE → those attributes are set on the returned object
• Setting is empty string / parses to None → TypeError caught (non-list), _secondary_nb_list=[], error logged
• Setting parses to a dict (not a list) → TypeError raised internally, caught, logged
• Element is not a dict (e.g. a string) → TypeError caught, logged
• Element with unknown key (e.g. NETBOX_FOO) → ValueError caught, logged
• Element missing NETBOX_URL (or empty) → ValueError caught, logged
• Element missing NETBOX_TOKEN (or empty/whitespace-only after strip) → ValueError caught, logged
• IGNORE_SSL_ERRORS defaulted to True when omitted
• NETBOX_TOKEN with surrounding whitespace → stripped before passing
• Invalid YAML → yaml.YAMLError caught, _secondary_nb_list=[]
• After error → _secondary_nb_initialized=True (cached negative result)

_get_timeout_http_adapter_class() — __init__.py:180

• First call → returns the _TimeoutHTTPAdapter class (subclass of HTTPAdapter)
• Second call → returns the same class (cached on module global)
• Instance: send(request, timeout=None) falls back to self.timeout; existing timeout=5 kwarg is preserved

NetBoxSessionManager.get_session(ignore_ssl_errors, timeout) — __init__.py:224

Patch requests.Session and urllib3.disable_warnings. Reset NetBoxSessionManager._session/_lock between tests.

• First call → creates Session(), mounts the timeout adapter on http:// and https:// (verify mount calls)
• ignore_ssl_errors=True → urllib3.disable_warnings() called and session.verify=False
• ignore_ssl_errors=False → session.verify not modified, disable_warnings not called
• Subsequent call → returns cached session (no second Session() instantiation)
• Custom timeout=30 → propagated into the adapter's timeout

NetBoxSessionManager.close_session() — __init__.py:255

• After call: _session.close() invoked and _session=None
• Calling again with no session → no-op (no error)

cleanup_netbox_sessions() — __init__.py:263

• Delegates to NetBoxSessionManager.close_session() (verify the call)

get_netbox_connection(netbox_url, netbox_token, ignore_ssl_errors=False, timeout=20) — __init__.py:268

Patch pynetbox.api, NetBoxSessionManager.get_session, atexit.register. Reset _cleanup_registered between tests.

• Both URL and token provided → creates pynetbox.api(url, token=token), sets nb.http_session to the shared session, registers atexit.register(cleanup_netbox_sessions) exactly once across multiple invocations
• URL missing → returns None, no pynetbox.api call
• Token missing → returns None
• pynetbox.api returns falsy → atexit.register not invoked, but caller still gets the falsy value back
• ignore_ssl_errors/timeout propagated to NetBoxSessionManager.get_session
• Second call → atexit.register not called again (_cleanup_registered short-circuit)

get_openstack_connection() — __init__.py:304

Patch openstack.connect and keystoneauth1.exceptions.auth_plugins.MissingRequiredOptions (just import).

• openstack.connect() succeeds → returns the connection
• openstack.connect() raises MissingRequiredOptions → wrapped in RuntimeError with "missing required authentication options"
• Other exceptions propagate unchanged

__getattr__(name) — __init__.py:659

Reset globals() redis/nb/secondary_nb_list between tests.

• osism.utils.redis → calls _init_redis, caches in globals()["redis"], second access uses the cached attribute (no second _init_redis call)
• osism.utils.nb → calls _init_nb, caches in globals()["nb"]
• osism.utils.secondary_nb_list → calls _init_secondary_nb_list, caches
• Unknown name (osism.utils.foo) → raises AttributeError with the expected message

Mocking hints

• Use a fixture (autouse=True) to reset module globals: _redis, _nb, _nb_initialized, _secondary_nb_list, _secondary_nb_initialized, _cleanup_registered, _TimeoutHTTPAdapterClass, NetBoxSessionManager._session, NetBoxSessionManager._lock, plus delete globals()["redis"|"nb"|"secondary_nb_list"] if present.
• Patch settings via mocker.patch.multiple("osism.utils.settings", NETBOX_URL="https://nb", NETBOX_TOKEN="t", ...).
• For NETBOX_SECONDARIES, set the setting to a YAML string (the production code calls yaml.safe_load(settings.NETBOX_SECONDARIES)).

Definition of Done

• tests/unit/utils/test_init_connections.py created
• All listed cases covered
• pytest --cov=osism.utils for the targeted helpers ≥ 90 %
• pipenv run pytest tests/unit/utils/test_init_connections.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.