fix(loadmanager): fix dynamic stop and proper destroy for the class and fix test

Author: SQuent

app/managers/load_manager.py

@@ -11,18 +11,20 @@ def __init__(self):
             os.getcwd(), "app/scripts/memory_stress.py"
         )
 
-        self.cpu_requested = int(os.getenv("CPU_REQUESTED", 0))
+        self.cpu_requested = float(os.getenv("CPU_REQUESTED", 0))
         self.memory_requested = int(os.getenv("MEMORY_REQUESTED", 0))
         self.cpu_processes = []
+        self.cpu_timers = []
+        self.memory_timers = []
         self.memory_process = None
 
         self.memory_at_start = int(os.getenv("INITIAL_MEMORY_LOAD", 50))
         self.memory_at_end = int(os.getenv("FINAL_MEMORY_LOAD", 256))
         self.memory_duration = int(os.getenv("MEMORY_LOAD_DURATION", 60))
         self.stop_memory_at_end = os.getenv("STOP_MEMORY_LOAD_AT_END", "true") == "true"
 
-        self.cpu_at_start = int(os.getenv("INITIAL_CPU_LOAD", 0))
-        self.cpu_at_end = int(os.getenv("FINAL_CPU_LOAD", 1))
+        self.cpu_at_start = float(os.getenv("INITIAL_CPU_LOAD", 0))
+        self.cpu_at_end = float(os.getenv("FINAL_CPU_LOAD", 1))
         self.cpu_duration = int(os.getenv("CPU_LOAD_DURATION", 60))
         self.stop_cpu_at_end = os.getenv("STOP_CPU_LOAD_AT_END", "true") == "true"
 
@@ -43,6 +45,10 @@ def __init__(self):
                 self.stop_cpu_at_end,
             )
 
+    def __del__(self):
+        self.stop_cpu_load()
+        self.stop_memory_load()
+
     def stop_cpu_load(self):
         for process in self.cpu_processes:
             try:
@@ -51,6 +57,9 @@ def stop_cpu_load(self):
             except Exception as e:
                 raise RuntimeError(f"Failed to terminate process {process.pid}: {e}")
         self.cpu_processes.clear()
+        for timer in self.cpu_timers:
+            timer.cancel()
+        self.cpu_timers.clear()
         self.cpu_requested = 0
 
     def add_cpu_load(self, value: int):
@@ -75,6 +84,10 @@ def stop_memory_load(self):
             try:
                 self.memory_process.terminate()
                 self.memory_process = None
+
+                for timer in self.memory_timers:
+                    timer.cancel()
+                self.memory_timers.clear()
                 self.memory_requested = 0
             except Exception as e:
                 raise RuntimeError(f"Failed to terminate memory stress: {e}")
