From 13e53cb830883b82ef2171d56d405fad4aa26783 Mon Sep 17 00:00:00 2001 From: Christian Berendt Date: Sat, 25 Apr 2026 12:28:44 +0200 Subject: [PATCH] Add unit tests for osism/tasks/conductor/sonic/{bgp,constants} - bgp.py: calculate_local_asn_from_ipv4 (docstring example, CIDR stripping, custom prefix, zero-padding, boundary octets, invalid formats), calculate_minimum_as_for_group (multiple/empty groups, invalid IPs skipped with logger.debug, None/empty primary_ip4 silently skipped, custom prefix, str() conversion of non-str IPs), and find_interconnected_spine_groups delegation to connections.find_interconnected_devices. - constants.py: invariants of module-level tables (default values, sorted-and-deduped DEFAULT_SONIC_ROLES, sample port-speed mappings across families, HIGH_SPEED_PORTS consistency with PORT_TYPE_TO_SPEED_MAP, HWSKU prefix and uniqueness). Importing these tests triggers osism/tasks/conductor/__init__, which transitively loads osism/tasks/conductor/utils.py with its module-level ``from ansible import ...`` statements. ansible-core lives in the optional ``[ansible]`` extra and is not installed in the unit-test environment, so tests/conftest.py registers lightweight sys.modules stubs for ``ansible``, ``ansible.constants``, ``ansible.errors`` and ``ansible.parsing.vault`` before collection. The stubs only satisfy the import statements and are skipped when the real package is installed. Closes #2198 AI-assisted: Claude Code Signed-off-by: Christian Berendt --- tests/conftest.py | 72 ++++++ tests/unit/tasks/__init__.py | 1 + tests/unit/tasks/conductor/__init__.py | 1 + tests/unit/tasks/conductor/sonic/__init__.py | 1 + tests/unit/tasks/conductor/sonic/test_bgp.py | 239 ++++++++++++++++++ .../tasks/conductor/sonic/test_constants.py | 150 +++++++++++ 6 files changed, 464 insertions(+) create mode 100644 tests/unit/tasks/__init__.py create mode 100644 tests/unit/tasks/conductor/__init__.py create mode 100644 tests/unit/tasks/conductor/sonic/__init__.py create mode 100644 tests/unit/tasks/conductor/sonic/test_bgp.py create mode 100644 tests/unit/tasks/conductor/sonic/test_constants.py diff --git a/tests/conftest.py b/tests/conftest.py index 988131360..c04f5171e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1 +1,73 @@ # SPDX-License-Identifier: Apache-2.0 + +"""Pytest configuration shared by all unit tests. + +``osism/tasks/conductor/utils.py`` imports ``ansible`` at module level, so +importing any submodule under ``osism.tasks.conductor`` triggers that import +via the package ``__init__`` chain. ``ansible-core`` lives in the optional +``[ansible]`` extra and is not installed in the unit-test environment. + +To keep test setup lightweight we register stub modules in ``sys.modules`` +before tests are collected. The stubs only need to satisfy the +``from ansible... import ...`` statements; no runtime behaviour is emulated. +Tests that genuinely exercise ansible-using code paths must install +``ansible-core`` (or replace these stubs with mocks at the test level). +""" + +import sys +import types + + +def _install_ansible_stubs() -> None: + try: + import ansible # noqa: F401 + except ImportError: + pass + else: + return + + ansible_mod = types.ModuleType("ansible") + ansible_mod.__path__ = [] # type: ignore[attr-defined] + + constants = types.ModuleType("ansible.constants") + setattr(constants, "DEFAULT_VAULT_ID_MATCH", "default") + + errors = types.ModuleType("ansible.errors") + + class AnsibleError(Exception): + pass + + setattr(errors, "AnsibleError", AnsibleError) + + parsing = types.ModuleType("ansible.parsing") + parsing.__path__ = [] # type: ignore[attr-defined] + + vault = types.ModuleType("ansible.parsing.vault") + + class VaultLib: + def __init__(self, *args, **kwargs): + pass + + def is_encrypted(self, *args, **kwargs): + return False + + class VaultSecret: + def __init__(self, *args, **kwargs): + pass + + setattr(vault, "VaultLib", VaultLib) + setattr(vault, "VaultSecret", VaultSecret) + + setattr(ansible_mod, "constants", constants) + setattr(ansible_mod, "errors", errors) + setattr(ansible_mod, "parsing", parsing) + setattr(parsing, "vault", vault) + + sys.modules["ansible"] = ansible_mod + sys.modules["ansible.constants"] = constants + sys.modules["ansible.errors"] = errors + sys.modules["ansible.parsing"] = parsing + sys.modules["ansible.parsing.vault"] = vault + + +_install_ansible_stubs() diff --git a/tests/unit/tasks/__init__.py b/tests/unit/tasks/__init__.py new file mode 100644 index 000000000..988131360 --- /dev/null +++ b/tests/unit/tasks/__init__.py @@ -0,0 +1 @@ +# SPDX-License-Identifier: Apache-2.0 diff --git a/tests/unit/tasks/conductor/__init__.py b/tests/unit/tasks/conductor/__init__.py new file mode 100644 index 000000000..988131360 --- /dev/null +++ b/tests/unit/tasks/conductor/__init__.py @@ -0,0 +1 @@ +# SPDX-License-Identifier: Apache-2.0 diff --git a/tests/unit/tasks/conductor/sonic/__init__.py b/tests/unit/tasks/conductor/sonic/__init__.py new file mode 100644 index 000000000..988131360 --- /dev/null +++ b/tests/unit/tasks/conductor/sonic/__init__.py @@ -0,0 +1 @@ +# SPDX-License-Identifier: Apache-2.0 diff --git a/tests/unit/tasks/conductor/sonic/test_bgp.py b/tests/unit/tasks/conductor/sonic/test_bgp.py new file mode 100644 index 000000000..433efcbac --- /dev/null +++ b/tests/unit/tasks/conductor/sonic/test_bgp.py @@ -0,0 +1,239 @@ +# SPDX-License-Identifier: Apache-2.0 + +from types import SimpleNamespace + +import pytest + +from osism.tasks.conductor.sonic.bgp import ( + calculate_local_asn_from_ipv4, + calculate_minimum_as_for_group, + find_interconnected_spine_groups, +) +from osism.tasks.conductor.sonic.constants import DEFAULT_LOCAL_AS_PREFIX + +# --------------------------------------------------------------------------- +# calculate_local_asn_from_ipv4 +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "ip,prefix,expected", + [ + # docstring example + ("192.168.45.123", None, 4200045123), + # CIDR suffix is stripped before parsing + ("192.168.45.123/32", None, 4200045123), + # third/fourth octet are zero-padded into the lower 6 digits + ("10.20.5.7", None, 4200005007), + # boundary octets + ("0.0.0.0", None, 4200000000), + ("255.255.255.255", None, 4200255255), + # custom prefix replaces DEFAULT_LOCAL_AS_PREFIX + ("10.0.1.2", 4201, 4201001002), + # custom prefix combined with CIDR + ("10.0.1.2/24", 4201, 4201001002), + ], +) +def test_calculate_local_asn_from_ipv4_valid_inputs(ip, prefix, expected): + if prefix is None: + assert calculate_local_asn_from_ipv4(ip) == expected + else: + assert calculate_local_asn_from_ipv4(ip, prefix=prefix) == expected + + +def test_calculate_local_asn_from_ipv4_returns_int(): + result = calculate_local_asn_from_ipv4("192.168.45.123") + + assert isinstance(result, int) + + +def test_calculate_local_asn_from_ipv4_uses_default_prefix(): + assert calculate_local_asn_from_ipv4( + "192.168.45.123" + ) == calculate_local_asn_from_ipv4("192.168.45.123", prefix=DEFAULT_LOCAL_AS_PREFIX) + + +def test_calculate_local_asn_from_ipv4_only_third_and_fourth_octets_matter(): + # First two octets are not part of the AS number — only the lower two are. + assert calculate_local_asn_from_ipv4("1.2.45.123") == calculate_local_asn_from_ipv4( + "99.99.45.123" + ) + + +@pytest.mark.parametrize( + "invalid_input,expected_match", + [ + ("192.168.45", "Invalid IPv4 address format"), + ("not-an-ip", "Invalid IPv4 address format"), + ("192.168.45.999", "Invalid octet values"), + ("192.168.999.45", "Invalid octet values"), + ("", "Invalid IPv4 address format"), + ("192.168.45.123.1", "Invalid IPv4 address format"), + ("192.168.45.abc", "invalid literal for int"), + ], +) +def test_calculate_local_asn_from_ipv4_invalid_inputs(invalid_input, expected_match): + with pytest.raises(ValueError, match=expected_match): + calculate_local_asn_from_ipv4(invalid_input) + + +def test_calculate_local_asn_from_ipv4_error_message_includes_input(): + with pytest.raises(ValueError, match="192.168.45.999"): + calculate_local_asn_from_ipv4("192.168.45.999") + + +# --------------------------------------------------------------------------- +# calculate_minimum_as_for_group +# --------------------------------------------------------------------------- + + +def _device(name, primary_ip4): + return SimpleNamespace(name=name, primary_ip4=primary_ip4) + + +def test_calculate_minimum_as_for_group_three_devices(): + devices = [ + _device("sw1", "192.168.45.123/32"), + _device("sw2", "192.168.45.50/32"), + _device("sw3", "192.168.45.200/32"), + ] + + result = calculate_minimum_as_for_group(devices) + + assert result == 4200045050 + + +def test_calculate_minimum_as_for_group_skips_invalid_ip(mocker): + debug = mocker.patch("osism.tasks.conductor.sonic.bgp.logger.debug") + devices = [ + _device("sw1", "not-an-ip"), + _device("sw2", "192.168.45.50/32"), + _device("sw3", "192.168.45.200/32"), + ] + + result = calculate_minimum_as_for_group(devices) + + assert result == 4200045050 + debug.assert_called_once() + assert "sw1" in debug.call_args.args[0] + + +def test_calculate_minimum_as_for_group_all_invalid_returns_none(mocker): + debug = mocker.patch("osism.tasks.conductor.sonic.bgp.logger.debug") + devices = [ + _device("sw1", "not-an-ip"), + _device("sw2", "also-bad"), + ] + + assert calculate_minimum_as_for_group(devices) is None + assert debug.call_count == 2 + + +def test_calculate_minimum_as_for_group_empty_returns_none(): + assert calculate_minimum_as_for_group([]) is None + + +def test_calculate_minimum_as_for_group_skips_none_primary_ip4(mocker): + debug = mocker.patch("osism.tasks.conductor.sonic.bgp.logger.debug") + devices = [ + _device("sw1", None), + _device("sw2", "192.168.45.50/32"), + ] + + result = calculate_minimum_as_for_group(devices) + + assert result == 4200045050 + # None falls through the truthiness check, so logger.debug must NOT be called + # for the skipped device. + debug.assert_not_called() + + +def test_calculate_minimum_as_for_group_skips_empty_string_primary_ip4(mocker): + debug = mocker.patch("osism.tasks.conductor.sonic.bgp.logger.debug") + devices = [ + _device("sw1", ""), + _device("sw2", "192.168.45.50/32"), + ] + + result = calculate_minimum_as_for_group(devices) + + assert result == 4200045050 + debug.assert_not_called() + + +def test_calculate_minimum_as_for_group_all_none_returns_none(): + devices = [ + _device("sw1", None), + _device("sw2", None), + ] + + assert calculate_minimum_as_for_group(devices) is None + + +def test_calculate_minimum_as_for_group_single_device(): + devices = [_device("sw1", "192.168.45.123/32")] + + assert calculate_minimum_as_for_group(devices) == 4200045123 + + +def test_calculate_minimum_as_for_group_custom_prefix(): + devices = [ + _device("sw1", "10.0.1.2/32"), + _device("sw2", "10.0.5.10/32"), + ] + + assert calculate_minimum_as_for_group(devices, prefix=4201) == 4201001002 + + +def test_calculate_minimum_as_for_group_stringifies_primary_ip4(mocker): + # primary_ip4 may be a non-str object (e.g. NetBox IPAddress) — the function + # passes it through str() before parsing. + class _IP: + def __str__(self): + return "192.168.45.123/32" + + devices = [_device("sw1", _IP())] + + assert calculate_minimum_as_for_group(devices) == 4200045123 + + +# --------------------------------------------------------------------------- +# find_interconnected_spine_groups (deprecated wrapper) +# --------------------------------------------------------------------------- + + +def test_find_interconnected_spine_groups_delegates_with_default_roles(mocker): + mock_fn = mocker.patch( + "osism.tasks.conductor.sonic.connections.find_interconnected_devices", + return_value=[["device-a", "device-b"]], + ) + devices = ["device-a", "device-b", "device-c"] + + result = find_interconnected_spine_groups(devices) + + mock_fn.assert_called_once_with(devices, ["spine", "superspine"]) + assert result == [["device-a", "device-b"]] + + +def test_find_interconnected_spine_groups_passes_custom_target_roles(mocker): + mock_fn = mocker.patch( + "osism.tasks.conductor.sonic.connections.find_interconnected_devices", + return_value=[], + ) + devices = ["device-a"] + custom_roles = ["leaf", "accessleaf"] + + result = find_interconnected_spine_groups(devices, target_roles=custom_roles) + + mock_fn.assert_called_once_with(devices, custom_roles) + assert result == [] + + +def test_find_interconnected_spine_groups_returns_value_unchanged(mocker): + sentinel = object() + mocker.patch( + "osism.tasks.conductor.sonic.connections.find_interconnected_devices", + return_value=sentinel, + ) + + assert find_interconnected_spine_groups([]) is sentinel diff --git a/tests/unit/tasks/conductor/sonic/test_constants.py b/tests/unit/tasks/conductor/sonic/test_constants.py new file mode 100644 index 000000000..5207d4acb --- /dev/null +++ b/tests/unit/tasks/conductor/sonic/test_constants.py @@ -0,0 +1,150 @@ +# SPDX-License-Identifier: Apache-2.0 + +import pytest + +from osism.tasks.conductor.sonic.constants import ( + BGP_AF_L2VPN_EVPN_TAG, + DEFAULT_LOCAL_AS_PREFIX, + DEFAULT_SONIC_ROLES, + DEFAULT_SONIC_VERSION, + HIGH_SPEED_PORTS, + PORT_CONFIG_PATH, + PORT_TYPE_TO_SPEED_MAP, + SUPPORTED_HWSKUS, +) + +# --------------------------------------------------------------------------- +# Simple constant values +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "constant,expected", + [ + (BGP_AF_L2VPN_EVPN_TAG, "bgp-af-l2vpn-evpn"), + (DEFAULT_LOCAL_AS_PREFIX, 4200), + (PORT_CONFIG_PATH, "/etc/sonic/port_config"), + ], +) +def test_simple_constant_values(constant, expected): + assert constant == expected + + +def test_default_local_as_prefix_is_int(): + assert isinstance(DEFAULT_LOCAL_AS_PREFIX, int) + + +# --------------------------------------------------------------------------- +# DEFAULT_SONIC_ROLES +# --------------------------------------------------------------------------- + + +def test_default_sonic_roles_contains_expected_roles(): + expected = {"spine", "superspine", "leaf", "accessleaf"} + + assert expected.issubset(set(DEFAULT_SONIC_ROLES)) + + +def test_default_sonic_roles_is_sorted(): + assert DEFAULT_SONIC_ROLES == sorted(DEFAULT_SONIC_ROLES) + + +def test_default_sonic_roles_has_no_duplicates(): + assert len(DEFAULT_SONIC_ROLES) == len(set(DEFAULT_SONIC_ROLES)) + + +def test_default_sonic_roles_entries_are_non_empty_strings(): + for role in DEFAULT_SONIC_ROLES: + assert isinstance(role, str) + assert role + + +# --------------------------------------------------------------------------- +# DEFAULT_SONIC_VERSION +# --------------------------------------------------------------------------- + + +def test_default_sonic_version_is_non_empty_string(): + assert isinstance(DEFAULT_SONIC_VERSION, str) + assert DEFAULT_SONIC_VERSION + + +# --------------------------------------------------------------------------- +# PORT_TYPE_TO_SPEED_MAP +# --------------------------------------------------------------------------- + + +def test_port_type_to_speed_map_values_are_non_negative_ints(): + for key, value in PORT_TYPE_TO_SPEED_MAP.items(): + assert isinstance(value, int), key + assert value >= 0, key + + +def test_port_type_to_speed_map_keys_are_non_empty_strings(): + for key in PORT_TYPE_TO_SPEED_MAP: + assert isinstance(key, str) + assert key + + +@pytest.mark.parametrize( + "port_type,expected_speed", + [ + ("virtual", 0), + ("10gbase-t", 10000), + ("100gbase-x-qsfp28", 100000), + ("400gbase-x-qsfpdd", 400000), + ], +) +def test_port_type_to_speed_map_specific_values(port_type, expected_speed): + assert PORT_TYPE_TO_SPEED_MAP[port_type] == expected_speed + + +# --------------------------------------------------------------------------- +# HIGH_SPEED_PORTS +# --------------------------------------------------------------------------- + + +def test_high_speed_ports_value(): + assert HIGH_SPEED_PORTS == {100000, 200000, 400000, 800000} + + +def test_high_speed_ports_is_set(): + assert isinstance(HIGH_SPEED_PORTS, set) + + +def test_high_speed_ports_values_positive(): + for value in HIGH_SPEED_PORTS: + assert isinstance(value, int) + assert value > 0 + + +def test_high_speed_ports_consistent_with_port_type_map(): + # All high-speed ports must be representable by at least one port type, + # except 800000 which is reserved for future 800G ports and has no entry + # in PORT_TYPE_TO_SPEED_MAP yet. Asserting on the exact gap (rather than + # skipping 800000) makes the test fail once an 800000 port type is added, + # forcing the maintainer to remove this carve-out. + speeds_in_map = set(PORT_TYPE_TO_SPEED_MAP.values()) + unrepresented = HIGH_SPEED_PORTS - speeds_in_map + + assert unrepresented == {800000} + + +# --------------------------------------------------------------------------- +# SUPPORTED_HWSKUS +# --------------------------------------------------------------------------- + + +def test_supported_hwskus_is_non_empty(): + assert SUPPORTED_HWSKUS + + +def test_supported_hwskus_no_duplicates(): + assert len(SUPPORTED_HWSKUS) == len(set(SUPPORTED_HWSKUS)) + + +@pytest.mark.parametrize("hwsku", SUPPORTED_HWSKUS) +def test_supported_hwskus_entry_invariants(hwsku): + assert isinstance(hwsku, str) + assert hwsku + assert hwsku.startswith("Accton-")