get_log_file mock attempt

simulation-system/libs/csle-cluster/src/csle_cluster/cluster_manager/cluster_manager.py

@@ -385,7 +385,7 @@ def getLogFile(self, request: csle_cluster.cluster_manager.cluster_manager_pb2.G
         :param context: the gRPC context
         :return: a DTO with logs
         """
-        logging.info("Getting log file: {request.name}")
+        logging.info(f"Getting log file: {request.name}")
         data = []
         if os.path.exists(request.name):
             try:

simulation-system/libs/csle-cluster/src/csle_cluster/cluster_manager/query_cluster_manager.py

@@ -157,7 +157,7 @@ def start_flask(
     return service_status_dto
 
 
-def start_docker_statsmanager(
+def start_docker_stats_manager(
         stub: csle_cluster.cluster_manager.cluster_manager_pb2_grpc.ClusterManagerStub,
         timeout=constants.GRPC.TIMEOUT_SECONDS) \
         -> csle_cluster.cluster_manager.cluster_manager_pb2.ServiceStatusDTO:
@@ -327,7 +327,7 @@ def stop_flask(
     return service_status_dto
 
 
-def stop_docker_statsmanager(
+def stop_docker_stats_manager(
         stub: csle_cluster.cluster_manager.cluster_manager_pb2_grpc.ClusterManagerStub,
         timeout=constants.GRPC.TIMEOUT_SECONDS) \
         -> csle_cluster.cluster_manager.cluster_manager_pb2.ServiceStatusDTO:

simulation-system/libs/csle-cluster/tests/test_cluster_manager.py

@@ -3,6 +3,7 @@
 import pytest_mock
 from csle_cluster.cluster_manager.cluster_manager import ClusterManagerServicer
 from csle_cluster.cluster_manager.cluster_manager_pb2 import ServiceStatusDTO
+from csle_cluster.cluster_manager.cluster_manager_pb2 import LogsDTO
 from csle_common.dao.emulation_config.config import Config
 import csle_cluster.cluster_manager.query_cluster_manager
 from csle_cluster.cluster_manager.cluster_manager_pb2 import NodeStatusDTO
@@ -43,6 +44,17 @@ def grpc_stub_cls(self, grpc_channel):
         from csle_cluster.cluster_manager.cluster_manager_pb2_grpc import ClusterManagerStub
         return ClusterManagerStub
 
+    @staticmethod
+    def with_class():
+        class A:
+            def __init__(self):
+                pass
+            def __enter__(self):
+                pass
+            def __exit__(self, exc_type, exc_value, traceback):
+                pass
+        return A()
+
     def test_getNodeStatus(self, grpc_stub, mocker: pytest_mock.MockFixture, example_config: Config) -> None:
         """
         Tests the getNodeStatus grpc
@@ -366,3 +378,175 @@ def test_stopGrafana(self, grpc_stub, mocker: pytest_mock.MockFixture) -> None:
         response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.stop_grafana(
             stub=grpc_stub)
         assert not response.running
+
+    def test_startPrometheus(self, grpc_stub, mocker: pytest_mock.MockFixture) -> None:
+        """
+        Tests the startPrometheus grpc
+        
+        :param grpc_stub: the stub for the GRPC server to make the request to
+        :param mocker: the mocker object to mock functions with external dependencies
+        :return: None
+        """
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.start_prometheus', return_value=False)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.start_prometheus(
+            stub=grpc_stub)
+        assert response.running
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.start_prometheus', return_value=True)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.start_prometheus(
+            stub=grpc_stub)
+        assert response.running
+
+    def test_stopPrometheus(self, grpc_stub, mocker: pytest_mock.MockFixture) -> None:
+        """
+        Tests the stopPrometheus grpc
+        
+        :param grpc_stub: the stub for the GRPC server to make the request to
+        :param mocker: the mocker object to mock functions with external dependencies
+        :return: None
+        """
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.stop_prometheus', return_value=False)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.stop_prometheus(
+            stub=grpc_stub)
+        assert not response.running
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.stop_prometheus', return_value=True)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.stop_prometheus(
+            stub=grpc_stub)
+        assert not response.running
+
+    def test_startPgAdmin(self, grpc_stub, mocker: pytest_mock.MockFixture) -> None:
+        """
+        Tests the startPgAdmin grpc
+        
+        :param grpc_stub: the stub for the GRPC server to make the request to
+        :param mocker: the mocker object to mock functions with external dependencies
+        :return: None
+        """
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.start_pgadmin', return_value=False)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.start_pgadmin(
+            stub=grpc_stub)
+        assert response.running
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.start_pgadmin', return_value=True)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.start_pgadmin(
+            stub=grpc_stub)
+        assert response.running
+
+    def test_stopPgAdmin(self, grpc_stub, mocker: pytest_mock.MockFixture) -> None:
+        """
+        Tests the stopPgAdmin grpc
+        
+        :param grpc_stub: the stub for the GRPC server to make the request to
+        :param mocker: the mocker object to mock functions with external dependencies
+        :return: None
+        """
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.stop_pgadmin', return_value=False)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.stop_pgadmin(
+            stub=grpc_stub)
+        assert not response.running
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.stop_pgadmin', return_value=True)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.stop_pgadmin(
+            stub=grpc_stub)
+        assert not response.running
+
+    def test_startFlask(self, grpc_stub, mocker: pytest_mock.MockFixture) -> None:
+        """
+        Tests the startFlask grpc
+        
+        :param grpc_stub: the stub for the GRPC server to make the request to
+        :param mocker: the mocker object to mock functions with external dependencies
+        :return: None
+        """
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.start_flask', return_value=False)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.start_flask(
+            stub=grpc_stub)
+        assert response.running
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.start_flask', return_value=True)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.start_flask(
+            stub=grpc_stub)
+        assert response.running
+
+    def test_stopFlask(self, grpc_stub, mocker: pytest_mock.MockFixture) -> None:
+        """
+        Tests the startFlask grpc
+        
+        :param grpc_stub: the stub for the GRPC server to make the request to
+        :param mocker: the mocker object to mock functions with external dependencies
+        :return: None
+        """
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.stop_flask', return_value=False)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.stop_flask(
+            stub=grpc_stub)
+        assert not response.running
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.stop_flask', return_value=True)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.stop_flask(
+            stub=grpc_stub)
+        assert not response.running
+
+    def test_startDockerStatsManager(self, grpc_stub, mocker: pytest_mock.MockFixture) -> None:
+        """
+        Tests the startDockerStatsManager grpc
+        
+        :param grpc_stub: the stub for the GRPC server to make the request to
+        :param mocker: the mocker object to mock functions with external dependencies
+        :return: None
+        """
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.start_docker_stats_manager', return_value=False)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.start_docker_stats_manager(
+            stub=grpc_stub)
+        assert response.running
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.start_docker_stats_manager', return_value=True)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.start_docker_stats_manager(
+            stub=grpc_stub)
+        assert response.running
+
+    def test_stopDockerStatsManager(self, grpc_stub, mocker: pytest_mock.MockFixture) -> None:
+        """
+        Tests the stopDockerStatsManager grpc
+        
+        :param grpc_stub: the stub for the GRPC server to make the request to
+        :param mocker: the mocker object to mock functions with external dependencies
+        :return: None
+        """
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.stop_docker_stats_manager', return_value=False)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.stop_docker_stats_manager(
+            stub=grpc_stub)
+        assert not response.running
+        mocker.patch('csle_common.controllers.management_system_controller.'
+                     'ManagementSystemController.stop_docker_stats_manager', return_value=True)
+        response: ServiceStatusDTO = csle_cluster.cluster_manager.query_cluster_manager.stop_docker_stats_manager(
+            stub=grpc_stub)
+        assert not response.running
+
+    def test_getLogFile(self, grpc_stub, mocker: pytest_mock.MockFixture) -> None:
+        """
+        Tests the getLogFile grpc
+        
+        :param grpc_stub: the stub for the GRPC server to make the request to
+        :param mocker: the mocker object to mock functions with external dependencies
+        :return: None
+        """
+        mocker.patch('csle_cluster.cluster_manager.cluster_manager_util.ClusterManagerUtil.tail',
+                     return_value="abcdef")
+        mocker.patch("os.path.exists", return_value=True)
+        mocker.patch('builtins.open', return_value=TestClusterManagerSuite.with_class())
+        response: LogsDTO = csle_cluster.cluster_manager.query_cluster_manager.get_log_file(stub=grpc_stub,
+                                                                                            log_file_name="abcdef")
+        assert response.logs == ['abcdef']
+        mocker.patch('builtins.open', return_value=None)
+        response: LogsDTO = csle_cluster.cluster_manager.query_cluster_manager.get_log_file(stub=grpc_stub,
+                                                                                            log_file_name="abcdef")
+        assert response.logs == []