@@ -110,7 +123,6 @@ def dynamic_memory_load(
             raise ValueError(
                 "Arguments 'start_value', 'end_value', 'duration' have to be integers."
             ) from e
-
         num_intervals = duration // 10
 
         if num_intervals < 1:
@@ -127,10 +139,12 @@ def apply_dynamic_memory_load(interval_num):
             self.add_memory_load(int(current_memory))
 
             if interval_num + 1 < num_intervals:
-                Timer(10, apply_dynamic_memory_load, [interval_num + 1]).start()
-            else:
-                if stop_at_end:
-                    self.stop_memory_load
+                timer = Timer(10, apply_dynamic_memory_load, [interval_num + 1])
+                self.memory_timers.append(timer)
+                timer.start()
+            elif stop_at_end:
+                stop_timer = Timer(10, self.stop_memory_load)
+                stop_timer.start()
 
         apply_dynamic_memory_load(0)
 
@@ -164,9 +178,13 @@ def apply_dynamic_cpu_load(interval_num):
             self.add_cpu_load(current_cpu)
 
             if interval_num + 1 < num_intervals:
-                Timer(10, apply_dynamic_cpu_load, [interval_num + 1]).start()
-            else:
-                if stop_at_end:
-                    self.stop_cpu_load
+                timer = Timer(10, apply_dynamic_cpu_load, [interval_num + 1])
+                self.cpu_timers.append(timer)
+                timer.start()
+
+            elif stop_at_end:
+                # Correctly schedule stop_memory_load to execute after the last interval
+                stop_timer = Timer(10, self.stop_cpu_load)
+                stop_timer.start()
 
         apply_dynamic_cpu_load(0)

tests/conftest.py

@@ -29,7 +29,6 @@ def kill_stress_processes():
     """Kill all stress test processes"""
     for proc in psutil.process_iter(["pid", "name", "cmdline"]):
         try:
-            # Chercher les processus python qui exécutent cpu_stress.py ou memory_stress.py
             if proc.info["cmdline"] and "python" in proc.info["cmdline"][0]:
                 cmdline = " ".join(proc.info["cmdline"])
                 if "cpu_stress.py" in cmdline or "memory_stress.py" in cmdline:

tests/integration/test_load_endpoints.py

@@ -13,11 +13,7 @@ def client(self):
     def test_cpu_load_endpoints(self, client):
         """Test CPU load endpoints"""
         # Test start CPU load
-        response = client.post("/load/cpu/start", json={"value": 30})
-        assert response.status_code == 200
-
-        # Test stop CPU load
-        response = client.post("/load/cpu/stop")
+        response = client.post("/load/cpu/start", json={"value": 1})
         assert response.status_code == 200
 
     def test_memory_load_endpoints(self, client):
@@ -26,19 +22,15 @@ def test_memory_load_endpoints(self, client):
         response = client.post("/load/memory/start", json={"value": 50})
         assert response.status_code == 200
 
-        # Test stop memory load
-        response = client.post("/load/memory/stop")
-        assert response.status_code == 200
-
     def test_dynamic_load_endpoints(self, client):
         """Test dynamic load endpoints"""
         # Test dynamic CPU load
         cpu_response = client.post(
             "/load/cpu/dynamic",
             json={
-                "start_value": 10,
-                "end_value": 20,
-                "duration": 60,
+                "start_value": 0,
+                "end_value": 0.5,
+                "duration": 30,
                 "stop_at_end": True,
             },
         )
@@ -51,7 +43,7 @@ def test_dynamic_load_endpoints(self, client):
             json={
                 "start_value": 100,
                 "end_value": 200,
-                "duration": 60,
+                "duration": 20,
                 "stop_at_end": True,
             },
         )
@@ -67,6 +59,12 @@ def test_dynamic_load_endpoints(self, client):
         assert "cpu_active" in load_status
         assert "memory_active" in load_status
 
+        response = client.post("/load/memory/stop")
+        assert response.status_code == 200
+
+        response = client.post("/load/memory/stop")
+        assert response.status_code == 200
+
     def test_get_current_load(self, client):
         """Test get current load endpoint"""
         response = client.get("/load")

tests/test_api.py

@@ -51,7 +51,12 @@ def test_load_endpoints():
     # Test CPU operations
     response = client.post(
         "/load/cpu/dynamic",
-        json={"start_value": 10, "end_value": 20, "duration": 60, "stop_at_end": True},
+        json={
+            "start_value": 0.1,
+            "end_value": 0.3,
+            "duration": 20,
+            "stop_at_end": True,
+        },
     )
     assert response.status_code == 200
 
@@ -62,6 +67,17 @@ def test_load_endpoints():
     assert response.status_code == 200
 
     # Test Memory operations
+    response = client.post(
+        "/load/cpu/dynamic",
+        json={
+            "start_value": 100,
+            "end_value": 256,
+            "duration": 20,
+            "stop_at_end": True,
+        },
+    )
+    assert response.status_code == 200
+
     response = client.post("/load/memory/start", json={"value": 100})
     assert response.status_code == 200
 

tests/unit/test_load_manager.py

@@ -41,13 +41,13 @@ def test_initialization_with_defaults(self, load_manager):
 
     def test_initialization_with_env_vars(self, monkeypatch):
         """Test initialization with specific environment variables"""
