Merge branch 'main' into fix-autoliniting--precommit

Author: SQuent

app/main.py

@@ -3,7 +3,7 @@
 
 app = FastAPI(
     title="Stressed API",
-    description="The API simulates controlled workloads and failures to stress-test a system, assessing its resilience, performance, and recovery capabilities.",
+    description="simulates controlled workloads and failures to stress-test a system",
     version="1.0.0",
 )
 

app/managers/load_manager.py

@@ -71,7 +71,9 @@ def add_cpu_load(self, value: float):
         if value <= 0:
             raise ValueError("CPU load must be greater than 0")
         if value > self.system_cpus:
-            raise ValueError(f"CPU load cannot exceed system CPU count ({self.system_cpus})")
+            raise ValueError(
+                f"CPU load cannot exceed system CPU count ({self.system_cpus})"
+            )
 
         if not os.path.exists(self.cpu_script_path):
             raise RuntimeError("CPU stress script is missing")
@@ -103,7 +105,9 @@ def add_memory_load(self, value: int):
         if value <= 0:
             raise ValueError("Memory load must be greater than 0")
         if value > self.system_memory:
-            raise ValueError(f"Memory load cannot exceed system memory ({self.system_memory}MB)")
+            raise ValueError(
+                f"Memory load cannot exceed system memory ({self.system_memory}MB)"
+            )
 
         if not os.path.exists(self.memory_script_path):
             raise RuntimeError("Memory stress script is missing")
@@ -121,7 +125,9 @@ def dynamic_memory_load(
     ):
         """Dynamic memory load with validation"""
         if duration <= 0 or duration > self.max_duration:
-            raise ValueError(f"Duration must be between 1 and {self.max_duration} seconds")
+            raise ValueError(
+                f"Duration must be between 1 and {self.max_duration} seconds"
+            )
         if end_value < start_value:
             raise ValueError("End value must be greater than start value")
 
@@ -159,11 +165,17 @@ def apply_dynamic_memory_load(interval_num):
         apply_dynamic_memory_load(0)
 
     def dynamic_cpu_load(
-        self, start_value: float, end_value: float, duration: int, stop_at_end: bool = False
+        self,
+        start_value: float,
+        end_value: float,
+        duration: int,
+        stop_at_end: bool = False,
     ):
         """Dynamic CPU load with validation"""
         if duration <= 0 or duration > self.max_duration:
-            raise ValueError(f"Duration must be between 1 and {self.max_duration} seconds")
+            raise ValueError(
+                f"Duration must be between 1 and {self.max_duration} seconds"
+            )
         if end_value < start_value:
             raise ValueError("End value must be greater than start value")
 

app/managers/log_manager.py

@@ -85,19 +85,20 @@ def _start_automatic_logs(self):
             service=os.getenv("LOG_SERVICE", "auto-logger"),
             format=LogFormat(os.getenv("LOG_FORMAT", "json").lower()),
             interval=int(os.getenv("LOG_INTERVAL", "5")),
-            duration=int(os.getenv("LOG_DURATION", "60"))
+            duration=int(os.getenv("LOG_DURATION", "60")),
         )
-        
+
         # Create single log immediately
-        asyncio.create_task(self._create_single_log(
-            log_data, 
-            getattr(logging, log_data.level.upper())
-        ))
+        asyncio.create_task(
+            self._create_single_log(log_data, getattr(logging, log_data.level.upper()))
+        )
 
         # Setup recurring logs if needed
         if log_data.interval and log_data.duration:
             asyncio.create_task(
-                self._create_recurring_logs(log_data, log_data.interval, log_data.duration)
+                self._create_recurring_logs(
+                    log_data, log_data.interval, log_data.duration
+                )
             )
 
     async def create_log(self, log_data: LogRequest) -> dict | str:
@@ -169,4 +170,4 @@ async def log_creator(self, log_data: LogRequest, interval: int, duration: int):
         end_time = time.time() + duration
         while time.time() < end_time:
             await self.create_log(log_data)
-            await asyncio.sleep(interval)
+            await asyncio.sleep(interval)

app/managers/system_manager.py

@@ -1,15 +1,14 @@
 import psutil
 from datetime import datetime, timedelta
 import os
-import asyncio
 
 
 class SystemManager:
     def __init__(self):
         if os.getenv("ENABLE_AUTO_TERMINATION", "false").lower() == "true":
             delay = int(os.getenv("AUTO_TERMINATION_DELAY", "300"))
             self.terminate(delay)
-    
+
     def get_system_info(self):
         """Get system information"""
         return {

app/models/schemas.py

@@ -22,7 +22,9 @@ class DynamicMemoryLoadRequest(BaseModel):
 
 
 class LoadRequest(BaseModel):
-    value: float = Field(..., ge=0, description="Load value (CPU cores or MB of memory)")
+    value: float = Field(
+        ..., ge=0, description="Load value (CPU cores or MB of memory)"
+    )
 
 
 class ProbeRequest(BaseModel):

tests/unit/test_load_manager.py

@@ -162,6 +162,7 @@ def test_stop_all_loads(self, load_manager):
 
     def test_add_cpu_load_script_missing(self, load_manager, monkeypatch):
         """Test failure when cpu_stress.py is missing"""
+
         def mock_exists(path):
             return False
 
@@ -233,4 +234,6 @@ async def test_edge_cases_dynamic_load(self, load_manager):
 
         # Test with maximum duration
         with pytest.raises(ValueError, match="Duration must be between 1 and"):
-            load_manager.dynamic_memory_load(100, 200, 3601, True)  # Duration exceeds max_duration
+            load_manager.dynamic_memory_load(
+                100, 200, 3601, True
+            )  # Duration exceeds max_duration

tests/unit/test_log_manager.py

@@ -3,7 +3,6 @@
 from datetime import datetime, UTC
 from app.managers.log_manager import LogManager, JsonFormatter, PlainTextFormatter
 from app.models.schemas import LogRequest, LogFormat
-import os
 import asyncio
 
 
@@ -202,14 +201,18 @@ async def test_automatic_logs_initialization(self, monkeypatch, caplog):
             manager = LogManager()
             # Wait a bit for the log to be created
             await asyncio.sleep(0.1)
-            
+
         # Verify logs were created with correct configuration
-        assert any("Test auto message" in record.message for record in caplog.records), \
-            "Expected log message not found in records"
-        assert all(record.levelname == "INFO" for record in caplog.records), \
-            "Not all logs are at INFO level"
-        assert all(hasattr(record, "service") and record.service == "test-service" 
-                  for record in caplog.records), \
-            "Service attribute missing or incorrect"
-        assert isinstance(manager.logger.handlers[0].formatter, JsonFormatter), \
-            "Incorrect formatter type"
+        assert any(
+            "Test auto message" in record.message for record in caplog.records
+        ), "Expected log message not found in records"
+        assert all(
+            record.levelname == "INFO" for record in caplog.records
+        ), "Not all logs are at INFO level"
+        assert all(
+            hasattr(record, "service") and record.service == "test-service"
+            for record in caplog.records
+        ), "Service attribute missing or incorrect"
+        assert isinstance(
+            manager.logger.handlers[0].formatter, JsonFormatter
+        ), "Incorrect formatter type"