#!/usr/bin/env python3 """ Test cases for time_metrics module. Run with: pytest /opt/server-agents/orchestrator/tests/test_time_metrics.py -v """ import json import os import sys import tempfile import time from datetime import datetime, timedelta from pathlib import Path from unittest.mock import patch, MagicMock # Add lib to path sys.path.insert(0, str(Path(__file__).parent.parent / "lib")) import pytest # Import module under test from time_metrics import ( get_utc_now, get_utc_now_with_offset, parse_iso_timestamp, calculate_duration_seconds, format_duration, format_duration_human, elapsed_since, convert_to_local_time, format_timestamp_with_local, get_system_load, get_memory_usage, get_disk_usage, capture_system_context, TaskTimeTracker, create_task_time_metadata, update_task_completion_metadata, format_job_with_timing, format_logs_header, DEFAULT_TIMEZONE ) class TestTimestampFunctions: """Test timestamp generation and parsing.""" def test_get_utc_now_format(self): """UTC timestamp should be in ISO 8601 format with Z suffix.""" ts = get_utc_now() assert ts.endswith("Z") # Should be parseable dt = datetime.fromisoformat(ts[:-1]) assert dt is not None def test_get_utc_now_with_offset_format(self): """UTC timestamp with offset should end with +00:00.""" ts = get_utc_now_with_offset() assert "+00:00" in ts def test_parse_iso_timestamp_z_suffix(self): """Parse timestamp with Z suffix.""" ts = "2026-01-11T03:31:57Z" dt = parse_iso_timestamp(ts) assert dt is not None assert dt.year == 2026 assert dt.month == 1 assert dt.day == 11 assert dt.hour == 3 assert dt.minute == 31 assert dt.second == 57 def test_parse_iso_timestamp_with_offset(self): """Parse timestamp with timezone offset.""" ts = "2026-01-11T00:31:57-03:00" dt = parse_iso_timestamp(ts) assert dt is not None assert dt.hour == 0 # Local hour def test_parse_iso_timestamp_no_tz(self): """Parse timestamp without timezone (assume UTC).""" ts = "2026-01-11T03:31:57" dt = parse_iso_timestamp(ts) assert dt is not None assert dt.hour == 3 def test_parse_iso_timestamp_none(self): """None input should return None.""" assert parse_iso_timestamp(None) is None assert parse_iso_timestamp("") is None def test_parse_iso_timestamp_invalid(self): """Invalid timestamp should return None.""" assert parse_iso_timestamp("not-a-timestamp") is None assert parse_iso_timestamp("2026-99-99T99:99:99Z") is None class TestDurationCalculations: """Test duration calculation and formatting.""" def test_calculate_duration_seconds(self): """Calculate duration between two timestamps.""" start = "2026-01-11T10:00:00Z" end = "2026-01-11T10:01:00Z" duration = calculate_duration_seconds(start, end) assert duration == 60.0 def test_calculate_duration_hours(self): """Calculate duration spanning hours.""" start = "2026-01-11T10:00:00Z" end = "2026-01-11T12:30:00Z" duration = calculate_duration_seconds(start, end) assert duration == 2.5 * 3600 # 2.5 hours def test_calculate_duration_negative(self): """Duration can be negative if end is before start.""" start = "2026-01-11T12:00:00Z" end = "2026-01-11T10:00:00Z" duration = calculate_duration_seconds(start, end) assert duration < 0 def test_calculate_duration_none(self): """Invalid inputs should return None.""" assert calculate_duration_seconds(None, "2026-01-11T10:00:00Z") is None assert calculate_duration_seconds("2026-01-11T10:00:00Z", None) is None def test_format_duration_seconds(self): """Format durations under a minute.""" assert format_duration(0) == "00:00:00" assert format_duration(45) == "00:00:45" assert format_duration(59) == "00:00:59" def test_format_duration_minutes(self): """Format durations in minutes.""" assert format_duration(60) == "00:01:00" assert format_duration(125) == "00:02:05" assert format_duration(3599) == "00:59:59" def test_format_duration_hours(self): """Format durations in hours.""" assert format_duration(3600) == "01:00:00" assert format_duration(3661) == "01:01:01" assert format_duration(7200) == "02:00:00" def test_format_duration_none(self): """None or negative should return placeholder.""" assert format_duration(None) == "--:--:--" assert format_duration(-1) == "--:--:--" def test_format_duration_human_seconds(self): """Human-readable format for seconds.""" assert format_duration_human(0) == "0s" assert format_duration_human(45) == "45s" assert format_duration_human(59) == "59s" def test_format_duration_human_minutes(self): """Human-readable format for minutes.""" assert format_duration_human(60) == "1m 0s" assert format_duration_human(125) == "2m 5s" assert format_duration_human(3599) == "59m 59s" def test_format_duration_human_hours(self): """Human-readable format for hours.""" assert format_duration_human(3600) == "1h 0m 0s" assert format_duration_human(3661) == "1h 1m 1s" assert format_duration_human(7200) == "2h 0m 0s" def test_format_duration_human_days(self): """Human-readable format for days.""" assert format_duration_human(86400) == "1d 0h 0m" assert format_duration_human(90061) == "1d 1h 1m" def test_format_duration_human_none(self): """None or negative should return 'unknown'.""" assert format_duration_human(None) == "unknown" assert format_duration_human(-1) == "unknown" class TestTimezoneConversion: """Test timezone conversion functions.""" def test_convert_to_local_time_montevideo(self): """Convert UTC to Montevideo time (UTC-3).""" utc_ts = "2026-01-11T03:31:57Z" local = convert_to_local_time(utc_ts, "America/Montevideo") # Should be 00:31:57 local (UTC-3) assert "00:31:57" in local def test_convert_to_local_time_invalid(self): """Invalid timestamp should return original.""" result = convert_to_local_time("invalid", "America/Montevideo") assert "invalid" in result def test_format_timestamp_with_local(self): """Format timestamp showing both UTC and local.""" utc_ts = "2026-01-11T03:31:57Z" result = format_timestamp_with_local(utc_ts, "America/Montevideo") assert "2026-01-11T03:31:57Z" in result assert "America/Montevideo" in result class TestSystemContext: """Test system context capture functions.""" def test_get_system_load_returns_tuple(self): """System load should return 3-element tuple.""" load = get_system_load() assert isinstance(load, tuple) assert len(load) == 3 assert all(isinstance(l, (int, float)) for l in load) def test_get_memory_usage_keys(self): """Memory usage should have expected keys.""" mem = get_memory_usage() assert "total_mb" in mem assert "available_mb" in mem assert "used_mb" in mem assert "used_percent" in mem assert 0 <= mem["used_percent"] <= 100 def test_get_disk_usage_keys(self): """Disk usage should have expected keys.""" disk = get_disk_usage("/") assert "total_gb" in disk assert "free_gb" in disk assert "used_gb" in disk assert "used_percent" in disk assert 0 <= disk["used_percent"] <= 100 def test_capture_system_context_structure(self): """System context should have complete structure.""" ctx = capture_system_context() assert "timestamp" in ctx assert "system_load" in ctx assert "memory" in ctx assert "disk" in ctx assert isinstance(ctx["system_load"], list) assert len(ctx["system_load"]) == 3 assert "used_percent" in ctx["memory"] assert "available_mb" in ctx["memory"] assert "used_percent" in ctx["disk"] assert "free_gb" in ctx["disk"] class TestTaskTimeTracker: """Test TaskTimeTracker class.""" def test_tracker_initialization(self): """Tracker should initialize with task_id and project.""" tracker = TaskTimeTracker("test-001", "admin") assert tracker.task_id == "test-001" assert tracker.project == "admin" assert tracker.dispatch_time is None assert tracker.completion_time is None def test_mark_dispatched(self): """mark_dispatched should record dispatch time and context.""" tracker = TaskTimeTracker("test-001", "admin") result = tracker.mark_dispatched() assert tracker.dispatch_time is not None assert tracker.dispatch_context is not None assert "dispatch" in result assert "utc_time" in result["dispatch"] assert "system_load" in result["dispatch"] assert "memory_percent" in result["dispatch"] def test_mark_started(self): """mark_started should record start time.""" tracker = TaskTimeTracker("test-001", "admin") tracker.mark_dispatched() result = tracker.mark_started() assert tracker.start_time is not None assert "start_time" in result def test_mark_completed(self): """mark_completed should calculate duration.""" tracker = TaskTimeTracker("test-001", "admin") tracker.mark_dispatched() time.sleep(1.1) # Delay for measurable duration (must be > 1 sec for second resolution) result = tracker.mark_completed(exit_code=0) assert tracker.completion_time is not None assert "completion" in result assert "utc_time" in result["completion"] assert "duration_seconds" in result["completion"] # Duration should be at least 1 second assert result["completion"]["duration_seconds"] >= 1.0 assert "exit_code" in result["completion"] assert result["completion"]["exit_code"] == 0 def test_get_full_metrics_running(self): """get_full_metrics for running task should show elapsed.""" tracker = TaskTimeTracker("test-001", "admin") tracker.mark_dispatched() metrics = tracker.get_full_metrics() assert metrics["status"] == "running" assert "elapsed" in metrics assert "dispatch" in metrics def test_get_full_metrics_completed(self): """get_full_metrics for completed task should show duration.""" tracker = TaskTimeTracker("test-001", "admin") tracker.mark_dispatched() tracker.mark_completed(0) metrics = tracker.get_full_metrics() assert metrics["status"] == "completed" assert "completion" in metrics assert "duration_seconds" in metrics["completion"] class TestMetadataFunctions: """Test metadata helper functions.""" def test_create_task_time_metadata(self): """create_task_time_metadata should return dispatch info.""" meta = create_task_time_metadata("test-001", "admin") assert "time_metrics" in meta assert "time_tracker_data" in meta assert "dispatch" in meta["time_metrics"] assert meta["time_tracker_data"]["task_id"] == "test-001" assert meta["time_tracker_data"]["project"] == "admin" def test_update_task_completion_metadata(self): """update_task_completion_metadata should add completion info.""" # Create initial metadata meta = create_task_time_metadata("test-001", "admin") time.sleep(0.1) # Update with completion updated = update_task_completion_metadata(meta, exit_code=0) assert "time_metrics" in updated assert "completion" in updated["time_metrics"] assert updated["time_metrics"]["completion"]["exit_code"] == 0 class TestOutputFormatters: """Test output formatting functions.""" def test_format_job_with_timing_complete(self): """Format job with timing info should include all fields.""" job = { "id": "123456-abcd", "project": "admin", "status": "completed", "time_metrics": { "dispatch": { "utc_time": "2026-01-11T10:00:00Z", "system_load": [0.5, 0.6, 0.7] }, "completion": { "duration_formatted": "00:05:30" } } } result = format_job_with_timing(job) assert "123456-abcd" in result assert "admin" in result assert "completed" in result assert "10:00:00" in result assert "00:05:30" in result def test_format_job_with_timing_running(self): """Format running job should show elapsed time.""" job = { "id": "123456-abcd", "project": "admin", "status": "running", "time_metrics": { "dispatch": { "utc_time": "2026-01-11T10:00:00Z", "system_load": [0.5, 0.6, 0.7] } } } result = format_job_with_timing(job) assert "123456-abcd" in result assert "running" in result def test_format_logs_header_structure(self): """Logs header should contain timing sections.""" job = { "id": "123456-abcd", "project": "admin", "status": "completed", "time_metrics": { "dispatch": { "utc_time": "2026-01-11T10:00:00Z", "system_load": [0.5, 0.6, 0.7], "memory_percent": 65, "disk_percent": 45 }, "completion": { "utc_time": "2026-01-11T10:05:30Z", "duration_formatted": "00:05:30" } } } header = format_logs_header(job) assert "═" in header # Box drawing assert "Job:" in header assert "Agent: admin" in header assert "Dispatched:" in header assert "Status:" in header assert "System:" in header class TestElapsedSince: """Test elapsed_since function.""" def test_elapsed_since_recent(self): """elapsed_since should calculate time from now.""" # Use a timestamp 5 seconds ago past = datetime.utcnow() - timedelta(seconds=5) past_ts = past.strftime("%Y-%m-%dT%H:%M:%SZ") elapsed = elapsed_since(past_ts) # Should be around 5s (allow some tolerance) assert "s" in elapsed # Should have seconds format if __name__ == "__main__": pytest.main([__file__, "-v"])