-        monkeypatch.setenv("CPU_REQUESTED", "20")
+        monkeypatch.setenv("CPU_REQUESTED", "0.5")
         monkeypatch.setenv("MEMORY_REQUESTED", "40")
         monkeypatch.setenv("ENABLE_DYNAMIC_CPU_LOAD", "false")
         monkeypatch.setenv("ENABLE_DYNAMIC_MEMORY_LOAD", "false")
 
         manager = LoadManager()
-        assert manager.cpu_requested == 20
+        assert manager.cpu_requested == 0.5
         assert manager.memory_requested == 40
 
     # 2. Tests des méthodes CPU
@@ -65,7 +65,7 @@ def test_add_cpu_load_with_invalid_values(self, load_manager):
 
     def test_stop_cpu_load(self, load_manager):
         """Test stopping CPU load"""
-        load_manager.add_cpu_load(30)
+        load_manager.add_cpu_load(1)
         load_manager.stop_cpu_load()
         assert load_manager.cpu_requested == 0
         assert not load_manager.cpu_processes
@@ -78,20 +78,19 @@ def test_stop_cpu_load_no_processes(self, load_manager):
 
     def test_dynamic_cpu_load(self, load_manager):
         """Test dynamic CPU load"""
-        load_manager.dynamic_cpu_load(10, 80, 1, True)
-        assert load_manager.cpu_requested >= 10
-        assert load_manager.cpu_requested <= 80
+        load_manager.dynamic_cpu_load(0.1, 1, 1, True)
+        assert load_manager.cpu_requested >= 0.1
+        assert load_manager.cpu_requested <= 1
 
-    @pytest.mark.asyncio
-    async def test_dynamic_cpu_load_invalid_duration(self, load_manager):
+    def test_dynamic_cpu_load_invalid_duration(self, load_manager):
         """Test invalid duration for dynamic CPU load"""
         with pytest.raises(ValueError):
-            load_manager.dynamic_cpu_load(-1, 80, -1, True)
+            load_manager.dynamic_cpu_load(-1, 1, -1, True)
 
     def test_dynamic_cpu_load_zero_duration(self, load_manager):
         """Test dynamic CPU load with zero duration"""
-        load_manager.dynamic_cpu_load(10, 90, 0, True)
-        assert load_manager.cpu_requested == 90
+        load_manager.dynamic_cpu_load(0, 1, 0, False)
+        assert load_manager.cpu_requested == 1
 
     # 3. Tests des méthodes mémoire
     def test_add_memory_load(self, load_manager):
@@ -121,26 +120,25 @@ def test_stop_memory_load_no_processes(self, load_manager):
 
     def test_dynamic_memory_load(self, load_manager):
         """Test dynamic memory load"""
-        load_manager.dynamic_memory_load(100, 200, 1, True)
+        load_manager.dynamic_memory_load(100, 200, 1, False)
         assert load_manager.memory_requested >= 100
         assert load_manager.memory_requested <= 200
 
-    @pytest.mark.asyncio
-    async def test_dynamic_memory_load_invalid_duration(self, load_manager):
+    def test_dynamic_memory_load_invalid_duration(self, load_manager):
         """Test invalid duration for dynamic memory load"""
         with pytest.raises(ValueError):
             load_manager.dynamic_memory_load(-1, 200, -1, True)
 
     def test_dynamic_memory_load_zero_duration(self, load_manager):
         """Test dynamic memory load with zero duration"""
-        load_manager.dynamic_memory_load(100, 200, 0, True)
-        assert load_manager.memory_requested == 200
+        load_manager.dynamic_memory_load(100, 256, 0, False)
+        assert load_manager.memory_requested == 256
 
     # 4. Tests combinés et autres
     def test_stop_all_loads(self, load_manager):
         """Test stopping all loads"""
-        load_manager.add_cpu_load(30)
-        load_manager.add_memory_load(50)
+        load_manager.add_cpu_load(1)
+        load_manager.add_memory_load(100)
         load_manager.stop_cpu_load()
         load_manager.stop_memory_load()
         assert load_manager.cpu_requested == 0