import copy import io import importlib.util import json import unittest from datetime import date from contextlib import redirect_stdout from pathlib import Path from types import SimpleNamespace from unittest.mock import Mock, patch import pandas as pd import pyarrow.parquet as pq from google.api_core.exceptions import Conflict, NotFound import billing_inputs as inputs import monthly_billing as worker BUCKET = "fixture-artifacts" SCOPE = {"close_run_id": "close-fixture-01", "bill_month": "2026-08", "mode": "actual", "definition_hash": "a" * 16} ALLOCATOR_SHA = inputs.sha256((Path(__file__).parents[1] / "bigquery/monthly_cloud_billing_revrec.py").read_bytes()) def fixture(custom_frames=None, scope=None, snapshot_date="2026-09-01"): scope = scope or SCOPE freeze_id = "freeze-attempt-01" prefix = "gs://" + BUCKET + "/finance-portal/frozen/" + inputs.sha256(inputs.canonical([scope["close_run_id"], freeze_id])) frames = custom_frames if custom_frames is not None else { "billing": pd.DataFrame([{"billing_month": "2026-08", "tenant_id": "t1", "accumulate_revenue_to_org_tenant_id": "t1", "actual_revenue_recognition_date": "2026-09-01"}]), "orders": pd.DataFrame([{"date_day": date(2026, 9, 1), "order_service_start_date": date(2026, 1, 1), "accumulate_revenue_to_org_tenant_id": "t1", "accumulated_actual_revenue_before_billing": 3.5, "accumulated_actual_adjusted_revenue_before_billing": 3.5, "accumulated_actual_credit_amount_before_billing": 0.0}]), } manifest = {**scope, "contract": "finance-billing-frozen-inputs/v1", "freeze_attempt_id": freeze_id, "allocator_sha256": ALLOCATOR_SHA, "order_snapshot_date": snapshot_date, "source_cutoff": "2026-09-01T00:00:00Z", "upstream_receipts": [{"run_id": "fixture-input-run", "sha256": "b" * 64}], "business_scope": {"kind": "billing_tenant_set", "sha256": inputs.sha256(inputs.canonical(sorted(set(frames["billing"]["accumulate_revenue_to_org_tenant_id"]))))}, "inputs": {}} objects = {} for kind, frame in frames.items(): body = frame.to_parquet(index=False) uri = prefix + "/" + kind + ".parquet" schema = pq.ParquetFile(io.BytesIO(body)).schema_arrow.remove_metadata() manifest["inputs"][kind] = {"uri": uri, "generation": "123", "sha256": inputs.sha256(body), "rows": len(frame), "schema_sha256": inputs.sha256(schema.serialize().to_pybytes())} objects[uri] = body body = inputs.canonical(manifest) ref = {"uri": prefix + "/manifest.json", "generation": "123", "sha256": inputs.sha256(body)} objects[ref["uri"]] = body store = Mock() def blob(name, generation): body = objects["gs://" + BUCKET + "/" + name] value = Mock(size=len(body), generation=generation) value.download_as_bytes.return_value = body return value store.bucket.return_value.blob.side_effect = blob return frames, manifest, ref, objects, store class FrozenInputTest(unittest.TestCase): def test_parquet_roundtrip_preserves_real_allocation_results(self): path = Path(__file__).parents[1] / "bigquery/test_monthly_cloud_billing_revrec.py" spec = importlib.util.spec_from_file_location("allocation_fixtures", path) module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module) case = module.ActualClosingShortfallTest() billing = pd.DataFrame([case._actual_billing(2966.0)]) order = case._commitment_order("fixture-order", "tenant-1", 40000.0, 27330.29) order["date_day"] = date(2026, 7, 1) frames = {"billing": billing, "orders": pd.DataFrame([order])} scope = {**SCOPE, "bill_month": "2026-06"} _, _, ref, _, store = fixture(frames, scope, "2026-07-01") decoded, _ = inputs.load_frozen_inputs(store, BUCKET, ref, scope, ALLOCATOR_SHA) allocator = worker.load_allocator() expected = allocator.process(frames["orders"].copy(), frames["billing"].copy(), is_recognised_revenue=True) actual = allocator.process(decoded["orders"], decoded["billing"], is_recognised_revenue=True) for left, right in zip(expected, actual): pd.testing.assert_frame_equal(left, right) def test_reads_real_parquet_with_exact_generation_schema_scope(self): _, manifest, ref, _, store = fixture() frames, got = inputs.load_frozen_inputs(store, BUCKET, ref, SCOPE, ALLOCATOR_SHA) self.assertEqual(got, manifest) self.assertEqual(frames["orders"].iloc[0]["accumulated_actual_revenue_before_billing"], 3.5) self.assertEqual(store.bucket.return_value.blob.call_count, 3) self.assertTrue(all(call.kwargs["generation"] == 123 for call in store.bucket.return_value.blob.call_args_list)) def test_manifest_cannot_mix_scope_algorithm_attempt_or_inputs(self): _, original, ref, _, _ = fixture() changes = [("close_run_id", "other-close"), ("bill_month", "2026-07"), ("mode", "predicted"), ("allocator_sha256", "c" * 64), ("freeze_attempt_id", "other-attempt"), ("upstream_receipts", []), ("source_cutoff", "2026-09-01 00:00:00"), ("business_scope", {}), ("inputs", {})] for key, value in changes: manifest = copy.deepcopy(original); manifest[key] = value with self.subTest(key=key), self.assertRaises(inputs.InputContractError): inputs.validate_manifest(manifest, ref, SCOPE, ALLOCATOR_SHA, BUCKET) def test_bad_digest_or_generation_never_yields_inputs(self): _, _, ref, _, store = fixture() for change in ({"sha256": "0" * 64}, {"generation": "latest"}, {"uri": ref["uri"].replace(BUCKET, "other-bucket")}): with self.subTest(change=change), self.assertRaises(inputs.InputContractError): inputs.pinned_read(store, BUCKET, {**ref, **change}, inputs.MAX_MANIFEST_BYTES) def test_parquet_row_schema_and_decompression_limits(self): _, manifest, _, objects, _ = fixture() ref = manifest["inputs"]["orders"]; body = objects[ref["uri"]] for change in ({"rows": 2}, {"schema_sha256": "0" * 64}): with self.assertRaises(inputs.InputContractError): inputs.decode_frame(body, {**ref, **change}) with patch.object(inputs, "MAX_UNCOMPRESSED_BYTES", 1), self.assertRaises(inputs.InputContractError): inputs.decode_frame(body, ref) def test_mixed_business_inputs_are_rejected_not_silently_filtered(self): original, manifest, _, _, _ = fixture() for frame_name, column, value in (("billing", "billing_month", "2026-07"), ("billing", "actual_revenue_recognition_date", ""), ("billing", "accumulate_revenue_to_org_tenant_id", "t2"), ("orders", "date_day", date(2026, 9, 2)), ("orders", "order_service_start_date", date(2026, 9, 2)), ("orders", "accumulate_revenue_to_org_tenant_id", "t2")): frames = {k: v.copy() for k, v in original.items()}; frames[frame_name].loc[0, column] = value with self.subTest(column=column), self.assertRaises(inputs.InputContractError): inputs.validate_frames(frames, manifest, date(2026, 9, 1)) frames = {k: v.copy() for k, v in original.items()} frames["orders"].drop(columns=["accumulated_actual_revenue_before_billing"], inplace=True) with self.assertRaises(inputs.InputContractError): inputs.validate_frames(frames, manifest, date(2026, 9, 1)) def output_job(binding="a" * 64, destination="fixture.outputs.table", job_id="fixture-job"): return SimpleNamespace(job_type="load", destination=destination, write_disposition="WRITE_EMPTY", job_id=job_id, labels={"portal_binding_a": binding[:32], "portal_binding_b": binding[32:]}, output_rows=1, error_result=None, result=Mock()) class GuardedOutputTest(unittest.TestCase): def setUp(self): self.frame = pd.DataFrame([{"bill_month": "2026-08", "amount": 1.0}]) self.client = Mock() self.client.get_table.return_value = SimpleNamespace(num_rows=1) def run_load(self, reconcile=False): return worker.guarded_load(self.client, self.frame, "fixture.outputs.table", "fixture-job", "a" * 64, "us-west2", reconcile) def test_replay_waits_original_job_and_does_not_resubmit(self): job = output_job(); self.client.get_job.return_value = job self.assertIs(self.run_load(), job) self.client.load_table_from_dataframe.assert_not_called() job.result.assert_called_once() def test_first_load_is_write_empty_and_names_job(self): self.client.get_job.side_effect = NotFound("fixture") self.client.get_table.side_effect = [NotFound("fixture"), SimpleNamespace(num_rows=1)] self.client.load_table_from_dataframe.return_value = output_job() self.run_load() kwargs = self.client.load_table_from_dataframe.call_args.kwargs self.assertEqual(kwargs["job_id"], "fixture-job") self.assertEqual(kwargs["job_config"].write_disposition, "WRITE_EMPTY") def test_racing_submission_reconciles_conflict_without_new_identity(self): self.client.get_job.side_effect = [NotFound("fixture"), output_job()] self.client.get_table.side_effect = [NotFound("fixture"), SimpleNamespace(num_rows=1)] self.client.load_table_from_dataframe.side_effect = Conflict("fixture") self.run_load() self.assertEqual(self.client.load_table_from_dataframe.call_count, 1) def test_unknown_job_and_existing_empty_table_never_overwritten(self): self.client.get_job.side_effect = NotFound("fixture") self.client.get_table.return_value = SimpleNamespace(num_rows=0) with self.assertRaises(ValueError): self.run_load() self.client.load_table_from_dataframe.assert_not_called() def test_reconcile_only_never_submits_missing_job(self): self.client.get_job.side_effect = NotFound("fixture") with self.assertRaises(ValueError): self.run_load(reconcile=True) self.client.load_table_from_dataframe.assert_not_called() self.client.get_table.assert_not_called() def test_wrong_job_binding_destination_and_disposition_fail(self): for key, value in (("destination", "fixture.outputs.other"), ("write_disposition", "WRITE_TRUNCATE"), ("labels", {}), ("job_id", "other-job"), ("output_rows", 2)): job = output_job(); setattr(job, key, value); self.client.get_job.return_value = job with self.subTest(key=key), self.assertRaises(ValueError): self.run_load() self.client.load_table_from_dataframe.assert_not_called() def test_worker_reads_no_mutable_source_and_binds_attempt(self): _, _, ref, _, store = fixture() args = SimpleNamespace(run_id=SCOPE["close_run_id"], bill_month="2026-08", mode="actual", attempt_id="billing-attempt-01", definition_hash=SCOPE["definition_hash"], input_manifest_uri=ref["uri"], input_manifest_generation=ref["generation"], input_manifest_sha256=ref["sha256"], reconcile_only=False) allocator = Mock(); allocator.process.return_value = (self.frame, self.frame) def noisy_allocator(**kwargs): print("private-order-diagnostic-must-not-escape") return self.frame, self.frame allocator.process.side_effect = noisy_allocator with patch.object(worker, "load_allocator", return_value=allocator), patch.object(worker, "guarded_load", return_value=output_job()) as load: capture = io.StringIO() with redirect_stdout(capture): result = worker.run(args, self.client, store, BUCKET, "fixture.outputs", "us-west2") self.assertNotIn("private-order", capture.getvalue()) self.assertEqual(result["verification"], "not_checked") self.assertEqual(result["attempt_id"], args.attempt_id) self.assertEqual(load.call_count, 2) first_destinations = [call.args[2] for call in load.call_args_list] load.reset_mock(); args.attempt_id = "billing-attempt-02" worker.run(args, self.client, store, BUCKET, "fixture.outputs", "us-west2") self.assertTrue(set(first_destinations).isdisjoint(call.args[2] for call in load.call_args_list)) self.client.query.assert_not_called() def test_partial_outputs_never_produce_complete_receipt(self): _, _, ref, _, store = fixture() args = SimpleNamespace(run_id=SCOPE["close_run_id"], bill_month="2026-08", mode="actual", attempt_id="billing-attempt-01", definition_hash=SCOPE["definition_hash"], input_manifest_uri=ref["uri"], input_manifest_generation=ref["generation"], input_manifest_sha256=ref["sha256"]) allocator = Mock(); allocator.process.return_value = (self.frame, self.frame) with patch.object(worker, "load_allocator", return_value=allocator), patch.object(worker, "guarded_load", side_effect=[output_job(), RuntimeError("fixture failed")]), self.assertRaises(RuntimeError): worker.run(args, self.client, store, BUCKET, "fixture.outputs", "us-west2") if __name__ == "__main__": unittest.main()