"""CloudWatch 클라이언트 테스트
CloudWatch 클라이언트의 기능을 검증하는 단위 테스트입니다.
Requirements:
- 1.2: AWS 자격 증명 로드
- 6.2: 지표 데이터 조회
- 6.3: EBS 지표 목록
- 6.4: 통계 유형 지원
"""
import pytest
from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock, patch
from ebs_cloudwatch_metrics.cloudwatch_client import (
CloudWatchClient,
SUPPORTED_EBS_METRICS,
SUPPORTED_STATISTICS,
EBS_NAMESPACE,
)
from ebs_cloudwatch_metrics.models import MetricResult, MetricDataPoint
class TestCloudWatchClientInit:
"""CloudWatch 클라이언트 초기화 테스트"""
@patch("ebs_cloudwatch_metrics.cloudwatch_client.boto3.client")
def test_init_with_region(self, mock_boto_client):
"""리전을 지정하여 클라이언트 초기화"""
client = CloudWatchClient(region="ap-northeast-2")
mock_boto_client.assert_called_once_with(
"cloudwatch", region_name="ap-northeast-2"
)
assert client.region == "ap-northeast-2"
@patch("ebs_cloudwatch_metrics.cloudwatch_client.boto3.client")
def test_init_without_region(self, mock_boto_client):
"""리전 없이 클라이언트 초기화 (기본 리전 사용)"""
client = CloudWatchClient()
mock_boto_client.assert_called_once_with("cloudwatch")
assert client.region is None
class TestListAvailableMetrics:
"""list_available_metrics 메서드 테스트"""
@patch("ebs_cloudwatch_metrics.cloudwatch_client.boto3.client")
def test_returns_all_supported_metrics(self, mock_boto_client):
"""모든 지원 지표 목록 반환 확인"""
client = CloudWatchClient()
metrics = client.list_available_metrics()
# Requirements 6.3: 모든 EBS 지표 지원
expected_metrics = [
"VolumeReadOps",
"VolumeWriteOps",
"VolumeReadBytes",
"VolumeWriteBytes",
"VolumeTotalReadTime",
"VolumeTotalWriteTime",
"VolumeIdleTime",
"VolumeQueueLength",
"VolumeThroughputPercentage",
"VolumeConsumedReadWriteOps",
"BurstBalance",
]
assert metrics == expected_metrics
assert len(metrics) == 11
@patch("ebs_cloudwatch_metrics.cloudwatch_client.boto3.client")
def test_returns_copy_of_list(self, mock_boto_client):
"""반환된 목록이 원본의 복사본인지 확인"""
client = CloudWatchClient()
metrics1 = client.list_available_metrics()
metrics2 = client.list_available_metrics()
# 수정해도 원본에 영향 없음
metrics1.append("TestMetric")
assert "TestMetric" not in metrics2
class TestGetMetricStatistics:
"""get_metric_statistics 메서드 테스트"""
@pytest.fixture
def mock_client(self):
"""모킹된 CloudWatch 클라이언트 픽스처"""
with patch("ebs_cloudwatch_metrics.cloudwatch_client.boto3.client") as mock:
client = CloudWatchClient(region="ap-northeast-2")
yield client, mock.return_value
@pytest.mark.asyncio
async def test_invalid_metric_name_raises_error(self, mock_client):
"""지원하지 않는 지표 이름에 대해 ValueError 발생"""
client, _ = mock_client
with pytest.raises(ValueError) as exc_info:
await client.get_metric_statistics(
volume_id="vol-1234567890abcdef0",
metric_name="InvalidMetric",
start_time=datetime.now(timezone.utc) - timedelta(hours=1),
end_time=datetime.now(timezone.utc),
)
assert "지원하지 않는 지표 이름입니다" in str(exc_info.value)
assert "InvalidMetric" in str(exc_info.value)
@pytest.mark.asyncio
async def test_invalid_statistic_raises_error(self, mock_client):
"""지원하지 않는 통계 유형에 대해 ValueError 발생"""
client, _ = mock_client
with pytest.raises(ValueError) as exc_info:
await client.get_metric_statistics(
volume_id="vol-1234567890abcdef0",
metric_name="VolumeReadOps",
start_time=datetime.now(timezone.utc) - timedelta(hours=1),
end_time=datetime.now(timezone.utc),
statistics=["InvalidStat"],
)
assert "지원하지 않는 통계 유형입니다" in str(exc_info.value)
assert "InvalidStat" in str(exc_info.value)
@pytest.mark.asyncio
async def test_calls_cloudwatch_api_with_correct_params(self, mock_client):
"""CloudWatch API가 올바른 파라미터로 호출되는지 확인"""
client, boto_mock = mock_client
start_time = datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
end_time = datetime(2024, 1, 1, 1, 0, 0, tzinfo=timezone.utc)
boto_mock.get_metric_statistics.return_value = {"Datapoints": []}
await client.get_metric_statistics(
volume_id="vol-1234567890abcdef0",
metric_name="VolumeReadOps",
start_time=start_time,
end_time=end_time,
period=300,
statistics=["Average", "Maximum"],
)
# API 호출 파라미터 검증
boto_mock.get_metric_statistics.assert_called_once()
call_kwargs = boto_mock.get_metric_statistics.call_args[1]
assert call_kwargs["Namespace"] == "AWS/EBS"
assert call_kwargs["MetricName"] == "VolumeReadOps"
assert call_kwargs["Dimensions"] == [{"Name": "VolumeId", "Value": "vol-1234567890abcdef0"}]
assert call_kwargs["StartTime"] == start_time
assert call_kwargs["EndTime"] == end_time
assert call_kwargs["Period"] == 300
assert call_kwargs["Statistics"] == ["Average", "Maximum"]
@pytest.mark.asyncio
async def test_returns_metric_result_with_datapoints(self, mock_client):
"""데이터 포인트가 있는 MetricResult 반환 확인"""
client, boto_mock = mock_client
start_time = datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
end_time = datetime(2024, 1, 1, 1, 0, 0, tzinfo=timezone.utc)
# 모킹된 응답
boto_mock.get_metric_statistics.return_value = {
"Datapoints": [
{
"Timestamp": datetime(2024, 1, 1, 0, 30, 0, tzinfo=timezone.utc),
"Average": 100.0,
"Unit": "Count",
},
{
"Timestamp": datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
"Average": 50.0,
"Unit": "Count",
},
]
}
result = await client.get_metric_statistics(
volume_id="vol-1234567890abcdef0",
metric_name="VolumeReadOps",
start_time=start_time,
end_time=end_time,
statistics=["Average"],
)
assert isinstance(result, MetricResult)
assert result.metric_name == "VolumeReadOps"
assert result.volume_id == "vol-1234567890abcdef0"
assert len(result.datapoints) == 2
# 타임스탬프 기준 정렬 확인
assert result.datapoints[0].timestamp < result.datapoints[1].timestamp
assert result.datapoints[0].value == 50.0
assert result.datapoints[1].value == 100.0
# 평균 계산 확인
assert result.average == 75.0 # (50 + 100) / 2
@pytest.mark.asyncio
async def test_returns_empty_result_when_no_datapoints(self, mock_client):
"""데이터 포인트가 없을 때 빈 결과 반환"""
client, boto_mock = mock_client
boto_mock.get_metric_statistics.return_value = {"Datapoints": []}
result = await client.get_metric_statistics(
volume_id="vol-1234567890abcdef0",
metric_name="VolumeReadOps",
start_time=datetime.now(timezone.utc) - timedelta(hours=1),
end_time=datetime.now(timezone.utc),
)
assert result.datapoints == []
assert result.average is None
assert result.maximum is None
assert result.minimum is None
assert result.sum is None
@pytest.mark.asyncio
async def test_default_statistics_is_average(self, mock_client):
"""기본 통계 유형이 Average인지 확인"""
client, boto_mock = mock_client
boto_mock.get_metric_statistics.return_value = {"Datapoints": []}
await client.get_metric_statistics(
volume_id="vol-1234567890abcdef0",
metric_name="VolumeReadOps",
start_time=datetime.now(timezone.utc) - timedelta(hours=1),
end_time=datetime.now(timezone.utc),
)
call_kwargs = boto_mock.get_metric_statistics.call_args[1]
assert call_kwargs["Statistics"] == ["Average"]
@pytest.mark.asyncio
async def test_calculates_all_statistics(self, mock_client):
"""모든 통계 유형 계산 확인"""
client, boto_mock = mock_client
boto_mock.get_metric_statistics.return_value = {
"Datapoints": [
{"Timestamp": datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc), "Average": 10.0, "Maximum": 15.0, "Minimum": 5.0, "Sum": 100.0, "Unit": "Count"},
{"Timestamp": datetime(2024, 1, 1, 0, 5, 0, tzinfo=timezone.utc), "Average": 20.0, "Maximum": 25.0, "Minimum": 15.0, "Sum": 200.0, "Unit": "Count"},
{"Timestamp": datetime(2024, 1, 1, 0, 10, 0, tzinfo=timezone.utc), "Average": 30.0, "Maximum": 35.0, "Minimum": 25.0, "Sum": 300.0, "Unit": "Count"},
]
}
result = await client.get_metric_statistics(
volume_id="vol-1234567890abcdef0",
metric_name="VolumeReadOps",
start_time=datetime.now(timezone.utc) - timedelta(hours=1),
end_time=datetime.now(timezone.utc),
statistics=["Average", "Maximum", "Minimum", "Sum"],
)
# Average 값들의 평균: (10 + 20 + 30) / 3 = 20
assert result.average == 20.0
# Average 값들의 최대: max(10, 20, 30) = 30
assert result.maximum == 30.0
# Average 값들의 최소: min(10, 20, 30) = 10
assert result.minimum == 10.0
# Average 값들의 합: 10 + 20 + 30 = 60
assert result.sum == 60.0
class TestSupportedMetricsAndStatistics:
"""지원 지표 및 통계 상수 테스트"""
def test_supported_ebs_metrics_contains_required_metrics(self):
"""Requirements 6.3: 필수 EBS 지표 포함 확인"""
required_metrics = [
"VolumeReadOps",
"VolumeWriteOps",
"VolumeReadBytes",
"VolumeWriteBytes",
"VolumeTotalReadTime",
"VolumeTotalWriteTime",
"VolumeIdleTime",
"VolumeQueueLength",
"VolumeThroughputPercentage",
"VolumeConsumedReadWriteOps",
"BurstBalance",
]
for metric in required_metrics:
assert metric in SUPPORTED_EBS_METRICS, f"Missing metric: {metric}"
def test_supported_statistics_contains_required_types(self):
"""Requirements 6.4: 필수 통계 유형 포함 확인"""
required_statistics = ["Average", "Sum", "Minimum", "Maximum", "SampleCount"]
for stat in required_statistics:
assert stat in SUPPORTED_STATISTICS, f"Missing statistic: {stat}"
def test_ebs_namespace_is_correct(self):
"""CloudWatch 네임스페이스 확인"""
assert EBS_NAMESPACE == "AWS/EBS"
