diff --git a/scripts/us_cdc/cdc500_state/README.md b/scripts/us_cdc/cdc500_state/README.md index 79f5abb00c..21b50c6e73 100644 --- a/scripts/us_cdc/cdc500_state/README.md +++ b/scripts/us_cdc/cdc500_state/README.md @@ -5,37 +5,127 @@ Author: Padma Gundapaneni @padma-g ## Table of Contents 1. [About the Dataset](#about-the-dataset) 1. [Overview](#overview) + 2. [Data Sources and Tables](#data-sources-and-tables) + 3. [Aggregation Methodology](#aggregation-methodology) 2. [About the Import](#about-the-import) 1. [Artifacts](#artifacts) 2. [Import Procedure](#import-procedure) + 3. [Production Rollout / Deployment SOP](#production-rollout--deployment-sop) + 4. [Troubleshooting](#troubleshooting) ## About the Dataset ### Overview -The state level data is aggragated from city level data coming from CDC500 import. +The state-level dataset calculates aggregated health indicator prevalence estimates for US states from the city-level CDC 500 Cities (`CDC500`) project data, weighted by corresponding Census ACS 5-Year population counts. -To get the data for this import run: -```bash -$ python3 process.py -``` +CDC PLACES (Population Level Analysis and Community Estimates) is the official successor program to the original CDC 500 Cities project, expanding health estimates across all US counties, places, and census tracts. In Data Commons, city-level data from both the historical CDC 500 Cities project (2016–2019) and ongoing CDC PLACES releases (2020–present) are ingested under provenance `dc/base/CDC500`. + +### Data Sources and Tables + +The aggregation script queries Google Cloud BigQuery graph tables in dataset `datcom-store.spanner_dc_graph_prod_snap_latest`: + +1. **`TimeSeries`**: + - **CDC 500 Series**: Identifies CDC 500 Statistical Variables (`JSON_VALUE(facet.provenance) = 'dc/base/CDC500'` and `variable_measured LIKE 'Percent_%'`) and extracts their measurement methods (`JSON_VALUE(facet.measurementMethod)`). It maps each percentage health metric to its appropriate denominator demographic cohort StatVar (e.g., `Count_Person_18OrMoreYears`, `Count_Person_18To64Years`, `Count_Person_65OrMoreYears`, `Count_Person`, etc.). + - **Census ACS 5-Year Series**: Filters and joins population counts from Census ACS 5-Year Survey (`JSON_VALUE(facet.provenance) = 'dc/base/CensusACS5YearSurvey'`). + +2. **`Observation`**: + - **Health Indicator Percentages**: Fetches city-level percentage values (`value AS percent`), observation dates (`date`), and city geoIds (`entity1 LIKE 'geoId/%'`) for CDC 500 StatVars. To prevent double-counting Oahu/Honolulu County population in years where 13-character Census Designated Places (`geoId/15XXXXX`) are published (2017–2022), `geoId/15003` is explicitly filtered in the `WHERE` clause to include only years `<= '2016'` and `'2017'` county-only indicators (excluding the four 2017 blood pressure and cholesterol indicators that also have 13-character CDP data). + - **City Cohort Populations**: Fetches city-level population counts (`value AS population`) for the corresponding demographic cohort StatVars. + +### Aggregation Methodology + +For each state, indicator StatVar, and observation date: +- City observations are joined with their corresponding demographic population counts. +- City geoIds (`geoId/XXXXXXX`) are mapped to state geoIds (`geoId/XX`) using the first 8 characters (including the prefix). +- State-level prevalence percentages are computed as a population-weighted average: + +$$\text{State Percent} = \frac{\sum (\text{City Population} \times \text{City Percent})}{\sum \text{City Population}}$$ + +The output measurement method is prefixed with `dcAggregate/` (e.g., `dcAggregate/CrudePrevalence`). + +*Note on Demographic Cohorts:* Several adult health indicators (such as `Percent_Person_Smoking`, `Percent_Person_Obesity`, `Percent_Person_WithDiabetes`) lack explicit age tokens in their DCID and intentionally fall back to `Count_Person` weighting to preserve bitwise parity with existing Data Commons baseline observations. + +#### Excluded Indicators + +The following age-bracketed cancer screening indicators are omitted from state-level aggregation: +- `Percent_Person_50To74Years_Female_ReceivedMammography` +- `Percent_Person_21To65Years_Female_ReceivedCervicalCancerScreening` +- `Percent_Person_21To65Years_Female_ReceivedPapSmearTest` +- `Percent_Person_50To75Years_ReceivedColorectalCancerScreening` + +**Rationale**: The Census ACS 5-Year Survey does not publish single composite population StatVars for these non-standard multi-year age brackets (`50To74Years`, `21To65Years`, `50To75Years`). In the legacy query, these indicators were already silently dropped during the inner join with ACS population series. Rather than applying arbitrary proxy weights, these indicators are explicitly excluded in the query CTE, preserving full parity with the legacy pipeline while preventing unnecessary processing. ## About the Import ### Artifacts #### Scripts -[`process.py`](https://github.com/datacommonsorg/data/blob/master//scripts/us_cdc/cdc500_state/process.py) +[`process.py`](process.py) + +#### Unit Tests +[`process_test.py`](process_test.py) + +#### tMCF Template +[`cdc500_state.tmcf`](cdc500_state.tmcf) +#### Validation Config +[`validation_config.json`](validation_config.json) -#### tMCFs -[`cdc500_state.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/cdc500_state/cdc500_state.tmcf) +#### Manifest +[`manifest.json`](manifest.json) ### Import Procedure -#### Data Download and Processing Steps +#### Prerequisites + +Ensure Google Cloud authentication is configured with access to BigQuery dataset `datcom-store.spanner_dc_graph_prod_snap_latest`: + +```bash +$ gcloud auth application-default login +``` + +#### Running the Script -To get the data for this import run: +To run the BigQuery aggregation and write the output CSV to the default output directory (`scripts/us_cdc/cdc500_state/CDC500State_Output/CDC500State_Output.csv` when run from repository root): ```bash -$ python3 process.py +$ python3 scripts/us_cdc/cdc500_state/process.py ``` + +To specify a custom output directory: + +```bash +$ python3 scripts/us_cdc/cdc500_state/process.py --output_dir=/path/to/output +``` + +To specify a GCP project and timeout (in seconds) for BigQuery jobs (if not set in ambient environment): + +```bash +$ python3 scripts/us_cdc/cdc500_state/process.py --project=my-gcp-project --timeout=600 +``` + +#### Running Unit Tests + +Run the test suite using Python's `unittest` runner from the repository root: + +```bash +$ python3 -m unittest scripts/us_cdc/cdc500_state/process_test.py +# or via module path: +$ python3 -m unittest scripts.us_cdc.cdc500_state.process_test +``` + +#### Automation + +This import is automated via Data Commons Import Automation and scheduled to run weekly via Cloud Batch every Monday at 01:00 UTC (`cron_schedule: "0 1 * * 1"` in `manifest.json`). + +- **Import Type**: Automated (weekly Cloud Batch cron: `0 1 * * 1`) +- **Production GCS Path**: `gs://datcom-prod-imports/scripts/us_cdc/cdc500_state/CDC500_States/` +- **Test GCS Path**: `gs://datcom-import-test/scripts/us_cdc/cdc500_state/CDC500_States/` + +### Production Rollout / Deployment SOP + +Adding `unit: Percent` and `scalingFactor: 100` to `cdc500_state.tmcf` standardizes observation series identity keys across the Knowledge Graph. Manual manipulation of `latest_version.txt` in production GCS is strictly prohibited per `b/548893664` and `CHK-9.3`; baseline promotion is handled strictly through validated automated pipeline executions. + +### Troubleshooting + +- **`ValueError: BigQuery query returned 0 rows`**: Verify ADC authentication (`gcloud auth application-default login`) and ensure read permissions on `datcom-store.spanner_dc_graph_prod_snap_latest`. Pass `--project=` if running outside the default project. diff --git a/scripts/us_cdc/cdc500_state/cdc500_state.tmcf b/scripts/us_cdc/cdc500_state/cdc500_state.tmcf index 73a294f9cf..e5246d284d 100644 --- a/scripts/us_cdc/cdc500_state/cdc500_state.tmcf +++ b/scripts/us_cdc/cdc500_state/cdc500_state.tmcf @@ -4,5 +4,7 @@ variableMeasured: C:CDC->statvar observationAbout: C:CDC->observation_about observationDate: C:CDC->observation_date value: C:CDC->percent +unit: Percent +scalingFactor: 100 measurementMethod: C:CDC->measurement_method -observationPeriod: "P1Y" \ No newline at end of file +observationPeriod: "P1Y" diff --git a/scripts/us_cdc/cdc500_state/manifest.json b/scripts/us_cdc/cdc500_state/manifest.json index eae459b1c6..a1a8369f20 100644 --- a/scripts/us_cdc/cdc500_state/manifest.json +++ b/scripts/us_cdc/cdc500_state/manifest.json @@ -17,14 +17,15 @@ } ], "resource_limits": { - "cpu": 8, - "memory": 64, - "disk": 100 + "cpu": 2, + "memory": 8, + "disk": 30 }, "source_files": [ - "CDC500State_Output/CDC500State_Output.csv" + "validation_config.json" ], - "cron_schedule": "0 1 * * 1" + "cron_schedule": "0 1 * * 1", + "validation_config_file": "validation_config.json" } ] -} \ No newline at end of file +} diff --git a/scripts/us_cdc/cdc500_state/process.py b/scripts/us_cdc/cdc500_state/process.py index 0115b96e43..f219b0d31e 100644 --- a/scripts/us_cdc/cdc500_state/process.py +++ b/scripts/us_cdc/cdc500_state/process.py @@ -11,75 +11,211 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +"""Processes CDC 500 cities data into aggregated state-level health indicators.""" import os + +from absl import app +from absl import flags from absl import logging from google.cloud import bigquery +_FLAGS = flags.FLAGS _MODULE_DIR = os.path.dirname(os.path.abspath(__file__)) -_OUTPUT_FILE_PATH = os.path.join(_MODULE_DIR + '/CDC500State_Output') -if not os.path.exists(_OUTPUT_FILE_PATH): - os.mkdir(_OUTPUT_FILE_PATH) - -query = """ -SELECT distinct * from( -SELECT - statvar, - SUBSTR(observation_about,0,8) as observation_about, - observation_date, - CONCAT('dcAggregate/',measurement_method) as measurement_method, - population_statvar, - SUM(CAST(pop_count AS FLOAT64))*100/SUM(CAST(population AS FLOAT64)) as percent -FROM -( +_DEFAULT_OUTPUT_DIR = os.path.join(_MODULE_DIR, 'CDC500State_Output') +OUTPUT_FILENAME = 'CDC500State_Output.csv' +DEFAULT_BQ_TIMEOUT_SECONDS = 600 + +flags.DEFINE_string('output_dir', _DEFAULT_OUTPUT_DIR, + 'Directory to write output CSV.') +flags.DEFINE_string( + 'project', None, + 'GCP project ID for BigQuery. Defaults to ambient environment if omitted.') +flags.DEFINE_integer('timeout', DEFAULT_BQ_TIMEOUT_SECONDS, + 'Timeout in seconds for BigQuery query and download.') + +QUERY = """ +WITH cdc_sv AS ( + SELECT + variable_measured AS cdc500, + CASE + WHEN REGEXP_CONTAINS( + variable_measured, r'65OrMoreYears.*Female' + ) THEN 'Count_Person_65OrMoreYears_Female' + WHEN REGEXP_CONTAINS( + variable_measured, r'65OrMoreYears.*Male' + ) THEN 'Count_Person_65OrMoreYears_Male' + WHEN variable_measured LIKE '%65OrMoreYears%' THEN 'Count_Person_65OrMoreYears' + WHEN variable_measured LIKE '%18To64Years%' THEN 'Count_Person_18To64Years' + WHEN variable_measured LIKE '%18OrMoreYears%' THEN 'Count_Person_18OrMoreYears' + ELSE 'Count_Person' + END AS pop_statvar + FROM `datcom-store.spanner_dc_graph_prod_snap_latest.TimeSeries` + WHERE JSON_VALUE(facet.provenance) = 'dc/base/CDC500' + AND variable_measured LIKE 'Percent_%' + -- Exclude 4 cancer screening StatVars with custom age brackets (50To74Years, + -- 21To65Years, 50-75Years) that lack matching single-variable population + -- denominators in CensusACS5YearSurvey (see b/530863116). + AND variable_measured NOT IN ( + 'Percent_Person_50To74Years_Female_ReceivedMammography', + 'Percent_Person_21To65Years_Female_ReceivedCervicalCancerScreening', + 'Percent_Person_21To65Years_Female_ReceivedPapSmearTest', + 'Percent_Person_50To75Years_ReceivedColorectalCancerScreening' + ) + GROUP BY cdc500, pop_statvar +), + +svo_percent AS ( + SELECT + O.variable_measured AS statvar, + O.entity1 AS observation_about, + O.date AS observation_date, + O.value AS percent, + JSON_VALUE(T.facet.measurementMethod) AS measurement_method, + cdc_sv.pop_statvar + FROM `datcom-store.spanner_dc_graph_prod_snap_latest.Observation` AS O + INNER JOIN `datcom-store.spanner_dc_graph_prod_snap_latest.TimeSeries` AS T + ON O.variable_measured = T.variable_measured + AND O.entity1 = T.entity1 + AND O.facet_id = T.facet_id + AND JSON_VALUE(T.facet.provenance) = 'dc/base/CDC500' + INNER JOIN cdc_sv + ON O.variable_measured = cdc_sv.cdc500 + WHERE O.entity1 LIKE 'geoId/%' + AND ( + -- Filter input entities to 13-char city/CDP DCIDs (geoId/XXXXXXX) before + -- rolling up to 8-char state DCIDs (geoId/NN) via SUBSTR(..., 1, 8), + -- preventing county (11-char) and tract (17-char) double-counting. + LENGTH(O.entity1) = 13 + OR ( + -- Hawaii has no incorporated 13-char municipal cities; Honolulu is a + -- consolidated city-county (geoId/15003, 11 chars) in earlier releases. + O.entity1 = 'geoId/15003' + AND ( + -- Scope geoId/15003 to <= 2017: from 2018+, Hawaii is reported under + -- 13-char CDPs (e.g. geoId/1571550) captured by LENGTH(O.entity1) = 13. + O.date <= '2016' + OR ( + O.date = '2017' + -- 2017 odd-year BRFSS BP/Cholesterol modules were backfilled under + -- 13-char geoId/1571550; exclude here to avoid double-counting. + AND NOT REGEXP_CONTAINS( + O.variable_measured, r'HighBloodPressure|Cholesterol' + ) + ) + ) + ) + ) + AND SAFE_CAST(O.value AS FLOAT64) IS NOT NULL + QUALIFY ROW_NUMBER() OVER ( + PARTITION BY + O.variable_measured, O.entity1, O.date, + JSON_VALUE(T.facet.measurementMethod) + ORDER BY O.last_update_timestamp DESC, O.facet_id DESC + ) = 1 +), + +svo_count AS ( SELECT - SVO1.variable_measured as statvar, - SVO1.observation_about as observation_about, - SVO1.observation_date as observation_date, - SVO1.value as percent, - SVO1.measurement_method as measurement_method, - SVO2.variable_measured as population_statvar, - SVO2.value as population, - CAST(SVO2.value AS FLOAT64) * CAST(SVO1.value AS FLOAT64) / 100 as pop_count - FROM `datcom-store.dc_kg_latest.StatVarObservation` as SVO1 - JOIN `datcom-store.dc_kg_latest.StatVarObservation` as SVO2 ON TRUE - JOIN ( - # Get the statvars and corresponding population statvar - # with ‘Percent_’ replaced with ‘Count_’ and - # dropping the non-age, non-gender constraints. - SELECT - SVO.variable_measured as CDC500, - CONCAT('Count_', REGEXP_SUBSTR(SVO.variable_measured, '(Person_.*ale|Person_.*Years|Person)')) as pop_statvar - FROM `datcom-store.dc_kg_latest.StatVarObservation` as SVO - WHERE - SVO.prov_id = 'dc/base/CDC500' - AND SVO.variable_measured like 'Percent_%' - GROUP BY CDC500, pop_statvar - ) AS CDC_SV ON TRUE - WHERE - SVO1.prov_id = 'dc/base/CDC500' - AND SVO1.variable_measured LIKE 'Percent%' - AND SVO1.observation_about = SVO2.observation_about - AND SVO1.observation_date = SVO2.observation_date - AND SVO1.variable_measured = CDC_SV.CDC500 - AND SVO2.variable_measured = CDC_SV.pop_statvar - AND SVO1.observation_about like "geoId/%" -) group by 1,2,3,4,5 + O.variable_measured AS population_statvar, + O.entity1 AS observation_about, + O.date AS observation_date, + O.value AS population + FROM `datcom-store.spanner_dc_graph_prod_snap_latest.Observation` AS O + INNER JOIN `datcom-store.spanner_dc_graph_prod_snap_latest.TimeSeries` AS T + ON O.variable_measured = T.variable_measured + AND O.entity1 = T.entity1 + AND O.facet_id = T.facet_id + AND JSON_VALUE(T.facet.provenance) = 'dc/base/CensusACS5YearSurvey' + INNER JOIN ( + SELECT DISTINCT pop_statvar + FROM cdc_sv + ) AS pop + ON O.variable_measured = pop.pop_statvar + WHERE O.entity1 LIKE 'geoId/%' + AND (LENGTH(O.entity1) = 13 OR O.entity1 = 'geoId/15003') + AND SAFE_CAST(O.value AS FLOAT64) IS NOT NULL + QUALIFY ROW_NUMBER() OVER ( + PARTITION BY O.variable_measured, O.entity1, O.date + ORDER BY O.last_update_timestamp DESC, O.facet_id DESC + ) = 1 ) + +SELECT + p.statvar, + SUBSTR(p.observation_about, 1, 8) AS observation_about, + p.observation_date, + COALESCE( + 'dcAggregate/' || NULLIF(TRIM(p.measurement_method), ''), + 'DataCommonsAggregate' + ) AS measurement_method, + SAFE_DIVIDE( + SUM(SAFE_CAST(c.population AS FLOAT64) * SAFE_CAST(p.percent AS FLOAT64)), + SUM(SAFE_CAST(c.population AS FLOAT64)) + ) AS percent +FROM svo_percent AS p +INNER JOIN svo_count AS c + ON p.observation_about = c.observation_about + AND p.observation_date = c.observation_date + AND p.pop_statvar = c.population_statvar +GROUP BY 1, 2, 3, 4 +HAVING percent IS NOT NULL """ -client = bigquery.Client() -try: - logging.info("Running the query") - query_job = client.query(query) -except Exception as e: - logging.fatal(f"Error faced while running the query {e}") -try: - logging.info("Converting to dataframe") - results = query_job.to_dataframe() -except Exception as e: - logging.info(f"Error faced while fetching results: {e}") - -logging.info("Writing output to CSV") -output_file = os.path.join(_OUTPUT_FILE_PATH + "/CDC500State_Output.csv") -results.to_csv(output_file, index=False) + +def run_process(client: bigquery.Client, + output_file: str, + timeout: int = DEFAULT_BQ_TIMEOUT_SECONDS) -> bool: + """Executes the BigQuery query and writes the resulting DataFrame to output_file.""" + logging.info("Running BigQuery aggregation query on project %s...", + client.project) + query_job = client.query(QUERY, timeout=timeout) + logging.info("BigQuery job started with ID: %s", query_job.job_id) + + logging.info("Fetching query results into dataframe...") + df = query_job.to_dataframe(timeout=timeout) + + if df.empty: + logging.error("BigQuery query returned 0 rows.") + raise ValueError("BigQuery query returned 0 rows.") + + output_dir = os.path.dirname(output_file) + if output_dir: + os.makedirs(output_dir, exist_ok=True) + logging.info("Writing %d rows to %s", len(df), output_file) + temp_file = output_file + ".tmp" + try: + df.to_csv(temp_file, index=False) + if not os.path.exists(temp_file) or os.path.getsize(temp_file) == 0: + logging.error("Output file %s was created empty or missing.", + temp_file) + raise ValueError( + f"Output file {temp_file} was created empty or missing.") + with open(temp_file, 'r', encoding='utf-8') as f: + written_lines = sum(1 for _ in f) - 1 + if written_lines != len(df): + logging.error("Written record count %d != expected %d for file %s", + written_lines, len(df), temp_file) + raise ValueError( + f"Output file {temp_file} truncated: wrote {written_lines}, " + f"expected {len(df)}") + os.replace(temp_file, output_file) + finally: + if os.path.exists(temp_file): + os.remove(temp_file) + return True + + +def main(argv): + """Main entry point for the CDC 500 state aggregation script.""" + del argv # Unused. + output_file = os.path.join(_FLAGS.output_dir, OUTPUT_FILENAME) + try: + client = bigquery.Client(project=_FLAGS.project) + run_process(client, output_file, timeout=_FLAGS.timeout) + except Exception as e: # pylint: disable=broad-exception-caught + logging.fatal("CDC 500 state aggregation failed: %s", e, exc_info=True) + + +if __name__ == '__main__': + app.run(main) diff --git a/scripts/us_cdc/cdc500_state/process_test.py b/scripts/us_cdc/cdc500_state/process_test.py new file mode 100644 index 0000000000..61ac20c1d4 --- /dev/null +++ b/scripts/us_cdc/cdc500_state/process_test.py @@ -0,0 +1,791 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Unit tests for CDC 500 State aggregation script.""" + +import os +import re +import tempfile +import unittest +from unittest import mock + +from absl import flags +from absl.testing import flagsaver +import duckdb +import pandas as pd + +from scripts.us_cdc.cdc500_state import process + +FLAGS = flags.FLAGS + + +def _prepare_duckdb_query(sql: str) -> str: + """Adapts BigQuery SQL syntax in process.QUERY for in-memory DuckDB execution.""" + adapted = sql.replace( + '`datcom-store.spanner_dc_graph_prod_snap_latest.TimeSeries`', + 'TimeSeries') + adapted = adapted.replace( + '`datcom-store.spanner_dc_graph_prod_snap_latest.Observation`', + 'Observation') + adapted = adapted.replace('SAFE_CAST(', 'TRY_CAST(') + adapted = adapted.replace('AS FLOAT64)', 'AS DOUBLE)') + adapted = adapted.replace("r'", "'") + adapted = re.sub(r'JSON_VALUE\(([^)]+)\)', r'\1', adapted) + return adapted + + +class CDC500StateProcessTest(unittest.TestCase): + """Unit tests for CDC 500 State aggregation processing.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + if not FLAGS.is_parsed(): + FLAGS(['test']) + + def test_query_constants(self): + """Verifies key SQL clauses and excluded variables in process.QUERY.""" + query = process.QUERY + self.assertIn("spanner_dc_graph_prod_snap_latest.TimeSeries", query) + self.assertIn("spanner_dc_graph_prod_snap_latest.Observation", query) + self.assertIn("JSON_VALUE(facet.provenance) = 'dc/base/CDC500'", query) + self.assertIn("JSON_VALUE(T.facet.provenance) = 'dc/base/CDC500'", + query) + self.assertIn( + "JSON_VALUE(T.facet.provenance) = 'dc/base/CensusACS5YearSurvey'", + query) + self.assertIn("JSON_VALUE(T.facet.measurementMethod)", query) + self.assertNotIn("WHERE provenance =", query) + self.assertNotIn("T.provenance", query) + self.assertNotIn("T.measurement_method", query) + self.assertIn("SAFE_DIVIDE", query) + self.assertIn("SAFE_CAST", query) + self.assertIn("SAFE_CAST(O.value AS FLOAT64) IS NOT NULL", query) + self.assertIn("HAVING percent IS NOT NULL", query) + self.assertIn("SUBSTR(p.observation_about, 1, 8)", query) + self.assertIn( + "AND (LENGTH(O.entity1) = 13 OR O.entity1 = 'geoId/15003')", query) + self.assertIn("O.entity1 = 'geoId/15003'", query) + self.assertIn("O.date <= '2016'", query) + self.assertIn("O.date = '2017'", query) + self.assertIn("HighBloodPressure|Cholesterol", query) + self.assertNotIn("svo_percent_dedup", query) + self.assertIn("REGEXP_CONTAINS", query) + self.assertIn("QUALIFY ROW_NUMBER() OVER", query) + self.assertIn("O.last_update_timestamp DESC, O.facet_id DESC", query) + self.assertIn("Percent_Person_50To74Years_Female_ReceivedMammography", + query) + self.assertIn( + "Percent_Person_21To65Years_Female_ReceivedCervicalCancerScreening", + query) + self.assertIn("Percent_Person_21To65Years_Female_ReceivedPapSmearTest", + query) + self.assertIn( + "Percent_Person_50To75Years_ReceivedColorectalCancerScreening", + query) + self.assertIn("'Count_Person_65OrMoreYears_Female'", query) + self.assertIn("'Count_Person_65OrMoreYears_Male'", query) + self.assertIn("'Count_Person_65OrMoreYears'", query) + self.assertIn("'Count_Person_18To64Years'", query) + self.assertIn("'Count_Person_18OrMoreYears'", query) + self.assertIn("'Count_Person'", query) + self.assertIn( + "SUM(SAFE_CAST(c.population AS FLOAT64) * " + "SAFE_CAST(p.percent AS FLOAT64))", query) + self.assertNotIn("p.pop_statvar AS population_statvar", query) + + def test_demographic_cohort_regex_mapping(self): + """Executes cdc_sv CTE from process.QUERY in DuckDB to verify cohort mapping.""" + con = duckdb.connect(':memory:') + con.create_function( + 'REGEXP_CONTAINS', lambda s, p: bool(re.search(p, s)) + if s and p else False, [str, str], bool) + + test_cases = [ + ('Percent_Person_65OrMoreYears_Female_CorePreventiveServices', + 'Count_Person_65OrMoreYears_Female'), + ('Percent_Person_65OrMoreYears_Male_CorePreventiveServices', + 'Count_Person_65OrMoreYears_Male'), + ('Percent_Person_65OrMoreYears_CorePreventiveServices', + 'Count_Person_65OrMoreYears'), + ('Percent_Person_18To64Years_HealthInsurance', + 'Count_Person_18To64Years'), + ('Percent_Person_18OrMoreYears_WithAnyDisability', + 'Count_Person_18OrMoreYears'), + ('Percent_Person_18OrMoreYears_WithHighBloodPressure', + 'Count_Person_18OrMoreYears'), + ('Percent_Person_WithArthritis', 'Count_Person'), + ('Percent_Person_WithHighCholesterol', 'Count_Person'), + ] + excluded_statvars = [ + 'Percent_Person_50To74Years_Female_ReceivedMammography', + 'Percent_Person_21To65Years_Female_ReceivedCervicalCancerScreening', + 'Percent_Person_21To65Years_Female_ReceivedPapSmearTest', + 'Percent_Person_50To75Years_ReceivedColorectalCancerScreening', + ] + + rows = [{ + 'variable_measured': sv, + 'facet': { + 'provenance': 'dc/base/CDC500' + } + } for sv, _ in test_cases] + rows.extend([{ + 'variable_measured': sv, + 'facet': { + 'provenance': 'dc/base/CDC500' + } + } for sv in excluded_statvars]) + ts_df = pd.DataFrame(rows) + con.register('TimeSeries', ts_df) + + # Extract cdc_sv CTE from process.QUERY and execute against TimeSeries + adapted_sql = _prepare_duckdb_query(process.QUERY) + cdc_sv_sql = adapted_sql.split('svo_percent AS (', + maxsplit=1)[0].rstrip().rstrip(',') + result_df = con.execute( + f"{cdc_sv_sql} SELECT cdc500, pop_statvar FROM cdc_sv").df() + + result_map = dict(zip(result_df['cdc500'], result_df['pop_statvar'])) + self.assertEqual(len(result_map), len(test_cases)) + for sv, expected_cohort in test_cases: + with self.subTest(statvar=sv): + self.assertEqual(result_map.get(sv), expected_cohort) + for excluded_sv in excluded_statvars: + with self.subTest(excluded=excluded_sv): + self.assertNotIn(excluded_sv, result_map) + + def test_population_weighted_average_calculation(self): + """Executes full process.QUERY in DuckDB to verify state weighted average.""" + con = duckdb.connect(':memory:') + con.create_function( + 'REGEXP_CONTAINS', lambda s, p: bool(re.search(p, s)) + if s and p else False, [str, str], bool) + con.create_function( + 'SAFE_DIVIDE', lambda a, b: float(a) / float(b) + if (a is not None and b) else None, [float, float], float) + + ts_df = pd.DataFrame([ + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/0644000', + 'facet_id': 'f1', + 'facet': { + 'provenance': 'dc/base/CDC500', + 'measurementMethod': 'CrudePrevalence' + } + }, + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/0666000', + 'facet_id': 'f1', + 'facet': { + 'provenance': 'dc/base/CDC500', + 'measurementMethod': 'CrudePrevalence' + } + }, + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/0667000', + 'facet_id': 'f1', + 'facet': { + 'provenance': 'dc/base/CDC500', + 'measurementMethod': 'CrudePrevalence' + } + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/0644000', + 'facet_id': 'f2', + 'facet': { + 'provenance': 'dc/base/CensusACS5YearSurvey', + 'measurementMethod': '' + } + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/0666000', + 'facet_id': 'f2', + 'facet': { + 'provenance': 'dc/base/CensusACS5YearSurvey', + 'measurementMethod': '' + } + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/0667000', + 'facet_id': 'f2', + 'facet': { + 'provenance': 'dc/base/CensusACS5YearSurvey', + 'measurementMethod': '' + } + }, + ]) + obs_df = pd.DataFrame([ + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/0644000', + 'date': '2022', + 'value': '20.0', + 'facet_id': 'f1', + 'last_update_timestamp': 100 + }, + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/0666000', + 'date': '2022', + 'value': '30.0', + 'facet_id': 'f1', + 'last_update_timestamp': 100 + }, + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/0667000', + 'date': '2022', + 'value': '40.0', + 'facet_id': 'f1', + 'last_update_timestamp': 100 + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/0644000', + 'date': '2022', + 'value': '10000', + 'facet_id': 'f2', + 'last_update_timestamp': 100 + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/0666000', + 'date': '2022', + 'value': '20000', + 'facet_id': 'f2', + 'last_update_timestamp': 100 + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/0667000', + 'date': '2022', + 'value': '70000', + 'facet_id': 'f2', + 'last_update_timestamp': 100 + }, + ]) + con.register('TimeSeries', ts_df) + con.register('Observation', obs_df) + + sql = _prepare_duckdb_query(process.QUERY) + result_df = con.execute(sql).df() + + # Expected: (10000*20 + 20000*30 + 70000*40) / 100000 = 36.0 + self.assertEqual(len(result_df), 1) + self.assertEqual(result_df['statvar'].iloc[0], + 'Percent_Person_WithArthritis') + self.assertEqual(result_df['observation_about'].iloc[0], 'geoId/06') + self.assertEqual(result_df['observation_date'].iloc[0], '2022') + self.assertEqual(result_df['measurement_method'].iloc[0], + 'dcAggregate/CrudePrevalence') + self.assertAlmostEqual(result_df['percent'].iloc[0], 36.0, places=4) + + def test_hawaii_honolulu_county_edge_cases(self): + """Tests Honolulu County (geoId/15003) filtering across 2016, 2017, 2018.""" + con = duckdb.connect(':memory:') + con.create_function( + 'REGEXP_CONTAINS', lambda s, p: bool(re.search(p, s)) + if s and p else False, [str, str], bool) + con.create_function( + 'SAFE_DIVIDE', lambda a, b: float(a) / float(b) + if (a is not None and b) else None, [float, float], float) + + # Test 2016 (county included), 2017 non-BP (county included), + # 2017 BP (county excluded, 13-char CDP used), + # 2018 non-BP (county excluded, 13-char CDP used), and empty measurementMethod. + ts_df = pd.DataFrame([ + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/15003', + 'facet_id': 'f1', + 'facet': { + 'provenance': 'dc/base/CDC500', + 'measurementMethod': '' + } + }, + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/1571550', + 'facet_id': 'f1', + 'facet': { + 'provenance': 'dc/base/CDC500', + 'measurementMethod': 'CrudePrevalence' + } + }, + { + 'variable_measured': 'Percent_Person_WithHighBloodPressure', + 'entity1': 'geoId/15003', + 'facet_id': 'f1', + 'facet': { + 'provenance': 'dc/base/CDC500', + 'measurementMethod': 'CrudePrevalence' + } + }, + { + 'variable_measured': 'Percent_Person_WithHighBloodPressure', + 'entity1': 'geoId/1571550', + 'facet_id': 'f1', + 'facet': { + 'provenance': 'dc/base/CDC500', + 'measurementMethod': 'CrudePrevalence' + } + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/15003', + 'facet_id': 'f2', + 'facet': { + 'provenance': 'dc/base/CensusACS5YearSurvey', + 'measurementMethod': '' + } + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/1571550', + 'facet_id': 'f2', + 'facet': { + 'provenance': 'dc/base/CensusACS5YearSurvey', + 'measurementMethod': '' + } + }, + ]) + obs_df = pd.DataFrame([ + # 2016: geoId/15003 should be INCLUDED (value 22.5) + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/15003', + 'date': '2016', + 'value': '22.5', + 'facet_id': 'f1', + 'last_update_timestamp': 100 + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/15003', + 'date': '2016', + 'value': '950000', + 'facet_id': 'f2', + 'last_update_timestamp': 100 + }, + # 2017 non-BP Arthritis: geoId/15003 (24.0) must be INCLUDED + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/15003', + 'date': '2017', + 'value': '24.0', + 'facet_id': 'f1', + 'last_update_timestamp': 100 + }, + # 2017 HighBloodPressure: geoId/15003 (99.0) must be EXCLUDED; + # 13-char CDP geoId/1571550 (31.2) must be INCLUDED. + { + 'variable_measured': 'Percent_Person_WithHighBloodPressure', + 'entity1': 'geoId/15003', + 'date': '2017', + 'value': '99.0', + 'facet_id': 'f1', + 'last_update_timestamp': 100 + }, + { + 'variable_measured': 'Percent_Person_WithHighBloodPressure', + 'entity1': 'geoId/1571550', + 'date': '2017', + 'value': '31.2', + 'facet_id': 'f1', + 'last_update_timestamp': 100 + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/15003', + 'date': '2017', + 'value': '950000', + 'facet_id': 'f2', + 'last_update_timestamp': 100 + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/1571550', + 'date': '2017', + 'value': '350000', + 'facet_id': 'f2', + 'last_update_timestamp': 100 + }, + # 2018 Arthritis: geoId/15003 (88.0) must be EXCLUDED; + # 13-char CDP geoId/1571550 (20.4) must be INCLUDED. + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/15003', + 'date': '2018', + 'value': '88.0', + 'facet_id': 'f1', + 'last_update_timestamp': 100 + }, + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/1571550', + 'date': '2018', + 'value': '20.4', + 'facet_id': 'f1', + 'last_update_timestamp': 100 + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/15003', + 'date': '2018', + 'value': '950000', + 'facet_id': 'f2', + 'last_update_timestamp': 100 + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/1571550', + 'date': '2018', + 'value': '350000', + 'facet_id': 'f2', + 'last_update_timestamp': 100 + }, + ]) + con.register('TimeSeries', ts_df) + con.register('Observation', obs_df) + + sql = _prepare_duckdb_query(process.QUERY) + result_df = con.execute(sql).df().sort_values( + by=['observation_date', 'statvar']).reset_index(drop=True) + + self.assertEqual(len(result_df), 4) + # 2016 row: geoId/15003 included, empty measurementMethod -> 'DataCommonsAggregate' + self.assertEqual(result_df['observation_about'].iloc[0], 'geoId/15') + self.assertEqual(result_df['observation_date'].iloc[0], '2016') + self.assertEqual(result_df['statvar'].iloc[0], + 'Percent_Person_WithArthritis') + self.assertEqual(result_df['measurement_method'].iloc[0], + 'DataCommonsAggregate') + self.assertAlmostEqual(result_df['percent'].iloc[0], 22.5, places=4) + # 2017 non-BP row: geoId/15003 included + self.assertEqual(result_df['observation_about'].iloc[1], 'geoId/15') + self.assertEqual(result_df['observation_date'].iloc[1], '2017') + self.assertEqual(result_df['statvar'].iloc[1], + 'Percent_Person_WithArthritis') + self.assertEqual(result_df['measurement_method'].iloc[1], + 'DataCommonsAggregate') + self.assertAlmostEqual(result_df['percent'].iloc[1], 24.0, places=4) + # 2017 BP row: geoId/15003 excluded, only geoId/1571550 (31.2) included + self.assertEqual(result_df['observation_about'].iloc[2], 'geoId/15') + self.assertEqual(result_df['observation_date'].iloc[2], '2017') + self.assertEqual(result_df['statvar'].iloc[2], + 'Percent_Person_WithHighBloodPressure') + self.assertEqual(result_df['measurement_method'].iloc[2], + 'dcAggregate/CrudePrevalence') + self.assertAlmostEqual(result_df['percent'].iloc[2], 31.2, places=4) + # 2018 Arthritis row: geoId/15003 excluded, only geoId/1571550 (20.4) included + self.assertEqual(result_df['observation_about'].iloc[3], 'geoId/15') + self.assertEqual(result_df['observation_date'].iloc[3], '2018') + self.assertEqual(result_df['statvar'].iloc[3], + 'Percent_Person_WithArthritis') + self.assertEqual(result_df['measurement_method'].iloc[3], + 'dcAggregate/CrudePrevalence') + self.assertAlmostEqual(result_df['percent'].iloc[3], 20.4, places=4) + + def test_qualify_row_number_deduplication(self): + """Tests QUALIFY ROW_NUMBER() deduplication by timestamp and facet_id.""" + con = duckdb.connect(':memory:') + con.create_function( + 'REGEXP_CONTAINS', lambda s, p: bool(re.search(p, s)) + if s and p else False, [str, str], bool) + con.create_function( + 'SAFE_DIVIDE', lambda a, b: float(a) / float(b) + if (a is not None and b) else None, [float, float], float) + + ts_df = pd.DataFrame([ + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/0644000', + 'facet_id': 'f_old', + 'facet': { + 'provenance': 'dc/base/CDC500', + 'measurementMethod': 'CrudePrevalence' + } + }, + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/0644000', + 'facet_id': 'f_new', + 'facet': { + 'provenance': 'dc/base/CDC500', + 'measurementMethod': 'CrudePrevalence' + } + }, + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/0666000', + 'facet_id': 'f_city2', + 'facet': { + 'provenance': 'dc/base/CDC500', + 'measurementMethod': 'CrudePrevalence' + } + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/0644000', + 'facet_id': 'f_low', + 'facet': { + 'provenance': 'dc/base/CensusACS5YearSurvey', + 'measurementMethod': '' + } + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/0644000', + 'facet_id': 'f_high', + 'facet': { + 'provenance': 'dc/base/CensusACS5YearSurvey', + 'measurementMethod': '' + } + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/0666000', + 'facet_id': 'f_city2_pop', + 'facet': { + 'provenance': 'dc/base/CensusACS5YearSurvey', + 'measurementMethod': '' + } + }, + ]) + obs_df = pd.DataFrame([ + # Competing percent observations for geoId/0644000: + # newer timestamp (200) must win over (100) + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/0644000', + 'date': '2022', + 'value': '99.0', + 'facet_id': 'f_old', + 'last_update_timestamp': 100 + }, + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/0644000', + 'date': '2022', + 'value': '25.0', + 'facet_id': 'f_new', + 'last_update_timestamp': 200 + }, + # Second city in CA (geoId/0666000) with percent 10.0 and pop 10,000 + { + 'variable_measured': 'Percent_Person_WithArthritis', + 'entity1': 'geoId/0666000', + 'date': '2022', + 'value': '10.0', + 'facet_id': 'f_city2', + 'last_update_timestamp': 200 + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/0666000', + 'date': '2022', + 'value': '10000', + 'facet_id': 'f_city2_pop', + 'last_update_timestamp': 200 + }, + # Competing count observations for geoId/0644000 with identical timestamp (200): + # higher facet_id ('f_low' > 'f_high' alphabetically: 'f_low' (50,000) wins) + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/0644000', + 'date': '2022', + 'value': '10000', + 'facet_id': 'f_high', + 'last_update_timestamp': 200 + }, + { + 'variable_measured': 'Count_Person', + 'entity1': 'geoId/0644000', + 'date': '2022', + 'value': '50000', + 'facet_id': 'f_low', + 'last_update_timestamp': 200 + }, + ]) + con.register('TimeSeries', ts_df) + con.register('Observation', obs_df) + + sql = _prepare_duckdb_query(process.QUERY) + result_df = con.execute(sql).df() + + self.assertEqual(len(result_df), 1) + # Choosing 'f_low' (50,000) over 'f_high' (10,000) yields weighted average: + # (25.0 * 50,000 + 10.0 * 10,000) / (50,000 + 10,000) = 1,350,000 / 60,000 = 22.5. + # If 'f_high' (10,000) had been chosen, average would be 17.5. + self.assertAlmostEqual(result_df['percent'].iloc[0], 22.5, places=4) + + def test_run_process_success(self): + """Tests successful query execution and atomic output writing.""" + mock_client = mock.MagicMock() + sample_data = pd.DataFrame({ + 'statvar': ['Percent_Person_18OrMoreYears_WithAnyDisability'], + 'observation_about': ['geoId/06'], + 'observation_date': ['2022'], + 'measurement_method': ['dcAggregate/CrudePrevalence'], + 'percent': [29.6479] + }) + mock_client.query.return_value.to_dataframe.return_value = sample_data + + with tempfile.TemporaryDirectory() as tmp_dir: + output_file = os.path.join(tmp_dir, process.OUTPUT_FILENAME) + result = process.run_process(mock_client, output_file) + self.assertTrue(result) + mock_client.query.assert_called_once_with( + process.QUERY, timeout=process.DEFAULT_BQ_TIMEOUT_SECONDS) + mock_client.query.return_value.to_dataframe.assert_called_once_with( + timeout=process.DEFAULT_BQ_TIMEOUT_SECONDS) + self.assertTrue(os.path.exists(output_file)) + self.assertFalse(os.path.exists(output_file + '.tmp')) + saved_df = pd.read_csv(output_file) + self.assertEqual(len(saved_df), 1) + self.assertEqual(saved_df['observation_about'].iloc[0], 'geoId/06') + + def test_run_process_empty_dataframe_raises_value_error(self): + """Tests that empty query results raise ValueError.""" + mock_client = mock.MagicMock() + mock_client.query.return_value.to_dataframe.return_value = pd.DataFrame( + ) + with tempfile.TemporaryDirectory() as tmp_dir: + output_file = os.path.join(tmp_dir, process.OUTPUT_FILENAME) + with self.assertRaisesRegex(ValueError, + "BigQuery query returned 0 rows"): + process.run_process(mock_client, output_file) + + def test_run_process_query_error(self): + """Tests propagation of BigQuery query execution errors.""" + mock_client = mock.MagicMock() + mock_client.query.side_effect = RuntimeError("BigQuery Access Denied") + with tempfile.TemporaryDirectory() as tmp_dir: + output_file = os.path.join(tmp_dir, process.OUTPUT_FILENAME) + with self.assertRaisesRegex(RuntimeError, "BigQuery Access Denied"): + process.run_process(mock_client, output_file) + + def test_run_process_dataframe_error(self): + """Tests propagation of query result download errors.""" + mock_client = mock.MagicMock() + mock_query_job = mock.MagicMock() + mock_query_job.to_dataframe.side_effect = RuntimeError( + "Failed to fetch dataframe") + mock_client.query.return_value = mock_query_job + with tempfile.TemporaryDirectory() as tmp_dir: + output_file = os.path.join(tmp_dir, process.OUTPUT_FILENAME) + with self.assertRaisesRegex(RuntimeError, + "Failed to fetch dataframe"): + process.run_process(mock_client, output_file) + + def test_run_process_empty_output_file_raises_value_error(self): + """Tests that creating an empty (0-byte) output file raises ValueError.""" + mock_client = mock.MagicMock() + mock_df = mock.MagicMock() + mock_df.empty = False + mock_df.__len__.return_value = 1 + + def fake_to_csv(filepath, *args, **kwargs): + del args, kwargs # Unused. + with open(filepath, 'w', encoding='utf-8'): + pass # Create 0-byte file + + mock_df.to_csv.side_effect = fake_to_csv + mock_client.query.return_value.to_dataframe.return_value = mock_df + with tempfile.TemporaryDirectory() as tmp_dir: + output_file = os.path.join(tmp_dir, process.OUTPUT_FILENAME) + with self.assertRaisesRegex(ValueError, + "was created empty or missing"): + process.run_process(mock_client, output_file) + self.assertFalse(os.path.exists(output_file)) + self.assertFalse(os.path.exists(output_file + '.tmp')) + + def test_run_process_truncated_output_file_raises_value_error(self): + """Tests that writing fewer lines than DataFrame length raises ValueError.""" + mock_client = mock.MagicMock() + mock_df = mock.MagicMock() + mock_df.empty = False + mock_df.__len__.return_value = 5 + + def fake_to_csv(filepath, *args, **kwargs): + del args, kwargs # Unused. + with open(filepath, 'w', encoding='utf-8') as f: + f.write("header1,header2\nrow1,val1\n" + ) # Only 1 data row, expected 5 + + mock_df.to_csv.side_effect = fake_to_csv + mock_client.query.return_value.to_dataframe.return_value = mock_df + with tempfile.TemporaryDirectory() as tmp_dir: + output_file = os.path.join(tmp_dir, process.OUTPUT_FILENAME) + with self.assertRaisesRegex(ValueError, + "truncated: wrote 1, expected 5"): + process.run_process(mock_client, output_file) + self.assertFalse(os.path.exists(output_file)) + self.assertFalse(os.path.exists(output_file + '.tmp')) + + @mock.patch('scripts.us_cdc.cdc500_state.process.run_process') + @mock.patch('google.cloud.bigquery.Client') + def test_main(self, mock_bq_client_cls, mock_run_process): + """Tests process.main flag parsing and client instantiation.""" + mock_client_instance = mock.MagicMock() + mock_bq_client_cls.return_value = mock_client_instance + with tempfile.TemporaryDirectory() as tmp_dir: + with flagsaver.flagsaver(output_dir=tmp_dir, + project='test-project', + timeout=300): + process.main([]) + expected_output_file = os.path.join(tmp_dir, + process.OUTPUT_FILENAME) + mock_bq_client_cls.assert_called_once_with( + project='test-project') + mock_run_process.assert_called_once_with(mock_client_instance, + expected_output_file, + timeout=300) + + @mock.patch('scripts.us_cdc.cdc500_state.process.logging.fatal') + @mock.patch('scripts.us_cdc.cdc500_state.process.run_process') + @mock.patch('google.cloud.bigquery.Client') + def test_main_error_logs_fatal(self, mock_bq_client_cls, mock_run_process, + mock_logging_fatal): + """Tests process.main catches unhandled exceptions and logs via logging.fatal.""" + del mock_bq_client_cls # Unused. + mock_run_process.side_effect = RuntimeError( + "BigQuery query returned 0 rows.") + with tempfile.TemporaryDirectory() as tmp_dir: + with flagsaver.flagsaver(output_dir=tmp_dir): + process.main([]) + mock_logging_fatal.assert_called_once() + + @mock.patch('scripts.us_cdc.cdc500_state.process.logging.fatal') + @mock.patch('google.cloud.bigquery.Client') + def test_main_client_init_error_logs_fatal(self, mock_bq_client_cls, + mock_logging_fatal): + """Tests process.main catches client init errors and logs via logging.fatal.""" + mock_bq_client_cls.side_effect = RuntimeError( + "DefaultCredentialsError: Could not automatically determine credentials." + ) + with tempfile.TemporaryDirectory() as tmp_dir: + with flagsaver.flagsaver(output_dir=tmp_dir): + process.main([]) + mock_logging_fatal.assert_called_once() + + +if __name__ == '__main__': + unittest.main() diff --git a/scripts/us_cdc/cdc500_state/validation_config.json b/scripts/us_cdc/cdc500_state/validation_config.json new file mode 100644 index 0000000000..cb58d4579a --- /dev/null +++ b/scripts/us_cdc/cdc500_state/validation_config.json @@ -0,0 +1,40 @@ +{ + "schema_version": "1.0", + "rules": [ + { + "rule_id": "check_max_value_percentage", + "description": "Checks that all percentage StatVars do not exceed 100%.", + "validator": "MAX_VALUE_CHECK", + "params": { + "maximum": 100 + }, + "scope": { + "variables": { + "contains_all": ["Percent"] + } + } + }, + { + "rule_id": "check_min_value_percentage", + "description": "Checks that all percentage StatVars are not below 0%.", + "validator": "MIN_VALUE_CHECK", + "params": { + "minimum": 0 + }, + "scope": { + "variables": { + "contains_all": ["Percent"] + } + } + }, + { + "rule_id": "check_num_places_state_count", + "description": "Checks observations cover 50 US states plus DC (50-52 entities).", + "validator": "NUM_PLACES_COUNT", + "params": { + "minimum": 50, + "maximum": 52 + } + } + ] +}