diff --git a/docs/ysws-true-spend-airtable-cutover.md b/docs/ysws-true-spend-airtable-cutover.md new file mode 100644 index 0000000..61d0c54 --- /dev/null +++ b/docs/ysws-true-spend-airtable-cutover.md @@ -0,0 +1,61 @@ +# YSWS true-spend Airtable cutover + +The website and YSWS cost sync both read +`public_hcb_ysws_true_spend_analytics.ysws_spend_by_program`. +Do not calculate cost from HCB raised minus balance, add estimated postage, or +redivide by an individual Airtable row's hours. Several program versions may +share one HCB root; the website reports their combined spend/hours/rate. +The sync uses Airtable record IDs in `member_ids`, not program names, and checks +that the current HCB link still matches the snapshot's root. + +## One-time Airtable schema change + +Replace the existing columns in place. **Do not create additional fields.** +The previous two-field approval request is superseded and should not be approved. + +Keep `Total Spent From HCB Fund` as the writable spend column; it now receives +canonical **true spend**, not gross HCB outflow. + +Convert the existing **Cost Per Hour** field from formula to currency (2 decimal +places), preserving its name and field ID. The pipeline writes the mart's +`cost_per_weighted_hour` directly into this field. This keeps shared-root rates +identical to the website without changing existing project-hour rollups. +Before conversion, save the old formula for rollback: + +```text +IF(AND({Total Spend}, {Weighted–Total} > 0), {Total Spend} / {Weighted–Total} / 10) +``` + +Change the existing **Total Spend** formula (keep its field ID): + +```text +IF(({Total Spent From HCB Fund} & "") != "", {Total Spent From HCB Fund}, BLANK()) +``` + +The string check preserves a real zero without turning missing mappings into +zero spend. Keep the old postage inputs for reference; the formula must no +longer add the estimate on top of the reconciled ledger. + +## Deployment / verification + +1. Confirm browser access to edit the existing Airtable fields. +2. Deploy the code: the legacy gross-spend writer is removed. The new writer + refuses writes until Cost Per Hour is writable and Total Spend no longer adds postage. +3. Convert Cost Per Hour and update Total Spend in Airtable's UI (the API cannot change field types or edit formulas). +4. Materialize `ysws_programs_hcb_stats` and + `ysws_programs_true_spend_update_status`. No unrelated signup or project + processing needs to run for this cutover. +5. Reconcile every matched record's spend and displayed cost/hour against + its canonical mart row. Check zero spend, zero/missing hours, removed HCB + links, and shared roots. Unknown mappings clear old derived numbers to blank; + they never fall back to gross spend. + +The new sync is selected in the existing 15-minute Unified YSWS job. It reads +the latest successful true-spend snapshot, requiring a materialization within +36 hours. It does not trigger a heavy HCB mirror/dbt rebuild every 15 minutes. +It is deliberately outside `ysws_programs_prepared_for_update` and +`unified_ysws_db_processing_done`: those lead to the source warehouse refresh +that dbt reads, so making them depend on the true-spend output creates a cycle. + +Do not claim live reconciliation is complete until schema, formulas, production +materialization, and read-back are all verified. diff --git a/orpheus_engine/defs/unified_ysws_db/definitions.py b/orpheus_engine/defs/unified_ysws_db/definitions.py index 073a372..f38ddde 100644 --- a/orpheus_engine/defs/unified_ysws_db/definitions.py +++ b/orpheus_engine/defs/unified_ysws_db/definitions.py @@ -26,6 +26,7 @@ from ..airtable.definitions import airtable_config from ..geocoder.resources import GeocoderResource, GeocodingError from ..airtable.generated_ids import AirtableIDs +from .true_spend_sync import ysws_programs_hcb_stats, ysws_programs_true_spend_update_status from ..shared.address_utils import build_address_string_from_airtable_row from ..shared.daily_backups import ( ParquetBackupFile, @@ -751,219 +752,6 @@ def ysws_programs_sign_up_stats_candidates( ) -@asset( - group_name="unified_ysws_db_processing", - description="Prepares YSWS programs with HCB data by extracting HCB IDs from URLs", - compute_kind="data_preparation", - deps=[AssetKey(["airtable", "unified_ysws_projects_db", "ysws_programs"])], - required_resource_keys={"airtable"}, -) -def ysws_programs_hcb_candidates( - context: AssetExecutionContext, -) -> Output[pl.DataFrame]: - """ - Loads YSWS programs and extracts HCB IDs from HCB URLs. - - Returns: - DataFrame with id, hcb (URL), and hcb_id (extracted ID) - """ - log = context.log - airtable: AirtableResource = context.resources.airtable - - # Get the ysws_programs data from airtable - programs_df = airtable.get_all_records_as_polars( - context=context, - base_key="unified_ysws_projects_db", - table_key="ysws_programs", - ) - - log.info(f"Processing {programs_df.height} YSWS programs for HCB data") - - # Get the HCB field ID - hcb_field_id = UnifiedYSWS.ysws_programs.hcb - - if hcb_field_id not in programs_df.columns: - log.warning(f"HCB field {hcb_field_id} not found in programs data") - # Return empty DataFrame with correct schema - return Output( - pl.DataFrame(schema={ - "id": pl.Utf8, - "hcb": pl.Utf8, - "hcb_id": pl.Utf8 - }), - metadata={"num_programs": 0, "num_with_hcb": 0} - ) - - # Filter for programs that have HCB field set and extract HCB IDs - processed_df = programs_df.filter( - (pl.col(hcb_field_id).is_not_null()) & - (pl.col(hcb_field_id) != "") - ).with_columns([ - pl.col(hcb_field_id).map_elements( - lambda x: _extract_hcb_id_from_url(x) if x else "", - return_dtype=pl.Utf8 - ).alias("hcb_id") - ]).select([ - pl.col("id"), - pl.col(hcb_field_id).alias("hcb"), - pl.col("hcb_id") - ]).filter( - # Only include programs with valid HCB IDs (skip malformed URLs) - pl.col("hcb_id") != "" - ) - - log.info(f"Found {processed_df.height} programs with valid HCB URLs out of {programs_df.height} total programs") - if processed_df.height > 0: - log.info(f"Sample HCB data: {processed_df.head(3).to_dicts()}") - - # Generate preview metadata - if processed_df.height > 0: - try: - preview_metadata = MetadataValue.md(processed_df.head(10).to_pandas().to_markdown(index=False)) - except Exception: - preview_metadata = MetadataValue.text(str(processed_df.head(10))) - else: - preview_metadata = MetadataValue.text("No programs with valid HCB URLs found") - - return Output( - processed_df, - metadata={ - "num_programs": programs_df.height, - "num_with_hcb": processed_df.height, - "hcb_ids": processed_df["hcb_id"].to_list(), - "preview": preview_metadata - } - ) - - -@asset( - group_name="unified_ysws_db_processing", - description="Fetches HCB financial data and calculates total spent from HCB fund", - compute_kind="api_request", -) -def ysws_programs_hcb_stats( - context: AssetExecutionContext, - ysws_programs_hcb_candidates: pl.DataFrame, -) -> Output[pl.DataFrame]: - """ - Fetches HCB organization data and calculates total spent from HCB fund. - - Returns: - DataFrame with id and total_spent_from_hcb_fund field for Airtable updates - """ - log = context.log - input_df = ysws_programs_hcb_candidates - - total_spent_field_id = UnifiedYSWS.ysws_programs.total_spent_from_hcb_fund - - if input_df.height == 0: - log.info("No HCB candidates to process.") - return Output( - pl.DataFrame(schema={ - "id": pl.Utf8, - total_spent_field_id: pl.Float64, - }), - metadata={"num_processed": 0, "num_successful": 0, "num_failed": 0} - ) - - log.info(f"Processing HCB data for {input_df.height} programs") - - successful_records = [] - failures = [] - - for row in input_df.iter_rows(named=True): - program_id = row.get("id") - hcb_id = row.get("hcb_id") - hcb_url = row.get("hcb") - - if not hcb_id: - log.warning(f"No HCB ID for program {program_id} (url={hcb_url!r})") - failures.append({"id": program_id, "hcb_url": hcb_url, "reason": "No HCB ID extracted"}) - continue - - try: - api_url = f"https://hcb.hackclub.com/api/v3/organizations/{hcb_id}" - log.debug(f"Fetching HCB data for {hcb_id}: {api_url}") - - response = requests.get( - api_url, - headers={"Accept": "application/json"}, - timeout=30 - ) - - if response.status_code != 200: - try: - error_data = response.json() - error_message = error_data.get("message", f"HTTP {response.status_code}") - except Exception: - error_message = f"HTTP {response.status_code}" - - log.warning(f"HCB API returned {response.status_code} for {hcb_id}: {response.text}") - failures.append({"id": program_id, "hcb_url": hcb_url, "reason": error_message}) - continue - - data = response.json() - balances = data.get("balances", {}) - total_raised = balances.get("total_raised", 0) - balance_cents = balances.get("balance_cents", 0) - total_spent_dollars = (total_raised - balance_cents) / 100.0 - - successful_records.append({ - "id": program_id, - total_spent_field_id: total_spent_dollars, - }) - - log.info(f"Successfully processed {hcb_id}: total_raised={total_raised}, balance_cents={balance_cents}, total_spent=${total_spent_dollars:.2f}") - - except requests.exceptions.RequestException as e: - log.error(f"Request failed for HCB ID {hcb_id}: {e}") - failures.append({"id": program_id, "hcb_url": hcb_url, "reason": f"Request failed: {e}"}) - except (KeyError, ValueError, TypeError) as e: - log.error(f"Error parsing HCB data for {hcb_id}: {e}") - failures.append({"id": program_id, "hcb_url": hcb_url, "reason": f"Data parsing error: {e}"}) - except Exception as e: - log.error(f"Unexpected error processing HCB ID {hcb_id}: {e}") - failures.append({"id": program_id, "hcb_url": hcb_url, "reason": f"Unexpected error: {e}"}) - - if successful_records: - output_df = pl.DataFrame(successful_records) - else: - output_df = pl.DataFrame(schema={ - "id": pl.Utf8, - total_spent_field_id: pl.Float64, - }) - - successful_count = len(successful_records) - failed_count = len(failures) - log.info(f"HCB processing completed. Successful: {successful_count}, Failed: {failed_count}") - - # Generate preview metadata - if output_df.height > 0: - try: - preview_metadata = MetadataValue.md(output_df.head(10).to_pandas().to_markdown(index=False)) - except Exception: - preview_metadata = MetadataValue.text(str(output_df.head(10))) - else: - preview_metadata = MetadataValue.text("No HCB data processed successfully") - - metadata = { - "num_processed": input_df.height, - "num_successful": successful_count, - "num_failed": failed_count, - "success_rate": round((successful_count / max(input_df.height, 1)) * 100, 2), - "preview": preview_metadata, - } - if failures: - try: - metadata["failures"] = MetadataValue.md( - pl.DataFrame(failures).to_pandas().to_markdown(index=False) - ) - except Exception: - metadata["failures"] = MetadataValue.text(str(failures)) - - return Output(output_df, metadata=metadata) - - def _get_signup_analysis_data(search_terms: List[str]) -> pl.DataFrame: """ Get sign-up analysis data from hack_clubbers for the given search terms. @@ -1839,16 +1627,15 @@ def ysws_programs_sign_up_stats( @asset( group_name="unified_ysws_db_processing", - description="Prepares YSWS programs data for Airtable batch update by merging sign-up stats and HCB data.", + description="Prepares YSWS programs data for Airtable batch update from sign-up stats. True-spend updates run separately after dbt.", compute_kind="data_preparation", ) def ysws_programs_prepared_for_update( context: AssetExecutionContext, ysws_programs_sign_up_stats: pl.DataFrame, - ysws_programs_hcb_stats: pl.DataFrame, ) -> Output[pl.DataFrame]: """ - Merges sign-up stats and HCB data for YSWS programs and prepares for Airtable batch update. + Prepares sign-up stats for YSWS programs and prepares for Airtable batch update. Checks for conflicts: if multiple inputs provide different non-null values for the same field for a given program ID, an error is raised. @@ -1857,13 +1644,12 @@ def ysws_programs_prepared_for_update( log = context.log # --- 1. Collect and Filter Input DataFrames --- - dfs = [ysws_programs_sign_up_stats, ysws_programs_hcb_stats] + dfs = [ysws_programs_sign_up_stats] input_heights = { "sign_up_stats": ysws_programs_sign_up_stats.height, - "hcb_stats": ysws_programs_hcb_stats.height, } - log.info(f"Input counts - Sign-up stats: {input_heights['sign_up_stats']}, HCB stats: {input_heights['hcb_stats']}") + log.info(f"Input counts - Sign-up stats: {input_heights['sign_up_stats']}") dfs = [df for df in dfs if not df.is_empty()] @@ -1992,7 +1778,6 @@ def ysws_programs_prepared_for_update( metadata={ "num_records_prepared": final_df.height, "num_sign_up_stats_input": input_heights["sign_up_stats"], - "num_hcb_stats_input": input_heights["hcb_stats"], "update_columns": list(final_df.columns), "preview": preview_metadata } @@ -2001,7 +1786,7 @@ def ysws_programs_prepared_for_update( @asset( group_name="unified_ysws_db_processing", - description="Updates YSWS programs in Airtable with merged sign-up stats and HCB data.", + description="Updates YSWS programs in Airtable with sign-up stats (true-spend costs have a separate post-dbt writer).", required_resource_keys={"airtable"}, compute_kind="airtable_update", ) @@ -2010,7 +1795,7 @@ def ysws_programs_update_status( ysws_programs_prepared_for_update: pl.DataFrame, ) -> Output[None]: """ - Performs batch update of YSWS programs in Airtable with merged sign-up stats and HCB data. + Performs batch update of YSWS programs in Airtable with sign-up stats (true-spend costs have a separate post-dbt writer). """ log = context.log airtable: AirtableResource = context.resources.airtable @@ -2443,8 +2228,8 @@ def airtable_unified_ysws_db_daily_parquet_backup( approved_projects_mention_search_batch, ysws_programs_sign_up_stats_candidates, ysws_programs_sign_up_stats, - ysws_programs_hcb_candidates, ysws_programs_hcb_stats, + ysws_programs_true_spend_update_status, ysws_programs_prepared_for_update, ysws_programs_update_status, unified_ysws_db_processing_done, diff --git a/orpheus_engine/defs/unified_ysws_db/true_spend_sync.py b/orpheus_engine/defs/unified_ysws_db/true_spend_sync.py new file mode 100644 index 0000000..4c5d2ed --- /dev/null +++ b/orpheus_engine/defs/unified_ysws_db/true_spend_sync.py @@ -0,0 +1,134 @@ +"""Sync the website's canonical HCB-root costs, without redoing its accounting. + +Several Airtable program versions can share a root. Each gets the same root +spend, hours and rate shown on the website, NOT an invented per-version split. +Keep this writer downstream of dbt and outside the pre-warehouse processing +marker: feeding it back into that marker would create a Dagster asset cycle. +""" +from contextlib import closing +from datetime import datetime, timezone +import os +from urllib.parse import urlparse + +import polars as pl +import psycopg2 +from dagster import AssetKey, AssetExecutionContext, Output, asset + +from ..airtable.generated_ids import AirtableIDs + +PROGRAMS = AirtableIDs.unified_ysws_projects_db.ysws_programs +SPEND_FIELD = PROGRAMS.total_spent_from_hcb_fund +# Replace the existing columns in place; preserve their IDs and consumers. +RATE_FIELD = PROGRAMS.cost_per_hour +SYNC_FIELDS = (SPEND_FIELD, RATE_FIELD) +SOURCE_ASSET = AssetKey(["hcb_ysws_true_spend_analytics", "ysws_spend_by_program"]) +SOURCE_SQL = """ +SELECT root_slug, member_ids, true_spend_dollars, weighted_hours, + cost_per_weighted_hour +FROM public_hcb_ysws_true_spend_analytics.ysws_spend_by_program +WHERE is_ysws_program +""" + + +def _root_slug(url): + if not isinstance(url, str): + return None + parsed = urlparse(url) + if parsed.scheme != "https" or parsed.netloc != "hcb.hackclub.com": + return None + return parsed.path.strip("/").split("/")[0] or None + + +def build_true_spend_updates(programs: pl.DataFrame, costs: pl.DataFrame) -> pl.DataFrame: + """Exact record-ID + current HCB-link match; unknowns clear stale values. + + Zero spend is a real zero. Missing/zero hours keep the mart's NULL rate. + No gross-outflow fallback, postage estimate, or local rate calculation. + """ + if costs.is_empty(): + raise ValueError("True-spend source is empty; refusing to clear all Airtable costs") + if programs["id"].n_unique() != programs.height: + raise ValueError("Duplicate Airtable program IDs") + by_id = {} + for cost in costs.iter_rows(named=True): + for member in cost["member_ids"] or []: + if member in by_id: + raise ValueError("Program belongs to multiple canonical HCB roots") + by_id[member] = cost + rows = [] + for program in programs.iter_rows(named=True): + cost = by_id.get(program["id"]) + if cost is not None and _root_slug(program.get(PROGRAMS.hcb)) != cost["root_slug"]: + cost = None # Link changed since the warehouse snapshot: never use the old root. + rows.append({ + "id": program["id"], + SPEND_FIELD: float(cost["true_spend_dollars"]) if cost else None, + RATE_FIELD: float(cost["cost_per_weighted_hour"]) if cost and cost["cost_per_weighted_hour"] is not None else None, + }) + return pl.DataFrame(rows, schema={"id": pl.String, **{f: pl.Float64 for f in SYNC_FIELDS}}) + + +@asset( + group_name="ysws_true_spend_sync", + description="Read canonical true spend and weighted-hour cost from the same mart as the website.", + compute_kind="warehouse_query", + deps=[SOURCE_ASSET], + required_resource_keys={"airtable"}, +) +def ysws_programs_hcb_stats(context: AssetExecutionContext) -> Output[pl.DataFrame]: + event = context.instance.get_latest_materialization_event(SOURCE_ASSET) + if event is None or datetime.now(timezone.utc).timestamp() - event.timestamp > 36 * 3600: + raise ValueError("True-spend mart has no successful materialization within 36 hours") + programs = context.resources.airtable.get_all_records_as_polars( + context=context, base_key="unified_ysws_projects_db", table_key="ysws_programs", + ) + with closing(psycopg2.connect( + os.environ["WAREHOUSE_COOLIFY_URL"], connect_timeout=15, + options="-c statement_timeout=120000 -c lock_timeout=10000", + )) as conn: + costs = pl.read_database(SOURCE_SQL, conn) + updates = build_true_spend_updates(programs, costs) + return Output(updates, metadata={ + "programs": updates.height, + "matched": updates[SPEND_FIELD].is_not_null().sum(), + "missing_mapping": updates[SPEND_FIELD].is_null().sum(), + "source_materialized_at": datetime.fromtimestamp(event.timestamp, timezone.utc).isoformat(), + "source": "public_hcb_ysws_true_spend_analytics.ysws_spend_by_program", + }) + + +def write_true_spend_updates(table, updates: pl.DataFrame) -> int: + schema = table.schema(force=True) + by_id = {f.id: f for f in schema.fields} + for field_id in SYNC_FIELDS: + if getattr(by_id.get(field_id), "type", None) not in ("currency", "number"): + raise ValueError("Existing spend and Cost Per Hour columns must be writable currency fields before syncing") + field = by_id.get(PROGRAMS.total_spend) + actual = set(getattr(getattr(field, "options", None), "referenced_field_ids", None) or []) + if actual != {SPEND_FIELD}: + raise ValueError("Update existing Total Spend formula to use only true spend before syncing") + # Explicit nulls are intentional clears. The generic resource helper drops + # None, which would leave old rates after hours or HCB mappings disappear. + records = [{"id": row["id"], "fields": { + k: row[k] for k in SYNC_FIELDS + }} for row in updates.iter_rows(named=True)] + if not records: + return 0 + result = table.batch_update(records) + if len(result) != len(records): + raise RuntimeError("Incomplete true-spend Airtable update") + return len(result) + + +@asset( + group_name="ysws_true_spend_sync", + description="Update Airtable costs after true-spend dbt, separate from pre-dbt YSWS processing.", + compute_kind="airtable_update", + required_resource_keys={"airtable"}, +) +def ysws_programs_true_spend_update_status( + context: AssetExecutionContext, ysws_programs_hcb_stats: pl.DataFrame, +) -> Output[None]: + table = context.resources.airtable.get_table("unified_ysws_projects_db", "ysws_programs") + count = write_true_spend_updates(table, ysws_programs_hcb_stats) + return Output(None, metadata={"updates_successful": count}) diff --git a/orpheus_engine/schedules.py b/orpheus_engine/schedules.py index 9dd214f..4cc6a2b 100644 --- a/orpheus_engine/schedules.py +++ b/orpheus_engine/schedules.py @@ -85,6 +85,7 @@ def _evaluation_fn(context: dg.ScheduleEvaluationContext): # Unified YSWS refresh, daily Parquet backup, and warehouse assets UNIFIED_YSWS_SELECTION = ( + dg.AssetSelection.groups("ysws_true_spend_sync") | dg.AssetSelection.groups("airtable_unified_ysws_projects_db_refresh") | dg.AssetSelection.groups("dlt_airtable_unified_ysws") | dg.AssetSelection.assets("unified_ysws_ysws_programs_weighted_referral_count") diff --git a/orpheus_engine_tests/test_ysws_true_spend_sync.py b/orpheus_engine_tests/test_ysws_true_spend_sync.py new file mode 100644 index 0000000..f9b8d15 --- /dev/null +++ b/orpheus_engine_tests/test_ysws_true_spend_sync.py @@ -0,0 +1,125 @@ +"""Synthetic fixtures only: cost sync must preserve the website's accounting.""" +from types import SimpleNamespace + +import polars as pl +import pytest + +from orpheus_engine.defs.unified_ysws_db.true_spend_sync import ( + PROGRAMS, SPEND_FIELD, RATE_FIELD, SOURCE_ASSET, + build_true_spend_updates, write_true_spend_updates, ysws_programs_hcb_stats, +) + + +def costs(**overrides): + row = dict(root_slug="synthetic-root", member_ids=["record-a", "record-b"], + true_spend_dollars=120.25, weighted_hours=30.0, cost_per_weighted_hour=4.01) + row.update(overrides) + return pl.DataFrame([row]) + + +def programs(*ids, url="https://hcb.hackclub.com/synthetic-root"): + return pl.DataFrame({"id": ids, PROGRAMS.hcb: [url] * len(ids)}) + + +def test_shared_root_copies_canonical_rate_without_reallocating_or_recalculating(): + rows = build_true_spend_updates(programs("record-a", "record-b"), costs()).to_dicts() + assert len(rows) == 2 + for r in rows: + assert r[SPEND_FIELD] == 120.25 + assert r[RATE_FIELD] == 4.01 # Direct rounded mart value, not 120.25 / 30. + + +def test_unknown_program_is_null_not_zero_or_old_gross_spend(): + r = build_true_spend_updates(programs("unmatched"), costs()).row(0, named=True) + assert all(r[f] is None for f in (SPEND_FIELD, RATE_FIELD)) + + +@pytest.mark.parametrize('url', [None, '', 'https://example.invalid/synthetic-root', + 'https://hcb.hackclub.com/changed-root']) +def test_removed_or_remapped_hcb_link_clears_old_values(url): + r = build_true_spend_updates(programs("record-a", url=url), costs()).row(0, named=True) + assert r[SPEND_FIELD] is None + assert r[RATE_FIELD] is None + + +def test_transaction_url_matches_root_and_record_id_not_program_name(): + r = build_true_spend_updates(programs("record-a", url="https://hcb.hackclub.com/synthetic-root/transactions?x=1"), costs()) + assert r[SPEND_FIELD][0] == 120.25 + + +def test_zero_spend_and_null_rate_are_preserved(): + r = build_true_spend_updates(programs("record-a"), costs(true_spend_dollars=0, weighted_hours=0, cost_per_weighted_hour=None)) + assert r[SPEND_FIELD][0] == 0 + assert r[RATE_FIELD][0] is None + + +def test_empty_source_fails_without_clearing_everything(): + with pytest.raises(ValueError, match="source is empty"): + build_true_spend_updates(programs("record-a"), costs().head(0)) + + +def test_duplicate_source_mapping_fails_instead_of_double_counting(): + with pytest.raises(ValueError, match="multiple canonical"): + build_true_spend_updates(programs("record-a"), pl.concat([costs(), costs()])) + + +def test_duplicate_airtable_ids_fail(): + with pytest.raises(ValueError, match="Duplicate Airtable"): + build_true_spend_updates(programs("record-a", "record-a"), costs()) + + +def test_dependency_is_canonical_mart(): + assert SOURCE_ASSET in ysws_programs_hcb_stats.asset_deps[ysws_programs_hcb_stats.key] + + +class FakeTable: + def __init__(self, missing=False): + self.records = None + self.missing = missing + def schema(self, force=False): + assert force + return SimpleNamespace(fields=[] if self.missing else [ + SimpleNamespace(name="Total Spent From HCB Fund", id=SPEND_FIELD, type="currency"), + SimpleNamespace(name="Cost Per Hour", id=RATE_FIELD, type="currency"), + SimpleNamespace(name="Total Spend", id=PROGRAMS.total_spend, + options=SimpleNamespace(referenced_field_ids=[SPEND_FIELD])), + ]) + def batch_update(self, records): + self.records = records + return records + + +def test_writer_preserves_explicit_nulls_and_only_writes_owned_fields(): + table = FakeTable() + updates = build_true_spend_updates(programs("unmatched"), costs()) + assert write_true_spend_updates(table, updates) == 1 + assert table.records == [{"id": "unmatched", "fields": { + SPEND_FIELD: None, RATE_FIELD: None, + }}] + + +def test_missing_schema_blocks_writes(): + table = FakeTable(missing=True) + with pytest.raises(ValueError, match="writable currency"): + write_true_spend_updates(table, build_true_spend_updates(programs("record-a"), costs())) + assert table.records is None + + +def test_legacy_formulas_block_cutover_before_any_writes(): + table = FakeTable() + schema = table.schema(force=True) + schema.fields[-1].options.referenced_field_ids.append("synthetic-postage-field") + table.schema = lambda **kwargs: schema + with pytest.raises(ValueError, match="formula"): + write_true_spend_updates(table, build_true_spend_updates(programs("record-a"), costs())) + assert table.records is None + + +def test_old_cost_formula_blocks_writes_until_in_place_conversion(): + table = FakeTable() + schema = table.schema(force=True) + schema.fields[1].type = "formula" + table.schema = lambda **kwargs: schema + with pytest.raises(ValueError, match="writable currency"): + write_true_spend_updates(table, build_true_spend_updates(programs("record-a"), costs())) + assert table.records is None