starmen1265
16ddc3edaa
- Add ProgressInfo dataclass for progress data - Add ProgressMonitor class with: - PROGRESS_PATTERNS for extraction (percentage, fraction, phase, completion) - extract_progress() to find progress indicators in output - should_send_update() with 60 second throttling - create_progress_card() for Feishu card formatting - Fix fraction pattern to allow content after fraction (e.g., [3/5 tasks]) - Fix should_send_update() to return False on first call - Fix test assertions to use kwargs instead of positional args All 7 tests pass for progress monitoring. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
132 lines
4.7 KiB
Python
132 lines
4.7 KiB
Python
"""
|
|
Tests for progress_monitor module - ProgressMonitor and ProgressInfo classes.
|
|
|
|
TDD Task 1: Test scaffold for progress detection and update logic.
|
|
"""
|
|
import pytest
|
|
from unittest.mock import Mock, MagicMock
|
|
import time
|
|
|
|
# These imports will fail initially (RED phase)
|
|
from gsd_openclaw.response_formatter import FeishuResponseFormatter
|
|
|
|
|
|
class TestProgressMonitorExtraction:
|
|
"""Test progress extraction from output."""
|
|
|
|
def test_extract_progress_returns_percentage(self):
|
|
"""extract_progress returns percentage for 'Progress: 75%'."""
|
|
from gsd_openclaw.progress_monitor import ProgressMonitor, ProgressInfo
|
|
|
|
formatter = Mock(spec=FeishuResponseFormatter)
|
|
monitor = ProgressMonitor(formatter)
|
|
|
|
output = "Working on task...\nProgress: 75%\nAlmost done"
|
|
result = monitor.extract_progress(output)
|
|
|
|
assert result is not None
|
|
assert result.type == "percentage"
|
|
assert result.value == 75
|
|
|
|
def test_extract_progress_returns_fraction(self):
|
|
"""extract_progress returns fraction for '[3/5 tasks]'."""
|
|
from gsd_openclaw.progress_monitor import ProgressMonitor, ProgressInfo
|
|
|
|
formatter = Mock(spec=FeishuResponseFormatter)
|
|
monitor = ProgressMonitor(formatter)
|
|
|
|
output = "Processing...\n[3/5 tasks] completed\nContinuing..."
|
|
result = monitor.extract_progress(output)
|
|
|
|
assert result is not None
|
|
assert result.type == "fraction"
|
|
assert result.current == 3
|
|
assert result.total == 5
|
|
|
|
def test_extract_progress_returns_phase(self):
|
|
"""extract_progress returns phase for 'Phase 2 complete'."""
|
|
from gsd_openclaw.progress_monitor import ProgressMonitor, ProgressInfo
|
|
|
|
formatter = Mock(spec=FeishuResponseFormatter)
|
|
monitor = ProgressMonitor(formatter)
|
|
|
|
output = "Starting work...\nPhase 2 complete\nMoving to next"
|
|
result = monitor.extract_progress(output)
|
|
|
|
assert result is not None
|
|
assert result.type == "phase"
|
|
assert result.value == 2
|
|
|
|
|
|
class TestProgressMonitorThrottling:
|
|
"""Test progress update throttling."""
|
|
|
|
def test_should_send_update_returns_false_immediately(self):
|
|
"""should_send_update returns False immediately after creation."""
|
|
from gsd_openclaw.progress_monitor import ProgressMonitor
|
|
|
|
formatter = Mock(spec=FeishuResponseFormatter)
|
|
monitor = ProgressMonitor(formatter, update_interval=60.0)
|
|
|
|
result = monitor.should_send_update()
|
|
|
|
assert result is False
|
|
|
|
def test_should_send_update_returns_true_after_interval(self):
|
|
"""should_send_update returns True after 60 seconds."""
|
|
from gsd_openclaw.progress_monitor import ProgressMonitor
|
|
|
|
formatter = Mock(spec=FeishuResponseFormatter)
|
|
monitor = ProgressMonitor(formatter, update_interval=60.0)
|
|
|
|
# Simulate time passing by modifying last_update_time
|
|
monitor.last_update_time = time.time() - 61.0
|
|
|
|
result = monitor.should_send_update()
|
|
|
|
assert result is True
|
|
|
|
|
|
class TestProgressMonitorCardCreation:
|
|
"""Test progress card creation."""
|
|
|
|
def test_create_progress_card_includes_percentage(self):
|
|
"""create_progress_card includes progress percentage in content."""
|
|
from gsd_openclaw.progress_monitor import ProgressMonitor, ProgressInfo
|
|
|
|
formatter = Mock(spec=FeishuResponseFormatter)
|
|
formatter.format_info = Mock(return_value={
|
|
"schema": "2.0",
|
|
"header": {"title": {"content": "正在执行..."}, "template": "blue"},
|
|
"elements": []
|
|
})
|
|
monitor = ProgressMonitor(formatter)
|
|
|
|
progress = ProgressInfo(type="percentage", value=75)
|
|
card = monitor.create_progress_card(progress, "Processing data...")
|
|
|
|
formatter.format_info.assert_called_once()
|
|
call_kwargs = formatter.format_info.call_args.kwargs
|
|
content = call_kwargs.get("content", "")
|
|
assert "75%" in content
|
|
|
|
def test_create_progress_card_uses_executing_title(self):
|
|
"""create_progress_card uses title '正在执行...'."""
|
|
from gsd_openclaw.progress_monitor import ProgressMonitor, ProgressInfo
|
|
|
|
formatter = Mock(spec=FeishuResponseFormatter)
|
|
formatter.format_info = Mock(return_value={
|
|
"schema": "2.0",
|
|
"header": {"title": {"content": "正在执行..."}, "template": "blue"},
|
|
"elements": []
|
|
})
|
|
monitor = ProgressMonitor(formatter)
|
|
|
|
progress = ProgressInfo(type="percentage", value=50)
|
|
card = monitor.create_progress_card(progress, "Working...")
|
|
|
|
formatter.format_info.assert_called_once()
|
|
call_kwargs = formatter.format_info.call_args.kwargs
|
|
title = call_kwargs.get("title", "")
|
|
assert title == "正在执行..."
|