Unit tests for osism/tasks/conductor/sonic/cache.py

[berendt]

Background

Follow-up to #2192 (foundation) and PR #2193 (pytest + Zuul infrastructure). Part of Tier 2 (#2199). cache.py is a thin wrapper around utils.nb.dcim.interfaces.filter() that adds a thread-local cache and a lock — small, self-contained, and easy to test once NetBox is mocked.

Covering this module early is useful because the connections / device / netbox modules in later Tier 2 issues all go through get_cached_device_interfaces(), so a stable behaviour contract here helps the next tests.

Scope

Add tests/unit/tasks/conductor/sonic/test_cache.py covering the InterfaceCache class and the four module-level helpers in osism/tasks/conductor/sonic/cache.py.

Test targets

InterfaceCache class — cache.py:12

__init__() — cache.py:15

• After construction, _cache == {} and _lock is a threading.Lock instance (assert via isinstance(cache._lock, type(threading.Lock())) or just cache._lock.acquire(blocking=False) round-trip).

get_device_interfaces(device_id) — cache.py:19

Patch osism.tasks.conductor.sonic.cache.utils.nb.

• Cache miss: nb.dcim.interfaces.filter(device_id=…) is called once; result (an iterable of mocked interfaces) is converted to a list and returned.
• Subsequent call with same device_id: NetBox is not queried again — same list returned (cache hit). Assert nb.dcim.interfaces.filter.call_count == 1.
• Different device_id: NetBox queried again; both entries coexist in the cache.
• NetBox raises Exception on first call: returns [], cache stores [], no further calls on subsequent invocations (cached empty list).
• Verify the call uses the keyword form device_id=device_id (not positional).

clear() — cache.py:49

• After populating two devices, clear() empties _cache and a subsequent get_device_interfaces re-queries NetBox.

get_cache_stats() — cache.py:56

• Empty cache → {\"cached_devices\": 0, \"total_interfaces\": 0}
• Two devices with 3 and 5 interfaces respectively → {\"cached_devices\": 2, \"total_interfaces\": 8}

Lock semantics

• Replace cache._lock with a MagicMock(wraps=threading.Lock()) (or assign a real Lock and patch its __enter__/__exit__ via a wrapper) and assert that get_device_interfaces, clear, and get_cache_stats each acquire it exactly once. A simple smoke test that two threads can call get_device_interfaces for the same device_id without raising is sufficient — full concurrency proofs are out of scope.

Module-level helpers

get_interface_cache() — cache.py:76

• Two consecutive calls return the same InterfaceCache instance (thread-local lazy init).
• Calling from a different thread (use threading.Thread) returns a different instance — assert via a queue/list collected by the worker.

get_cached_device_interfaces(device_id) — cache.py:87

• Delegates to the thread-local cache: patch the cache's get_device_interfaces and assert it is called with the right device_id; return value is passed through.

clear_interface_cache() — cache.py:100

• No-op when no thread-local cache exists yet (call it on a fresh thread, must not raise).
• After get_cached_device_interfaces populated the cache, calling clear_interface_cache() empties it (verify via get_interface_cache_stats() returning zeros).

get_interface_cache_stats() — cache.py:106

• Returns None when no thread-local cache has been created in the current thread.
• Returns the dict from InterfaceCache.get_cache_stats() after the cache has been used.

Mocking hints

• Patch osism.tasks.conductor.sonic.cache.utils.nb (the module-level reference inside cache.py). Returning a generator from filter() is fine — the code wraps it in list(…).
• Use MagicMock() for the individual interface objects; tests don't need real attributes for cache behaviour.
• Reset thread-local state between tests with a fixture that calls clear_interface_cache() in a yield/teardown block, and clears _thread_local.__dict__ if needed, so cross-test pollution does not occur.
• Real threading.Lock and threading.local are fine — no need to mock them.

Definition of Done

• tests/unit/tasks/__init__.py, tests/unit/tasks/conductor/__init__.py, tests/unit/tasks/conductor/sonic/__init__.py created (if not already)
• tests/unit/tasks/conductor/sonic/test_cache.py covers all listed cases
• pytest --cov=osism.tasks.conductor.sonic.cache shows 100 %
• pipenv run pytest tests/unit/tasks/conductor/sonic/test_cache.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.