from datetime import datetime, time from zoneinfo import ZoneInfo import pytest from app.costing.engine import IntervalReading, evaluate from app.costing.tariff_models import ( ALL_DAYS, FeedInComponent, TariffPlan, TieredFeedIn, TimeWindow, UsageComponent, find_usage_coverage_gaps, ) BNE = ZoneInfo("Australia/Brisbane") def bne(y, m, d, hh, mm=0) -> datetime: return datetime(y, m, d, hh, mm, tzinfo=BNE) def test_flat_plan_with_feed_in(): plan = TariffPlan( plan_id="flat-1", retailer="Test Co", plan_name="Flat Saver", supply_charge_c_per_day=100, usage_components=[UsageComponent(label="flat", rate_c_per_kwh=30)], feed_in_tariffs=[FeedInComponent(label="fit", rate_c_per_kwh=8)], ) readings = [ IntervalReading(ts=bne(2026, 1, 1, 10), import_kwh=5, export_kwh=2), IntervalReading(ts=bne(2026, 1, 1, 20), import_kwh=3, export_kwh=0), IntervalReading(ts=bne(2026, 1, 2, 10), import_kwh=4, export_kwh=1), ] result = evaluate(plan, readings) assert result.days_with_data == 2 assert result.usage_breakdown[0].kwh == pytest.approx(12) assert result.feed_in_breakdown[0].kwh == pytest.approx(3) assert result.supply_charge_dollars == pytest.approx(2.00) assert result.total_dollars == pytest.approx(5.36) assert result.modeling_notes == [] def test_simple_time_of_use_plan(): peak = UsageComponent( label="peak", rate_c_per_kwh=40, time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(16, 0), end_time=time(20, 0))], ) offpeak = UsageComponent( label="offpeak", rate_c_per_kwh=20, time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(20, 0), end_time=time(16, 0))], ) plan = TariffPlan( plan_id="tou-1", retailer="Test Co", plan_name="Time of Use", supply_charge_c_per_day=80, usage_components=[peak, offpeak], feed_in_tariffs=[FeedInComponent(label="fit", rate_c_per_kwh=10)], ) # No structural gaps: peak + offpeak windows partition the full 24h. assert find_usage_coverage_gaps([peak, offpeak]) == [] readings = [ IntervalReading(ts=bne(2026, 3, 10, 6), import_kwh=2, export_kwh=0), IntervalReading(ts=bne(2026, 3, 10, 12), import_kwh=3, export_kwh=5), IntervalReading(ts=bne(2026, 3, 10, 18), import_kwh=4, export_kwh=0), IntervalReading(ts=bne(2026, 3, 10, 22), import_kwh=1, export_kwh=0), ] result = evaluate(plan, readings) by_label = {c.label: c for c in result.usage_breakdown} assert by_label["offpeak"].kwh == pytest.approx(6) assert by_label["peak"].kwh == pytest.approx(4) assert result.feed_in_breakdown[0].kwh == pytest.approx(5) assert result.supply_charge_dollars == pytest.approx(0.80) assert result.total_dollars == pytest.approx(3.10) def test_seasonal_time_of_use_plan(): def peak(months, rate): return UsageComponent( label=f"peak-{months[0]}", rate_c_per_kwh=rate, season_months=months, time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(16, 0), end_time=time(20, 0))], ) def offpeak(months, rate): return UsageComponent( label=f"offpeak-{months[0]}", rate_c_per_kwh=rate, season_months=months, time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(20, 0), end_time=time(16, 0))], ) summer = [12, 1, 2] winter = [3, 4, 5, 6, 7, 8, 9, 10, 11] plan = TariffPlan( plan_id="seasonal-tou-1", retailer="Test Co", plan_name="Seasonal TOU", supply_charge_c_per_day=90, usage_components=[ peak(summer, 50), offpeak(summer, 22), peak(winter, 35), offpeak(winter, 18), ], ) readings = [ IntervalReading(ts=bne(2026, 1, 15, 18), import_kwh=2, export_kwh=0), # summer peak IntervalReading(ts=bne(2026, 1, 15, 8), import_kwh=3, export_kwh=0), # summer offpeak IntervalReading(ts=bne(2026, 6, 15, 18), import_kwh=2, export_kwh=0), # winter peak IntervalReading(ts=bne(2026, 6, 15, 8), import_kwh=3, export_kwh=0), # winter offpeak ] result = evaluate(plan, readings) assert result.days_with_data == 2 assert result.supply_charge_dollars == pytest.approx(1.80) # 2*0.50 + 3*0.22 + 2*0.35 + 3*0.18 = 1.00 + 0.66 + 0.70 + 0.54 = 2.90 assert sum(c.cost_dollars for c in result.usage_breakdown) == pytest.approx(2.90) assert result.total_dollars == pytest.approx(4.70) assert result.modeling_notes == [] def test_time_of_use_feed_in_tariff(): plan = TariffPlan( plan_id="tou-fit-1", retailer="Test Co", plan_name="Solar Soak", supply_charge_c_per_day=70, usage_components=[UsageComponent(label="flat", rate_c_per_kwh=25)], feed_in_tariffs=[ FeedInComponent( label="midday-soak", rate_c_per_kwh=5, time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(10, 0), end_time=time(15, 0))], ), FeedInComponent( label="standard", rate_c_per_kwh=12, time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(15, 0), end_time=time(10, 0))], ), ], ) readings = [ IntervalReading(ts=bne(2026, 2, 1, 8), import_kwh=1, export_kwh=0.5), IntervalReading(ts=bne(2026, 2, 1, 12), import_kwh=0, export_kwh=4), IntervalReading(ts=bne(2026, 2, 1, 18), import_kwh=2, export_kwh=0), ] result = evaluate(plan, readings) by_label = {c.label: c for c in result.feed_in_breakdown} assert by_label["midday-soak"].kwh == pytest.approx(4) assert by_label["standard"].kwh == pytest.approx(0.5) assert result.total_dollars == pytest.approx(1.19) def test_gap_in_usage_windows_is_reported_and_uncharged(): peak_only = UsageComponent( label="peak", rate_c_per_kwh=40, time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(16, 0), end_time=time(20, 0))], ) plan = TariffPlan( plan_id="broken-1", retailer="Test Co", plan_name="Missing Offpeak", supply_charge_c_per_day=0, usage_components=[peak_only], ) assert find_usage_coverage_gaps([peak_only]) != [] readings = [ IntervalReading(ts=bne(2026, 1, 1, 8), import_kwh=5, export_kwh=0), # falls in the gap IntervalReading(ts=bne(2026, 1, 1, 18), import_kwh=2, export_kwh=0), # matches peak ] result = evaluate(plan, readings) assert result.usage_breakdown[0].kwh == pytest.approx(2) assert result.total_dollars == pytest.approx(0.80) assert any("did not match any tariff window" in note for note in result.modeling_notes) def test_split_time_window_under_shared_label_is_not_double_counted(): # "Shoulder" needs two disjoint windows (7am-4pm and 9pm-10pm) under one # label -- this must sum once, not double-count the cost. shoulder_1 = UsageComponent( label="Shoulder", rate_c_per_kwh=36, time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(7, 0), end_time=time(16, 0))], ) shoulder_2 = UsageComponent( label="Shoulder", rate_c_per_kwh=36, time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(21, 0), end_time=time(22, 0))], ) peak = UsageComponent( label="Peak", rate_c_per_kwh=48, time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(16, 0), end_time=time(21, 0))], ) offpeak = UsageComponent( label="Off-peak", rate_c_per_kwh=31, time_windows=[TimeWindow(days=ALL_DAYS, start_time=time(22, 0), end_time=time(7, 0))], ) plan = TariffPlan( plan_id="split-shoulder-1", retailer="Test Co", plan_name="Split Shoulder", supply_charge_c_per_day=0, usage_components=[peak, shoulder_1, shoulder_2, offpeak], ) assert find_usage_coverage_gaps([peak, shoulder_1, shoulder_2, offpeak]) == [] readings = [ IntervalReading(ts=bne(2026, 1, 1, 9), import_kwh=2, export_kwh=0), # shoulder window 1 IntervalReading(ts=bne(2026, 1, 1, 21, 30), import_kwh=3, export_kwh=0), # shoulder window 2 ] result = evaluate(plan, readings) shoulder_rows = [c for c in result.usage_breakdown if c.label == "Shoulder"] assert len(shoulder_rows) == 1 assert shoulder_rows[0].kwh == pytest.approx(5) assert result.total_dollars == pytest.approx(5 * 0.36) def test_tiered_daily_feed_in_tariff(): # Modeled on Alinta's QLD "Solar Balance" FiT: first 10 kWh/day at 9c, # the remainder at 2c -- a *daily total export* threshold, not a # time-of-use one. plan = TariffPlan( plan_id="tiered-fit-1", retailer="Test Co", plan_name="Solar Balance", supply_charge_c_per_day=100, usage_components=[UsageComponent(label="flat", rate_c_per_kwh=20)], tiered_feed_in=TieredFeedIn(daily_threshold_kwh=10, rate_below_c_per_kwh=9, rate_above_c_per_kwh=2), ) readings = [ # Day 1: 7 kWh exported total, all below the 10 kWh threshold. IntervalReading(ts=bne(2026, 1, 1, 8), import_kwh=2, export_kwh=3), IntervalReading(ts=bne(2026, 1, 1, 12), import_kwh=1, export_kwh=4), # Day 2: 15 kWh exported total -- 10 below threshold, 5 above. IntervalReading(ts=bne(2026, 1, 2, 8), import_kwh=1, export_kwh=5), IntervalReading(ts=bne(2026, 1, 2, 12), import_kwh=1, export_kwh=10), ] result = evaluate(plan, readings) by_label = {c.label: c for c in result.feed_in_breakdown} assert by_label["First 10 kWh/day"].kwh == pytest.approx(17) # 7 + min(15,10) assert by_label["Remaining balance"].kwh == pytest.approx(5) # max(15-10, 0) # credit = 17*0.09 + 5*0.02 = 1.53 + 0.10 = 1.63 # usage = (2+1+1+1)*0.20 = 1.00; supply = 2 days * $1.00 = $2.00 # total = 1.00 + 2.00 - 1.63 = 1.37 assert result.total_dollars == pytest.approx(1.37) assert result.modeling_notes == [] def test_conditional_discount_kept_out_of_headline_total(): plan = TariffPlan( plan_id="discount-1", retailer="Test Co", plan_name="Pay On Time", supply_charge_c_per_day=0, usage_components=[UsageComponent(label="flat", rate_c_per_kwh=100)], discounts=[ {"description": "Pay on time discount", "percent": 10, "applies_to": "usage", "auto_apply": False} ], ) readings = [IntervalReading(ts=bne(2026, 1, 1, 12), import_kwh=10, export_kwh=0)] result = evaluate(plan, readings) assert result.total_dollars == pytest.approx(10.00) assert result.potential_conditional_savings_dollars == pytest.approx(1.00) assert any("Pay on time discount" in note for note in result.modeling_notes)