"""Unit tests for the cloud-efficiency rule engine, aggregation, cost matcher, and metric rollup — pure logic, no DB.""" from __future__ import annotations from datetime import UTC, datetime from decimal import Decimal from app.domains.cmdb.efficiency import ( EfficiencyRules, WindowStats, _percentile, aggregate_window, classify, estimate_savings, match_cost, ) from app.domains.cmdb.metrics_sync import _rollup_points from app.domains.cmdb.models import CmdbServerMetricsDaily from app.domains.cmdb.providers.base import ServerMetricPoint RULES = EfficiencyRules() def _stats(**kw) -> WindowStats: base = dict(has_any_rows=True, data_days=10, cpu_p95=50.0, cpu_max=60.0) base.update(kw) return WindowStats(**base) # ── classify ───────────────────────────────────────────────────────── def test_no_rows_is_no_metrics(): v = classify(WindowStats(has_any_rows=False), RULES) assert v.verdict == "no_metrics" assert v.recommendation == "none" def test_too_few_days_is_insufficient(): v = classify(_stats(data_days=1), RULES) assert v.verdict == "insufficient_data" def test_idle_recommends_stop(): v = classify(_stats(cpu_p95=2.0, cpu_max=4.0, net_in_avg=100, net_out_avg=200), RULES) assert v.verdict == "idle" assert v.recommendation == "stop" assert "cpu_p95_idle" in v.reasons def test_idle_cpu_but_high_memory_is_not_idle(): # Memory-bound box: low CPU but memory in use → must NOT recommend stop. v = classify( _stats(cpu_p95=2.0, cpu_max=4.0, net_in_avg=100, net_out_avg=100, mem_p95=80.0), RULES, ) assert v.verdict == "healthy" assert v.recommendation == "keep" assert "mem_bound" in v.reasons def test_oversized_recommends_downsize(): v = classify(_stats(cpu_p95=15.0, cpu_max=30.0, net_in_avg=5_000_000), RULES) assert v.verdict == "oversized" assert v.recommendation == "downsize" def test_oversized_but_memory_bound_investigates(): v = classify( _stats(cpu_p95=15.0, cpu_max=30.0, mem_p95=70.0), RULES, ) assert v.verdict == "healthy" assert v.recommendation == "investigate" def test_busy_keeps(): v = classify(_stats(cpu_p95=85.0, cpu_max=95.0), RULES) assert v.verdict == "busy" assert v.recommendation == "keep" def test_healthy_default_keeps(): v = classify(_stats(cpu_p95=50.0, cpu_max=65.0), RULES) assert v.verdict == "healthy" assert v.recommendation == "keep" def test_idle_needs_low_network_too(): # Low CPU but chatty network → not idle (falls through to oversized/healthy). v = classify( _stats(cpu_p95=2.0, cpu_max=4.0, net_in_avg=10_000_000, net_out_avg=10_000_000), RULES, ) assert v.verdict != "idle" # ── savings ────────────────────────────────────────────────────────── def test_savings_stop_is_full_cost(): v = classify(_stats(cpu_p95=2.0, cpu_max=3.0, net_in_avg=0, net_out_avg=0), RULES) assert estimate_savings(v, Decimal("100.00"), RULES) == Decimal("100.00") def test_savings_downsize_is_ratio(): v = classify(_stats(cpu_p95=15.0, cpu_max=30.0), RULES) assert estimate_savings(v, Decimal("100.00"), RULES) == Decimal("50.00") def test_savings_keep_is_zero(): v = classify(_stats(cpu_p95=85.0), RULES) assert estimate_savings(v, Decimal("100.00"), RULES) == Decimal("0.00") def test_savings_none_when_cost_unknown(): v = classify(_stats(cpu_p95=2.0, cpu_max=3.0, net_in_avg=0, net_out_avg=0), RULES) assert estimate_savings(v, None, RULES) is None # ── cost matcher ───────────────────────────────────────────────────── def test_match_cost_exact(): cm = {"i-0abc12345": Decimal("42.00")} assert match_cost("i-0abc12345", cm) == Decimal("42.00") def test_match_cost_arn_suffix(): arn = "arn:aws:ec2:us-east-1:123456789012:instance/i-0abc12345" cm = {arn: Decimal("7.50")} assert match_cost("i-0abc12345", cm) == Decimal("7.50") def test_match_cost_short_id_guarded(): # Too-short id must not fuzzy-match arbitrary keys. cm = {"arn:aws:.../instance/abc": Decimal("9.99")} assert match_cost("abc", cm) is None def test_match_cost_no_match_returns_none(): cm = {"i-999": Decimal("1.00")} assert match_cost("i-0abc12345", cm) is None def test_match_cost_azure_arm_by_candidate_and_name(): # Azure: instance_id is the vmId GUID, but the cost resource_id is the # ARM path (ending in the VM name, often lowercased). match_cost must # find it via the raw ARM id candidate OR the name suffix. arm = "/subscriptions/s1/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/prod-web-01" cm = {arm.lower(): Decimal("77.00")} vmid = "1a2b3c4d-0000-1111-2222-333344445555" # exact (case-insensitive) on the ARM identity candidate assert match_cost([vmid, arm], cm, name_candidates=["prod-web-01"]) == Decimal("77.00") # name-suffix path: only the name is known assert match_cost([vmid], cm, name_candidates=["prod-web-01"]) == Decimal("77.00") # vmId alone never matches an ARM path assert match_cost([vmid], cm) is None def test_match_cost_name_not_used_in_exact_phase(): # A display NAME must NOT exact-match a DIFFERENT resource's real cost # resource_id. Server A is named "webserver01"; that string is literally # the cost resource_id of a different box B. The name is suffix-phase # only (no "/" or ":" anchor on the bare key) → A must not steal it. cm = {"webserver01": Decimal("500.00")} assert match_cost(["i-aaaa1111bb"], cm, name_candidates=["webserver01"]) is None # An identity candidate that IS the exact key still matches (that's B). assert match_cost(["webserver01"], cm) == Decimal("500.00") def test_match_cost_name_suffix_still_matches(): # The legitimate Azure case: the name is an anchored suffix of the ARM # path — must still attribute via the suffix phase. cm = {"/subscriptions/s/.../virtualmachines/prod-web-01": Decimal("42.00")} assert match_cost(["i-vmid"], cm, name_candidates=["prod-web-01"]) == Decimal("42.00") def test_match_cost_list_all_none(): assert match_cost([None, None], {"x": Decimal("1")}) is None # ── percentile ─────────────────────────────────────────────────────── def test_percentile_basic(): assert _percentile([10.0], 0.95) == 10.0 assert _percentile([], 0.95) is None # p95 of 1..100 ≈ 95-96 p = _percentile([float(i) for i in range(1, 101)], 0.95) assert 94.0 <= p <= 97.0 # ── aggregation ────────────────────────────────────────────────────── def test_aggregate_window_counts_data_days(): rows = [ CmdbServerMetricsDaily( cpu_pct_avg=10.0, cpu_pct_p95=20.0, cpu_pct_max=30.0, net_in_bytes_avg=100, net_out_bytes_avg=200, sample_count=24, ), CmdbServerMetricsDaily( cpu_pct_avg=None, cpu_pct_p95=None, cpu_pct_max=None, sample_count=0, ), ] stats = aggregate_window(rows) assert stats.has_any_rows is True assert stats.data_days == 1 # only the row with a CPU sample counts assert stats.cpu_avg == 10.0 def test_aggregate_empty(): stats = aggregate_window([]) assert stats.has_any_rows is False assert stats.data_days == 0 # ── metric rollup (hourly points → daily) ──────────────────────────── def test_rollup_points_buckets_by_day(): pts = [ ServerMetricPoint( instance_id="i-1", timestamp=datetime(2026, 7, 1, h, 0, tzinfo=UTC), cpu_pct=float(h), cpu_pct_max=float(h) + 1, net_in_bytes=1000.0, net_out_bytes=2000.0, ) for h in range(24) ] # add one point on the next day pts.append(ServerMetricPoint( instance_id="i-1", timestamp=datetime(2026, 7, 2, 0, 0, tzinfo=UTC), cpu_pct=99.0, )) rollups = _rollup_points(pts) assert (("i-1", datetime(2026, 7, 1).date())) in rollups day1 = rollups[("i-1", datetime(2026, 7, 1).date())] assert day1["sample_count"] == 24 assert day1["cpu_pct_avg"] == sum(range(24)) / 24 assert day1["cpu_pct_max"] == 24.0 # max of hourly maxes (23+1) assert day1["net_in_bytes_avg"] == 1000 day2 = rollups[("i-1", datetime(2026, 7, 2).date())] assert day2["cpu_pct_avg"] == 99.0 assert day2["sample_count"] == 1 def test_rollup_missing_memory_stays_none(): pts = [ServerMetricPoint( instance_id="i-2", timestamp=datetime(2026, 7, 1, 0, 0, tzinfo=UTC), cpu_pct=5.0, # no mem/disk )] r = _rollup_points(pts)[("i-2", datetime(2026, 7, 1).date())] assert r["mem_pct_avg"] is None assert r["disk_used_pct_avg"] is None assert r["cpu_pct_avg"] == 5.0