Nettacker/tests/core/utils/test_common.py at a3da8a568d88a0ed1ffe72b85e1e10284b4a70d8 · OWASP/Nettacker · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
from unittest.mock import patch

from nettacker.core.utils import common as common_utils


def test_arrays_to_matrix():
    assert sorted(common_utils.arrays_to_matrix({"ports": [1, 2, 3, 4, 5]})) == [
        [1],
        [2],
        [3],
        [4],
        [5],
    ]

    assert sorted(common_utils.arrays_to_matrix({"x": [1, 2], "y": [3, 4], "z": [5, 6]})) == [
        [1, 3, 5],
        [1, 3, 6],
        [1, 4, 5],
        [1, 4, 6],
        [2, 3, 5],
        [2, 3, 6],
        [2, 4, 5],
        [2, 4, 6],
    ]


def test_generate_target_groups_empty_list():
    targets = []
    set_hardware_usage = 3
    result = common_utils.generate_target_groups(targets, set_hardware_usage)
    assert result == []


def test_generate_target_groups_set_hardware_less_than_targets_total():
    targets = [1, 2, 3, 4, 5]
    set_hardware_usage = 2
    result = common_utils.generate_target_groups(targets, set_hardware_usage)
    assert result == [[1, 2, 3], [4, 5]]


def test_generate_target_groups_set_hardware_equal_to_targets_total():
    targets = [1, 2, 3, 4, 5]
    set_hardware_usage = 5
    result = common_utils.generate_target_groups(targets, set_hardware_usage)
    assert result == [[1], [2], [3], [4], [5]]


def test_generate_target_groups_set_hardware_greater_than_targets_total():
    targets = [1, 2, 3]
    set_hardware_usage = 5
    result = common_utils.generate_target_groups(targets, set_hardware_usage)
    assert result == [[1], [2], [3]]


def test_sort_dictionary():
    input_dict = {
        "a": 1,
        "c": 3,
        "d": 23,
        "b": 2,
    }
    expected_dict = {
        "a": 1,
        "b": 2,
        "c": 3,
        "d": 23,
    }
    input_dict_keys = tuple(input_dict.keys())
    expected_dict_keys = tuple(expected_dict.keys())
    assert input_dict_keys != expected_dict_keys
    sorted_dict_keys = tuple(common_utils.sort_dictionary(input_dict).keys())
    assert sorted_dict_keys == expected_dict_keys


@patch("multiprocessing.cpu_count")
def test_select_maximum_cpu_core(cpu_count_mock):
    cores_mapping = {
        1: {"low": 1, "normal": 1, "high": 1, "maximum": 1},
        2: {"low": 1, "normal": 1, "high": 1, "maximum": 1},
        4: {"low": 1, "normal": 1, "high": 2, "maximum": 3},
        6: {"low": 1, "normal": 1, "high": 3, "maximum": 5},
        8: {"low": 1, "normal": 2, "high": 4, "maximum": 7},
        10: {"low": 1, "normal": 2, "high": 5, "maximum": 9},
        12: {"low": 1, "normal": 3, "high": 6, "maximum": 11},
        16: {"low": 2, "normal": 4, "high": 8, "maximum": 15},
        32: {"low": 4, "normal": 8, "high": 16, "maximum": 31},
        48: {"low": 6, "normal": 12, "high": 24, "maximum": 47},
        64: {"low": 8, "normal": 16, "high": 32, "maximum": 63},
    }
    for num_cores, levels in cores_mapping.items():
        cpu_count_mock.return_value = num_cores
        for level in ("low", "normal", "high", "maximum"):
            assert common_utils.select_maximum_cpu_core(level) == levels[level]
        assert common_utils.select_maximum_cpu_core("invalid") == 1


def test_merge_logs_to_list_simple():
    result = {"log": "error occurred"}
    assert common_utils.merge_logs_to_list(result) == ["error occurred"]


def test_merge_logs_to_list_nested():
    result = {
        "log": "outer",
        "nested": {"log": "inner"},
    }
    logs = common_utils.merge_logs_to_list(result)
    assert sorted(logs) == ["inner", "outer"]


def test_merge_logs_to_list_no_log_key():
    result = {"status": "ok", "data": {"value": 42}}
    assert common_utils.merge_logs_to_list(result) == []


def test_merge_logs_to_list_deduplicates():
    result = {
        "log": "same",
        "nested": {"log": "same"},
    }
    assert common_utils.merge_logs_to_list(result) == ["same"]


def test_merge_logs_to_list_no_shared_state_between_calls():
    """Verify that consecutive calls without explicit log_list don't leak state."""
    result_a = {"log": "first"}
    result_b = {"log": "second"}
    logs_a = common_utils.merge_logs_to_list(result_a)
    logs_b = common_utils.merge_logs_to_list(result_b)
    assert logs_a == ["first"]
    assert logs_b == ["second"]