"""Tests for scale_operation_mcp extract_topology_info and _estimate_tikv_scale_in_minutes.""" import json import pytest from aiops.infra.agent.mcp.scale_operation_mcp import ( _check_scale_in_region_migration_rate, _estimate_tikv_scale_in_minutes, _extract_topology_info_from_payload, _parse_json_payload, ) # Payload structure from real logs (ModifyClusterSize event) EVENT_PAYLOAD = { "author": "hisashi.tako@super-studio.jp", "author_id": "22969372", "cluster_id": "1379661944646414964", "detail": { "cluster_id": "1379661944646414964", "cluster_type": "dedicated_cluster", "event_created_at_ms": "1770664408279", "event_type": "ClusterScaling", "extra_info": { "topology_after": { "components": { "pd": {"quantity": 3, "tier_name": "16C32G"}, "tidb": {"quantity": 9, "tier_name": "16C128G"}, "tikv": {"quantity": 9, "tier_name": "16C64G"}, } }, "topology_before": { "components": { "pd": {"quantity": 3, "tier_name": "16C32G"}, "tidb": {"quantity": 9, "tier_name": "16C64G"}, "tikv": {"quantity": 9, "tier_name": "16C64G"}, } }, }, "status": "SCALING", "operator_id": "22969372", "project_id": "1372813089454542217", "tenant_id": "1372813089209031895", "user_email": "hisashi.tako@super-studio.jp", }, "display_event_type": "ModifyClusterSize", "display_status": "started", "happened_at_ms": "1770664408279", "project_id": "1372813089454542217", "summary": "Start to modify cluster size.", "tenant_id": "1372813089209031895", } class TestParseJsonPayload: """Test _parse_json_payload with log payload structure.""" def test_valid_json_string(self): s = json.dumps(EVENT_PAYLOAD) out = _parse_json_payload(s) assert out is not None assert "detail" in out assert "extra_info" in out["detail"] assert "topology_before" in out["detail"]["extra_info"] assert "topology_after" in out["detail"]["extra_info"] def test_extra_data_trailing_content(self): """JSON with trailing content (Extra data, simulates agent/sdk quirk).""" s = json.dumps(EVENT_PAYLOAD) + '\n{"extra": 1}' out = _parse_json_payload(s) assert out is not None assert "detail" in out assert "topology_before" in out["detail"]["extra_info"] def test_invalid_json_returns_none(self): assert _parse_json_payload("not json") is None class TestExtractTopologyInfo: """Test extract_topology_info with payload from logs.""" def test_string_payload(self): result = _extract_topology_info_from_payload(json.dumps(EVENT_PAYLOAD)) assert "error" not in result assert result["before"]["components"]["tidb"]["tier_name"] == "16C64G" assert result["after"]["components"]["tidb"]["tier_name"] == "16C128G" def test_string_payload_extra_data(self): payload_str = json.dumps(EVENT_PAYLOAD) + "\n}" result = _extract_topology_info_from_payload(payload_str) assert "error" not in result assert result["before"]["components"]["tidb"]["tier_name"] == "16C64G" assert result["after"]["components"]["tidb"]["tier_name"] == "16C128G" def test_dict_payload(self): result = _extract_topology_info_from_payload(EVENT_PAYLOAD) assert "error" not in result assert result["before"]["components"]["tidb"]["tier_name"] == "16C64G" assert result["after"]["components"]["tidb"]["tier_name"] == "16C128G" def _make_prometheus_series(instance: str, points: list) -> dict: """Build a Prometheus result series: values are [[ts, str(val)], ...].""" return { "metric": {"instance": instance}, "values": [[ts, str(val)] for ts, val in points], } @pytest.mark.asyncio async def test_estimate_tikv_scale_in_minutes_mock_data(): """Test _estimate_tikv_scale_in_minutes with mock Prometheus data.""" from unittest.mock import AsyncMock, MagicMock event_create_time = 1700000000 step = 30 def make_points(): base_ts = event_create_time - 10 * 60 points = [] for i in range(50): ts = base_ts + i * step if i < 20: val = 600 - i * 2 else: idx_to_val = {20: 500, 25: 450, 30: 400, 35: 350, 40: 300, 45: 250} val = idx_to_val.get(i, 200) points.append((float(ts), float(val))) return points points_a = make_points() points_b = make_points() points_b[20] = (points_b[20][0], 600) points_b[45] = (points_b[45][0], 200) mock_cluster = MagicMock() mock_cluster.org_id = "test-org" mock_client = MagicMock() mock_client.get_cluster = AsyncMock(return_value=mock_cluster) mock_client.query_metrics = AsyncMock( return_value={ "data": { "result": [ _make_prometheus_series("tikv-1", points_a), _make_prometheus_series("tikv-2", points_b), ], }, } ) result = await _estimate_tikv_scale_in_minutes( mock_client, "cluster-1", event_create_time ) assert result is not None instance_names, estimated = result # Instance A: v[-30]=500, v[-5]=250, dt=25*30=750s, rate=250/750, migrate=500/rate=1500s=25min # Instance B: v[-30]=600, v[-5]=200, rate=400/750, migrate=1125s=18.75min # Max total (offset 0 for both) = 25 assert estimated == pytest.approx(25.0, abs=1.0) assert "tikv-1" in instance_names and "tikv-2" in instance_names @pytest.mark.asyncio async def test_estimate_tikv_scale_in_minutes_no_scaled_in(): """When no instance has decreasing tail, returns None.""" from unittest.mock import AsyncMock, MagicMock event_create_time = 1700000000 base_ts = event_create_time - 10 * 60 # Flat/increasing tail - not scaled-in points = [(base_ts + i * 30, float(100 + i)) for i in range(50)] mock_cluster = MagicMock() mock_cluster.org_id = "test-org" mock_client = MagicMock() mock_client.get_cluster = AsyncMock(return_value=mock_cluster) mock_client.query_metrics = AsyncMock( return_value={ "data": {"result": [_make_prometheus_series("tikv-1", points)]}, } ) result = await _estimate_tikv_scale_in_minutes( mock_client, "cluster-1", event_create_time ) assert result is None @pytest.mark.asyncio async def test_estimate_tikv_scale_in_minutes_insufficient_data(): """When fewer than 30 points, returns None.""" from unittest.mock import AsyncMock, MagicMock mock_cluster = MagicMock() mock_cluster.org_id = "test-org" mock_client = MagicMock() mock_client.get_cluster = AsyncMock(return_value=mock_cluster) mock_client.query_metrics = AsyncMock( return_value={ "data": { "result": [ _make_prometheus_series( "tikv-1", [(1700000000 + i * 30, float(500 - i)) for i in range(20)], ), ], }, } ) result = await _estimate_tikv_scale_in_minutes(mock_client, "cluster-1", 1700000000) assert result is None @pytest.mark.asyncio async def test_check_scale_in_region_migration_rate_normal(): """When scale-in instances (from initial) have decreasing region count, is_decreasing_normally=True.""" from unittest.mock import AsyncMock, MagicMock now_ts = 1700000000 base_ts = now_ts - 5 * 60 # Decreasing: 500 -> 400 over 5 min points_a = [(base_ts + i * 30, float(500 - i * 2)) for i in range(11)] # Decreasing: 600 -> 450 points_b = [(base_ts + i * 30, float(600 - i * 3)) for i in range(11)] mock_cluster = MagicMock() mock_cluster.org_id = "test-org" mock_client = MagicMock() mock_client.get_cluster = AsyncMock(return_value=mock_cluster) mock_client.query_metrics = AsyncMock( return_value={ "data": { "result": [ _make_prometheus_series("tikv-1", points_a), _make_prometheus_series("tikv-2", points_b), ], }, } ) result = await _check_scale_in_region_migration_rate( mock_client, "cluster-1", ["tikv-1", "tikv-2"] ) assert result["is_decreasing_normally"] is True assert "tikv-1" in result["scale_in_instances"] assert "tikv-2" in result["scale_in_instances"] assert result["rate_per_instance"]["tikv-1"] > 0 assert result["rate_per_instance"]["tikv-2"] > 0 @pytest.mark.asyncio async def test_check_scale_in_region_migration_rate_stuck(): """When scale-in instance (from initial) has flat region count (v_start==v_end), rate=0, stuck.""" from unittest.mock import AsyncMock, MagicMock now_ts = 1700000000 base_ts = now_ts - 5 * 60 # Flat: v_start == v_end, rate = 0 points = [(base_ts + i * 30, float(500)) for i in range(11)] mock_cluster = MagicMock() mock_cluster.org_id = "test-org" mock_client = MagicMock() mock_client.get_cluster = AsyncMock(return_value=mock_cluster) mock_client.query_metrics = AsyncMock( return_value={ "data": { "result": [_make_prometheus_series("tikv-1", points)], }, } ) result = await _check_scale_in_region_migration_rate( mock_client, "cluster-1", ["tikv-1"] ) assert result["is_decreasing_normally"] is False assert "tikv-1" in result["scale_in_instances"] assert result["rate_per_instance"]["tikv-1"] == 0 assert "stuck" in result["summary"].lower() @pytest.mark.asyncio async def test_check_scale_in_region_migration_rate_one_stuck(): """One instance decreasing (normal), one flat (stuck, rate=0) - is_decreasing_normally=False.""" from unittest.mock import AsyncMock, MagicMock now_ts = 1700000000 base_ts = now_ts - 5 * 60 points_decreasing = [(base_ts + i * 30, float(500 - i * 2)) for i in range(11)] # Flat: v_start == v_end, rate = 0 (stuck) points_flat = [(base_ts + i * 30, float(500)) for i in range(11)] mock_cluster = MagicMock() mock_cluster.org_id = "test-org" mock_client = MagicMock() mock_client.get_cluster = AsyncMock(return_value=mock_cluster) mock_client.query_metrics = AsyncMock( return_value={ "data": { "result": [ _make_prometheus_series("tikv-1", points_decreasing), _make_prometheus_series("tikv-2", points_flat), ], }, } ) result = await _check_scale_in_region_migration_rate( mock_client, "cluster-1", ["tikv-1", "tikv-2"] ) assert result["is_decreasing_normally"] is False assert "tikv-1" in result["scale_in_instances"] assert "tikv-2" in result["scale_in_instances"] assert result["rate_per_instance"]["tikv-1"] > 0 assert result["rate_per_instance"]["tikv-2"] == 0 assert "stuck" in result["summary"].lower() @pytest.mark.asyncio async def test_check_scale_in_region_migration_rate_empty_instances(): """When scale_in_instances from initial is empty, return not applicable.""" from unittest.mock import AsyncMock, MagicMock mock_cluster = MagicMock() mock_cluster.org_id = "test-org" mock_client = MagicMock() mock_client.get_cluster = AsyncMock(return_value=mock_cluster) mock_client.query_metrics = AsyncMock(return_value={"data": {"result": []}}) result = await _check_scale_in_region_migration_rate(mock_client, "cluster-1", []) assert result["is_decreasing_normally"] is True assert result["scale_in_instances"] == [] assert "skip" in result["summary"].lower()