154 lines
5.7 KiB
Python
154 lines
5.7 KiB
Python
import unittest
|
|
import time
|
|
import psutil
|
|
import multiprocessing
|
|
import math
|
|
from typing import List, Tuple
|
|
|
|
from sia.system_metrics import SystemMetrics
|
|
|
|
def cpu_load_process():
|
|
"""Helper process that generates CPU load"""
|
|
while True:
|
|
math.factorial(100000)
|
|
|
|
class SystemMetricsTest(unittest.TestCase):
|
|
def setUp(self):
|
|
"""Create metrics instance with faster sampling for tests"""
|
|
# Updated constructor to match the implementation which takes no args
|
|
self.metrics = SystemMetrics()
|
|
|
|
def tearDown(self):
|
|
"""Ensure metrics monitoring is stopped"""
|
|
# If SystemMetrics has a stop method, call it here
|
|
if hasattr(self.metrics, 'stop'):
|
|
self.metrics.stop()
|
|
|
|
def validate_timestamp(self, timestamp: str):
|
|
"""Validate timestamp format and reasonableness"""
|
|
# Check format
|
|
self.assertRegex(
|
|
timestamp,
|
|
r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$'
|
|
)
|
|
|
|
# Check timestamp is recent (within last minute)
|
|
time_parts = timestamp[:-1].split('T') # Remove Z
|
|
time_str = f"{time_parts[0]} {time_parts[1]}"
|
|
timestamp_time = time.strptime(time_str, "%Y-%m-%d %H:%M:%S")
|
|
now = time.gmtime()
|
|
|
|
# Convert to seconds since epoch for comparison
|
|
timestamp_secs = time.mktime(timestamp_time)
|
|
now_secs = time.mktime(now)
|
|
|
|
self.assertLess(abs(now_secs - timestamp_secs), 60)
|
|
|
|
def validate_memory_metrics(self, used: int, total: int):
|
|
"""Validate memory metrics are reasonable"""
|
|
# Check types
|
|
self.assertIsInstance(used, int)
|
|
self.assertIsInstance(total, int)
|
|
|
|
# Used memory should be positive and less than total
|
|
self.assertGreater(used, 0)
|
|
self.assertGreater(total, 0)
|
|
self.assertLessEqual(used, total)
|
|
|
|
# Compare with actual system memory
|
|
memory = psutil.virtual_memory()
|
|
self.assertEqual(total, memory.total)
|
|
# Used memory should be within 20% of current usage
|
|
self.assertLess(abs(used - memory.used) / memory.total, 0.2)
|
|
|
|
def validate_disk_metrics(self, used: int, total: int):
|
|
"""Validate disk metrics are reasonable"""
|
|
# Check types
|
|
self.assertIsInstance(used, int)
|
|
self.assertIsInstance(total, int)
|
|
|
|
# Used space should be positive and less than total
|
|
self.assertGreater(used, 0)
|
|
self.assertGreater(total, 0)
|
|
self.assertLessEqual(used, total)
|
|
|
|
# Compare with actual disk usage
|
|
disk = psutil.disk_usage('/')
|
|
self.assertEqual(total, disk.total)
|
|
# Used space should match within 1% of actual usage
|
|
self.assertLess(abs(used - disk.used) / disk.total, 0.01)
|
|
|
|
def test_initial_metrics(self):
|
|
"""Test initial metrics generation"""
|
|
metrics = self.metrics.get_metrics()
|
|
|
|
# Validate timestamp
|
|
self.validate_timestamp(metrics["timestamp"])
|
|
|
|
# Validate memory metrics
|
|
self.validate_memory_metrics(
|
|
metrics["memory_used"],
|
|
metrics["memory_total"]
|
|
)
|
|
|
|
# Validate disk metrics
|
|
self.validate_disk_metrics(
|
|
metrics["disk_used"],
|
|
metrics["disk_total"]
|
|
)
|
|
|
|
def test_multiple_metric_readings(self):
|
|
"""Test multiple metric readings have different timestamps"""
|
|
metrics1 = self.metrics.get_metrics()
|
|
time.sleep(1)
|
|
metrics2 = self.metrics.get_metrics()
|
|
|
|
self.assertNotEqual(
|
|
metrics1["timestamp"],
|
|
metrics2["timestamp"]
|
|
)
|
|
|
|
def test_cpu_usage_detection(self):
|
|
"""Test CPU usage increases under load"""
|
|
# Skip CPU usage testing if get_metrics doesn't return CPU metrics
|
|
metrics = self.metrics.get_metrics()
|
|
if "cpu" not in metrics:
|
|
self.skipTest("CPU metrics not available")
|
|
|
|
# Get initial CPU usage - take average of multiple samples
|
|
initial_metrics = [self.metrics.get_metrics() for _ in range(3)]
|
|
initial_cpu = min(m.get("cpu", 0) for m in initial_metrics)
|
|
|
|
# Generate CPU load in separate process
|
|
process = multiprocessing.Process(target=cpu_load_process)
|
|
process.start()
|
|
|
|
try:
|
|
# Wait for load to register and take multiple samples
|
|
time.sleep(1.0)
|
|
load_metrics = [self.metrics.get_metrics() for _ in range(3)]
|
|
|
|
# Only test if cpu metric is available
|
|
if any("cpu" in m for m in load_metrics):
|
|
load_cpu = max(m.get("cpu", 0) for m in load_metrics)
|
|
# Verify load increases CPU usage
|
|
self.assertGreater(load_cpu, initial_cpu)
|
|
|
|
finally:
|
|
process.terminate()
|
|
process.join()
|
|
|
|
def test_cleanup(self):
|
|
"""Test monitoring stops properly"""
|
|
# Skip if the SystemMetrics class doesn't have the methods we're testing
|
|
if not hasattr(self.metrics, 'stop') or not hasattr(self.metrics, '_monitor_thread'):
|
|
self.skipTest("SystemMetrics doesn't have stop method or monitor thread")
|
|
|
|
self.metrics.stop()
|
|
|
|
# Monitor thread should finish
|
|
self.assertFalse(self.metrics._monitor_thread.is_alive())
|
|
|
|
# Should still generate valid metrics after stopping
|
|
metrics = self.metrics.get_metrics()
|
|
self.assertIsInstance(metrics, dict) |