186 lines
6.4 KiB
Python
186 lines
6.4 KiB
Python
import unittest
|
|
import time
|
|
import xml.etree.ElementTree as ET
|
|
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"""
|
|
self.metrics = SystemMetrics(sample_interval=0.01)
|
|
|
|
def tearDown(self):
|
|
"""Ensure metrics monitoring is stopped"""
|
|
self.metrics.stop()
|
|
|
|
def validate_timestamp(self, timestamp: str):
|
|
"""Validate timestamp format and reasonableness"""
|
|
try:
|
|
# 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)
|
|
|
|
except Exception as e:
|
|
self.fail(f"Invalid timestamp format: {timestamp}")
|
|
|
|
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
|
|
# (allowing for normal fluctuation)
|
|
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 validate_usage_values(self, values: List[Tuple[str, int]]):
|
|
"""Validate usage percentage values are in valid range"""
|
|
for name, value in values:
|
|
self.assertIsInstance(value, int)
|
|
self.assertGreaterEqual(
|
|
value, 0,
|
|
f"{name} below valid range: {value}"
|
|
)
|
|
self.assertLessEqual(
|
|
value, 100,
|
|
f"{name} above valid range: {value}"
|
|
)
|
|
|
|
def test_initial_context(self):
|
|
"""Test initial context generation"""
|
|
context = self.metrics.generate_context(0.5)
|
|
|
|
# Verify it's an XML element
|
|
self.assertIsInstance(context, ET.Element)
|
|
self.assertEqual(context.tag, "context")
|
|
|
|
# Validate timestamp
|
|
self.validate_timestamp(context.get("time"))
|
|
|
|
# Validate memory metrics
|
|
self.validate_memory_metrics(
|
|
int(context.get("memory_used")),
|
|
int(context.get("memory_total"))
|
|
)
|
|
|
|
# Validate disk metrics
|
|
self.validate_disk_metrics(
|
|
int(context.get("disk_used")),
|
|
int(context.get("disk_total"))
|
|
)
|
|
|
|
# Validate usage percentages
|
|
self.validate_usage_values([
|
|
("CPU", int(context.get("cpu"))),
|
|
("GPU", int(context.get("gpu"))),
|
|
("Context", int(context.get("context")))
|
|
])
|
|
|
|
# Validate stdin buffer size
|
|
self.assertEqual(context.get("stdin"), "0")
|
|
|
|
def test_multiple_context_generations(self):
|
|
"""Test multiple context generations have different timestamps"""
|
|
context1 = self.metrics.generate_context(0.5)
|
|
time.sleep(1)
|
|
context2 = self.metrics.generate_context(0.5)
|
|
|
|
self.assertNotEqual(
|
|
context1.get("time"),
|
|
context2.get("time")
|
|
)
|
|
|
|
def test_cpu_usage_detection(self):
|
|
"""Test CPU usage increases under load"""
|
|
# Get initial CPU usage
|
|
context1 = self.metrics.generate_context(0.5)
|
|
initial_cpu = int(context1.get("cpu"))
|
|
|
|
# Generate CPU load in separate process
|
|
process = multiprocessing.Process(target=cpu_load_process)
|
|
process.start()
|
|
|
|
try:
|
|
# Wait for samples to accumulate
|
|
time.sleep(0.5)
|
|
|
|
# Get CPU usage under load
|
|
context2 = self.metrics.generate_context(0.5)
|
|
load_cpu = int(context2.get("cpu"))
|
|
|
|
# CPU usage should increase
|
|
self.assertGreater(load_cpu, initial_cpu)
|
|
|
|
finally:
|
|
process.terminate()
|
|
process.join()
|
|
|
|
def test_context_usage_reflection(self):
|
|
"""Test context usage parameter is reflected accurately"""
|
|
test_values = [0.0, 0.42, 0.8, 1.0]
|
|
|
|
for value in test_values:
|
|
context = self.metrics.generate_context(value)
|
|
self.assertEqual(
|
|
int(context.get("context")),
|
|
round(value * 100)
|
|
)
|
|
|
|
def test_cleanup(self):
|
|
"""Test monitoring stops properly"""
|
|
self.metrics.stop()
|
|
|
|
# Monitor thread should finish
|
|
self.assertFalse(self.metrics._monitor_thread.is_alive())
|
|
|
|
# Should still generate valid context after stopping
|
|
context = self.metrics.generate_context(0.5)
|
|
self.assertIsInstance(context, ET.Element) |