Skip to content

Unit tests for osism/tasks/conductor/sonic/connections.py #2205

Description

@berendt

Background

Follow-up to #2192 (foundation) and PR #2193 (pytest + Zuul infrastructure). Part of Tier 2 (#2199). connections.py is the topology layer used by both BGP and config generation. The functions are mostly readers around NetBox connected_endpoints, plus a BFS over interconnected devices.

This issue covers the four pure traversal helpers and the BFS — the heavier get_device_bgp_neighbors_via_loopback and get_connected_interface_ipv4_address (lots of branching across IPAM/FHRP) belong to a Tier 3 issue.

Scope

Add tests/unit/tasks/conductor/sonic/test_connections.py covering five functions in osism/tasks/conductor/sonic/connections.py:

  • get_connected_device_via_interfaceconnections.py:23
  • get_connected_interfacesconnections.py:63
  • get_connected_device_for_sonic_interfaceconnections.py:116
  • get_connected_device_for_port_channelconnections.py:153
  • find_interconnected_devicesconnections.py:216
  • load_vip_addresses_cache / clear_vip_addresses_cacheconnections.py:299 / connections.py:317

Out of scope (Tier 3): get_device_bgp_neighbors_via_loopback, get_connected_interface_ipv4_address.

Test targets

get_connected_device_via_interface(interface, source_device_id)connections.py:23

Pure logic over interface attributes — use SimpleNamespace.

  • interface.mgmt_only=True → returns None
  • interface.connected_endpoints is empty/missing → returns None
  • interface.connected_endpoints_reachable=False → returns None
  • One endpoint, endpoint.device.id == source_device_id → returns None (excludes source)
  • One endpoint, endpoint.device.id != source_device_id → returns that device
  • Multiple endpoints, first one is the source → returns the next device
  • Endpoint without device attribute → caught by the generic exception handler, returns None

get_connected_interfaces(device, portchannel_info=None)connections.py:63

Patch get_cached_device_interfaces, get_connected_device_via_interface, and convert_netbox_interface_to_sonic (all in the connections module namespace).

  • Device with two non-mgmt interfaces, both connected → both names appear in the first set; second set (port channels) is empty when portchannel_info=None
  • Mgmt-only interface is skipped (not in either set)
  • Interface not connected (get_connected_device_via_interfaceNone) → not in either set
  • portchannel_info={\"PortChannel1\": {\"members\": [\"Ethernet0\"]}} and Ethernet0 is connected → \"Ethernet0\" in first set, \"PortChannel1\" in second set
  • get_connected_device_via_interface raises → caught by inner try/except, interface skipped, warning logged
  • Returns are set instances (not list/tuple)

get_connected_device_for_sonic_interface(device, sonic_interface_name)connections.py:116

  • sonic_interface_name=\"PortChannel5\" → delegates to get_connected_device_for_port_channel (patch and assert call args + pass-through)
  • Regular interface (\"Ethernet0\") where the cached interface list contains a NetBox interface that converts to \"Ethernet0\" → returns the connected device
  • Regular interface that does not match any cached interface → returns None
  • Underlying call raises → returns None (debug-logged)

get_connected_device_for_port_channel(device, portchannel_name)connections.py:153

Patch osism.tasks.conductor.sonic.interface.detect_port_channels (imported lazily inside the function).

  • Port channel not in detect_port_channels(device) result → returns None
  • Port channel present but members empty → returns None
  • Port channel with one member; the matching NetBox interface is connected → returns that connected device
  • Port channel with two members; first member not connected, second member connected → returns the second device
  • All members unconnected → returns None

find_interconnected_devices(devices, target_roles=[\"spine\", \"superspine\"])connections.py:216

The BFS core. Build small device stubs with SimpleNamespace(id=…, role=SimpleNamespace(slug=…)) and patch get_cached_device_interfaces + get_connected_device_via_interface to encode the topology.

  • Empty input → returns []
  • No devices match target_roles → returns []
  • One spine, no peers → returns [] (groups of size 1 are dropped)
  • Two spines connected to each other → returns one group [spineA, spineB]
  • Three spines in a line (A↔B, B↔C) → returns one group of three
  • Two pairs of disconnected spines → returns two groups of two
  • One spine connected to a leaf (leaf not in target_roles) → leaf is filtered, spine is alone → []
  • Two spines and two superspines all interconnected → spines and superspines are returned as separate groups (filtering happens per role)
  • get_cached_device_interfaces raises for a device → that device is skipped (warning logged), BFS continues for others
  • Custom target_roles=[\"leaf\"] → only leaves grouped

load_vip_addresses_cache() / clear_vip_addresses_cache()connections.py:299 / connections.py:317

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

  • load_vip_addresses_cache() calls nb.ipam.ip_addresses.filter(role=\"vip\") and stores the resulting list in the module-level _vip_addresses_cache
  • NetBox raises → cache is set to [] (warning logged, no exception propagated)
  • clear_vip_addresses_cache() resets _vip_addresses_cache to None
  • After clear, calling load again triggers a fresh NetBox call

Mocking hints

  • Patch the symbols in the connections module namespace, not in their original modules — i.e. osism.tasks.conductor.sonic.connections.get_cached_device_interfaces, …connections.get_connected_device_via_interface, …connections.convert_netbox_interface_to_sonic, …connections.utils.nb.
  • detect_port_channels is imported lazily inside get_connected_device_for_port_channel — patch osism.tasks.conductor.sonic.interface.detect_port_channels.
  • For BFS tests, the topology is easiest expressed as a dict[device_id, list[interface_stub]] returned by the mocked get_cached_device_interfaces, plus a small lookup dict for get_connected_device_via_interface.
  • Use a fixture that resets the module-level _vip_addresses_cache between tests.

Definition of Done

  • tests/unit/tasks/conductor/sonic/test_connections.py covers all listed cases
  • pytest --cov=osism.tasks.conductor.sonic.connections shows ≥ 90 % for the targeted functions (heavier helpers tracked in Tier 3 may stay uncovered for now)
  • pipenv run pytest tests/unit/tasks/conductor/sonic/test_connections.py passes locally
  • flake8, mypy, python-black remain green
  • Zuul job python-osism-unit-tests passes

Dependencies

Metadata

Metadata

Assignees

No one assigned

    Labels

    enhancementNew feature or request

    Type

    No type
    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions