from __future__ import annotations import importlib.util import json import tempfile import unittest from pathlib import Path SCRIPT = Path(__file__).parents[1] / "scripts" / "calculate_user_unit_cost.py" SPEC = importlib.util.spec_from_file_location("calculate_user_unit_cost", SCRIPT) assert SPEC and SPEC.loader MODULE = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(MODULE) def row( cost_price: str, raw_price: str | None = None, *, labels=None, invoice_id: int = 1, billing_day: str | None = None, consumed_source: str = "invoice", ): value = { "tenant_id": "1", "unit": "test-unit", "labels": {"clusterID": "c1"} if labels is None else labels, "consumed_source": consumed_source, "invoice_id": invoice_id, "internal": False, "cost_price": cost_price, "raw_price": raw_price if raw_price is not None else cost_price, } if billing_day is not None: value["billing_day"] = billing_day return value def calculate(values, **kwargs): return MODULE.calculate( ((index, value) for index, value in enumerate(values, start=1)), amount_fields=("cost_price", "raw_price"), precision=2, **kwargs, ) class UserUnitCostCalculationTest(unittest.TestCase): def test_emits_billing_month_overview_with_credits(self): result = calculate( [ row("8.888", "10.999", consumed_source="invoice"), row("3.123", "3.456", consumed_source="credits"), ] ) self.assertEqual(result["format"], "manual-invoice-user-unit-cost/v2") self.assertEqual( result["billing_overview"], { "runningTotal": "14.44", "discounts": "2.44", "credits": "3.12", "totalCost": "8.88", }, ) self.assertEqual( result["billing_overview_formula"], "totalCost = runningTotal - discounts - credits", ) def test_billing_overview_uses_exact_credits_source_value(self): result = calculate([row("1.00", "2.00", consumed_source="Credits")]) self.assertEqual(result["billing_overview"]["credits"], "0.00") self.assertEqual(result["billing_overview"]["totalCost"], "1.00") def test_sums_full_precision_before_group_truncation(self): result = calculate([row("6.889"), row("0.001")]) self.assertEqual(result["group_count"], 1) self.assertEqual(result["totals"]["cost_price"], "6.89") self.assertEqual(result["groups"][0]["full_precision"]["cost_price"], "6.890") def test_truncates_each_distinct_group_before_aggregate(self): result = calculate( [row("6.889", invoice_id=1), row("0.001", invoice_id=2)] ) self.assertEqual(result["group_count"], 2) self.assertEqual(result["full_precision_totals"]["cost_price"], "6.890") self.assertEqual(result["totals"]["cost_price"], "6.88") def test_ignore_invoice_id_changes_the_effective_group(self): values = [row("0.006", invoice_id=1), row("0.006", invoice_id=2)] self.assertEqual(calculate(values)["totals"]["cost_price"], "0.00") self.assertEqual( calculate(values, ignore_invoice_id=True)["totals"]["cost_price"], "0.01", ) def test_label_objects_use_canonical_key_order(self): result = calculate( [ row("0.006", labels={"a": 1, "b": 2}), row("0.006", labels='{"b":2,"a":1}'), ] ) self.assertEqual(result["group_count"], 1) self.assertEqual(result["totals"]["cost_price"], "0.01") def test_default_is_standard(self): values = [ row("0.006", labels={"sub_sku": "a"}), row("0.006", labels={"sub_sku": "b"}), ] standard = calculate(values) ignore_labels = calculate(values, ignore_labels=True) self.assertEqual(standard["grouping_mode"], "standard") self.assertIn("labels", standard["group_fields"]) self.assertEqual(standard["totals"]["cost_price"], "0.00") self.assertEqual(ignore_labels["grouping_mode"], "ignore_labels") self.assertNotIn("labels", ignore_labels["group_fields"]) self.assertEqual(ignore_labels["totals"]["cost_price"], "0.01") def test_truncation_is_toward_zero_for_negative_values(self): result = calculate([row("-1.239")]) self.assertEqual(result["totals"]["cost_price"], "-1.23") def test_june_daily_split_is_explicit(self): values = [ row("6.889", billing_day="2026-06-01"), row("0.001", billing_day="2026-06-02"), ] default = calculate(values) june_freeze = calculate(values, include_billing_day=True) self.assertEqual(default["totals"]["cost_price"], "6.89") self.assertEqual(june_freeze["totals"]["cost_price"], "6.88") def test_rolls_truncated_api_groups_up_to_sub_sku(self): values = [ row("0.006", labels={"sub_sku": "a"}, invoice_id=1), row("0.006", labels={"sub_sku": "a"}, invoice_id=1), row("0.019", labels={"sub_sku": "a"}, invoice_id=2), row("0.029", labels={"sub_sku": "b"}, invoice_id=1), ] result = calculate( values, ignore_labels=False, rollup_fields=(("sub_sku", "labels.sub_sku"),), ) self.assertEqual(result["totals"]["cost_price"], "0.04") self.assertEqual( result["rollups"], [ {"dimensions": {"sub_sku": "a"}, "totals": {"cost_price": "0.02", "raw_price": "0.02"}}, {"dimensions": {"sub_sku": "b"}, "totals": {"cost_price": "0.02", "raw_price": "0.02"}}, ], ) def test_rejects_rollup_that_splits_one_precision_group(self): first = row("0.006") first["sub_sku"] = "a" second = row("0.006") second["sub_sku"] = "b" with self.assertRaisesRegex(ValueError, "multiple rollup values"): calculate( [first, second], rollup_fields=(("sub_sku", "sub_sku"),) ) def test_loads_ignore_labels_grouping_decision(self): decision = { "format": "manual-invoice-user-unit-cost-grouping-decision/v1", "tenant_id": "1", "grouping_mode": "ignore_labels", "ignore_labels": True, } with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "decision.json" path.write_text(json.dumps(decision), encoding="utf-8") ignore_labels, loaded = MODULE.load_grouping_decision(path, {"1"}) self.assertTrue(ignore_labels) self.assertEqual(loaded, decision) def test_rejects_grouping_decision_for_another_tenant(self): decision = { "format": "manual-invoice-user-unit-cost-grouping-decision/v1", "tenant_id": "2", "grouping_mode": "standard", "ignore_labels": False, } with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "decision.json" path.write_text(json.dumps(decision), encoding="utf-8") with self.assertRaisesRegex(ValueError, "does not match"): MODULE.load_grouping_decision(path, {"1"}) def test_rejects_inconsistent_grouping_decision(self): decision = { "format": "manual-invoice-user-unit-cost-grouping-decision/v1", "tenant_id": "1", "grouping_mode": "standard", "ignore_labels": True, } with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "decision.json" path.write_text(json.dumps(decision), encoding="utf-8") with self.assertRaisesRegex(ValueError, "does not match"): MODULE.load_grouping_decision(path, {"1"}) if __name__ == "__main__": unittest.main()