From ff3be4a24d4fdb64cd4ec0836e9917c75f964dae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=9F=D0=B0=D0=B2=D0=B5=D0=BB=20=D0=96=D1=83=D0=BA=D0=BE?= =?UTF-8?q?=D0=B2?= <33721692+LeaveMyYard@users.noreply.github.com> Date: Fri, 26 May 2023 11:55:15 +0300 Subject: Fix tests to work without kubeconfig --- tests/conftest.py | 80 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 80 insertions(+) create mode 100644 tests/conftest.py (limited to 'tests/conftest.py') diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..ab9ca01 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,80 @@ +from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch +import pytest +import numpy as np +from datetime import datetime, timedelta +import random +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_config_loaded(): + with patch("robusta_krr.core.models.config.Config.config_loaded", new_callable=PropertyMock) as mock_config: + mock_config.return_value = True + 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.PrometheusLoader.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.PrometheusLoader.__init__", return_value=None): + yield -- cgit v1.2.3