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
|
import random
from datetime import datetime, timedelta
from unittest.mock import AsyncMock, patch
import numpy as np
import pytest
from robusta_krr.api.models import K8sObjectData, PodData, ResourceAllocations, ResourceHistoryData
TEST_OBJECT = K8sObjectData(
cluster="mock-cluster",
name="mock-object-1",
container="mock-container-1",
pods=[
PodData(name="mock-pod-1", deleted=False),
PodData(name="mock-pod-2", deleted=False),
PodData(name="mock-pod-3", deleted=True),
],
namespace="default",
kind="Deployment",
allocations=ResourceAllocations(
requests={"cpu": 1, "memory": 1}, # type: ignore
limits={"cpu": 2, "memory": 2}, # type: ignore
),
)
@pytest.fixture(autouse=True, scope="session")
def mock_list_clusters():
with patch(
"robusta_krr.core.integrations.kubernetes.KubernetesLoader.list_clusters",
new=AsyncMock(return_value=[TEST_OBJECT.cluster]),
):
yield
@pytest.fixture(autouse=True, scope="session")
def mock_list_scannable_objects():
with patch(
"robusta_krr.core.integrations.kubernetes.KubernetesLoader.list_scannable_objects",
new=AsyncMock(return_value=[TEST_OBJECT]),
):
yield
@pytest.fixture(autouse=True, scope="session")
def mock_load_kubeconfig():
with patch("robusta_krr.core.models.config.Config.load_kubeconfig", return_value=None):
yield
@pytest.fixture(autouse=True, scope="session")
def mock_prometheus_loader():
now = datetime.now()
start = now - timedelta(hours=1)
now_ts, start_ts = now.timestamp(), start.timestamp()
metric_points_data = np.array([(t, random.randrange(0, 100)) for t in np.linspace(start_ts, now_ts, 3600)])
with patch(
"robusta_krr.core.integrations.prometheus.loader.MetricsLoader.gather_data",
new=AsyncMock(
return_value=ResourceHistoryData(
data={pod.name: metric_points_data for pod in TEST_OBJECT.pods},
metric={ # type: ignore
"query": f"example_promql_metric{{pod_name=~\"{'|'.join(pod.name for pod in TEST_OBJECT.pods)}\"}}",
"start_time": start,
"end_time": now,
"step": "30s",
},
)
),
) as mock_prometheus_loader:
mock_prometheus_loader
yield
@pytest.fixture(autouse=True, scope="session")
def mock_prometheus_init():
with patch("robusta_krr.core.integrations.prometheus.loader.MetricsLoader.__init__", return_value=None):
yield
|
