diff --git a/scripts/us_cdc/cdc500_state/README.md b/scripts/us_cdc/cdc500_state/README.md
index 79f5abb00c..e2b6da4d34 100644
--- a/scripts/us_cdc/cdc500_state/README.md
+++ b/scripts/us_cdc/cdc500_state/README.md
@@ -5,6 +5,8 @@ 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)
@@ -12,30 +14,90 @@ Author: Padma Gundapaneni @padma-g
 ## 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
-```
+### Data Sources and Tables
+
+The aggregation script queries Google Cloud BigQuery graph tables in dataset `datcom-store.spanner_dc_graph_prod_DEFAULT`:
+
+1. **`TimeSeries`**:
+   - **CDC 500 Series**: Identifies CDC 500 Statistical Variables (`provenance = 'dc/base/CDC500'` and `variable_measured LIKE 'Percent_%'`) and extracts their measurement methods (`measurement_method`). 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 (`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/%' AND LENGTH(entity1) = 13`) for CDC 500 StatVars.
+   - **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`).
+
+#### 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`). Rather than applying arbitrary proxy weights or risking silent row omission, these indicators are explicitly excluded from state aggregation.
 
 ## About the Import
 
 ### Artifacts
 
 #### Scripts
-[`process.py`](https://github.com/datacommonsorg/data/blob/master//scripts/us_cdc/cdc500_state/process.py)
+[`process.py`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/cdc500_state/process.py)
 
+#### Unit Tests
+[`process_test.py`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/cdc500_state/process_test.py)
 
-#### tMCFs
+#### tMCF Template
 [`cdc500_state.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/cdc500_state/cdc500_state.tmcf)
 
+#### Validation Config
+[`validation_config.json`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/cdc500_state/validation_config.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_DEFAULT`:
+
+```bash
+$ gcloud auth application-default login
+```
+
+#### Running the Script
+
+To run the BigQuery aggregation and write the output CSV to the default output directory (`CDC500State_Output/CDC500State_Output.csv`):
 
-To get the data for this import run:
+```bash
+$ python3 scripts/us_cdc/cdc500_state/process.py
+```
+
+To specify a custom output directory:
 
 ```bash
-$ python3 process.py
+$ python3 scripts/us_cdc/cdc500_state/process.py --output_dir=/path/to/output
 ```
+
+#### 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
+```
+
+#### 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`).
+
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..3a314ec2a5 100644
--- a/scripts/us_cdc/cdc500_state/manifest.json
+++ b/scripts/us_cdc/cdc500_state/manifest.json
@@ -21,10 +21,8 @@
                 "memory": 64,
                 "disk": 100
             },
-            "source_files": [
-                "CDC500State_Output/CDC500State_Output.csv"
-            ],
-            "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..1c1f67b567 100644
--- a/scripts/us_cdc/cdc500_state/process.py
+++ b/scripts/us_cdc/cdc500_state/process.py
@@ -1,4 +1,4 @@
-# Copyright 2021 Google LLC
+# 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.
@@ -11,75 +11,149 @@
 # 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)
+_DEFAULT_OUTPUT_DIR = os.path.join(_MODULE_DIR, 'CDC500State_Output')
 
-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 
-(
+flags.DEFINE_string('output_dir', _DEFAULT_OUTPUT_DIR,
+                    'Directory to write output CSV.')
+
+QUERY = """
+WITH cdc_sv 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
+    variable_measured AS cdc500,
+    CASE
+      WHEN REGEXP_CONTAINS(variable_measured, r'65OrMoreYears.*Female|Female.*65OrMoreYears') THEN 'Count_Person_65OrMoreYears_Female'
+      WHEN REGEXP_CONTAINS(variable_measured, r'65OrMoreYears.*Male|Male.*65OrMoreYears') 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_DEFAULT.TimeSeries`
+  WHERE provenance = 'dc/base/CDC500'
+    AND variable_measured LIKE 'Percent_%'
+    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,
+    T.measurement_method AS measurement_method,
+    cdc_sv.pop_statvar
+  FROM `datcom-store.spanner_dc_graph_prod_DEFAULT.Observation` AS O
+  INNER JOIN `datcom-store.spanner_dc_graph_prod_DEFAULT.TimeSeries` AS T
+    ON O.variable_measured = T.variable_measured
+    AND O.entity1 = T.entity1
+    AND O.facet_id = T.facet_id
+    AND T.provenance = 'dc/base/CDC500'
+    AND T.variable_measured LIKE 'Percent_%'
+  INNER JOIN cdc_sv
+    ON O.variable_measured = cdc_sv.cdc500
+  WHERE O.entity1 LIKE 'geoId/%'
+    AND LENGTH(O.entity1) = 13
+    AND O.variable_measured LIKE 'Percent_%'
+    AND O.variable_measured NOT IN (
+      'Percent_Person_50To74Years_Female_ReceivedMammography',
+      'Percent_Person_21To65Years_Female_ReceivedCervicalCancerScreening',
+      'Percent_Person_21To65Years_Female_ReceivedPapSmearTest',
+      'Percent_Person_50To75Years_ReceivedColorectalCancerScreening'
+    )
+  QUALIFY ROW_NUMBER() OVER (
+    PARTITION BY O.variable_measured, O.entity1, O.date, T.measurement_method
+    ORDER BY O.last_update_timestamp DESC
+  ) = 1
+),
+
+svo_count AS (
+  SELECT
+    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_DEFAULT.Observation` AS O
+  INNER JOIN `datcom-store.spanner_dc_graph_prod_DEFAULT.TimeSeries` AS T
+    ON O.variable_measured = T.variable_measured
+    AND O.entity1 = T.entity1
+    AND O.facet_id = T.facet_id
+    AND T.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
+  QUALIFY ROW_NUMBER() OVER (
+    PARTITION BY O.variable_measured, O.entity1, O.date
+    ORDER BY O.last_update_timestamp DESC
+  ) = 1
 )
+
+SELECT 
+  p.statvar, 
+  SUBSTR(p.observation_about, 1, 8) AS observation_about,
+  p.observation_date, 
+  CONCAT('dcAggregate/', p.measurement_method) AS measurement_method,
+  p.pop_statvar AS population_statvar,
+  SAFE_DIVIDE(
+    SUM(CAST(c.population AS FLOAT64) * CAST(p.percent AS FLOAT64)),
+    SUM(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, 5
 """
 
-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) -> bool:
+    """Executes the BigQuery query and writes the resulting DataFrame to output_file."""
+    logging.info("Running BigQuery aggregation query...")
+    query_job = client.query(QUERY)
+
+    logging.info("Fetching query results into dataframe...")
+    df = query_job.to_dataframe()
+
+    if df.empty:
+        raise RuntimeError("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"
+    df.to_csv(temp_file, index=False)
+    os.replace(temp_file, output_file)
+    return True
+
+
+def main(argv):
+    del argv  # Unused.
+    client = bigquery.Client()
+    output_file = os.path.join(_FLAGS.output_dir, 'CDC500State_Output.csv')
+    run_process(client, output_file)
+
+
+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..e5eecaa8cf
--- /dev/null
+++ b/scripts/us_cdc/cdc500_state/process_test.py
@@ -0,0 +1,173 @@
+# 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 tempfile
+import unittest
+from unittest import mock
+from absl import flags
+import pandas as pd
+
+from scripts.us_cdc.cdc500_state import process
+
+FLAGS = flags.FLAGS
+
+
+class CDC500StateProcessTest(unittest.TestCase):
+
+    def test_query_constants(self):
+        query = process.QUERY
+        self.assertIn("spanner_dc_graph_prod_DEFAULT.TimeSeries", query)
+        self.assertIn("spanner_dc_graph_prod_DEFAULT.Observation", query)
+        self.assertIn("dc/base/CDC500", query)
+        self.assertIn("dc/base/CensusACS5YearSurvey", query)
+        self.assertIn("SAFE_DIVIDE", query)
+        self.assertIn("SUBSTR(p.observation_about, 1, 8)", query)
+        self.assertIn("LENGTH(O.entity1) = 13", query)
+        self.assertIn("REGEXP_CONTAINS", query)
+        self.assertIn("QUALIFY ROW_NUMBER() OVER", query)
+        self.assertIn("O.last_update_timestamp 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)
+
+    def test_demographic_cohort_regex_mapping(self):
+        """Verifies that representative StatVars match the intended demographic regex rules."""
+        import re
+        female_pattern = r'65OrMoreYears.*Female|Female.*65OrMoreYears'
+        male_pattern = r'65OrMoreYears.*Male|Male.*65OrMoreYears'
+
+        self.assertIn(female_pattern, process.QUERY)
+        self.assertIn(male_pattern, process.QUERY)
+
+        # Helper mapping that mirrors the SQL CASE WHEN logic
+        def map_statvar(sv: str) -> str:
+            if re.search(female_pattern, sv):
+                return 'Count_Person_65OrMoreYears_Female'
+            elif re.search(male_pattern, sv):
+                return 'Count_Person_65OrMoreYears_Male'
+            elif '65OrMoreYears' in sv:
+                return 'Count_Person_65OrMoreYears'
+            elif '18To64Years' in sv:
+                return 'Count_Person_18To64Years'
+            elif '18OrMoreYears' in sv:
+                return 'Count_Person_18OrMoreYears'
+            else:
+                return 'Count_Person'
+
+        test_cases = [
+            ('Percent_Person_65OrMoreYears_Female_CorePreventiveServices', 'Count_Person_65OrMoreYears_Female'),
+            ('Percent_Person_Female_65OrMoreYears_CorePreventiveServices', 'Count_Person_65OrMoreYears_Female'),
+            ('Percent_Person_65OrMoreYears_Male_CorePreventiveServices', 'Count_Person_65OrMoreYears_Male'),
+            ('Percent_Person_Male_65OrMoreYears_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'),
+        ]
+
+        for sv, expected in test_cases:
+            with self.subTest(statvar=sv):
+                self.assertEqual(map_statvar(sv), expected)
+
+    def test_population_weighted_average_calculation(self):
+        """Verifies the population-weighted average calculation and city-to-state FIPS aggregation."""
+        # Simulated city-level records for California (geoId/06)
+        city_records = pd.DataFrame({
+            'city_geoid': ['geoId/0644000', 'geoId/0666000', 'geoId/0667000'],
+            'city_percent': [20.0, 30.0, 40.0],
+            'city_pop': [10000, 20000, 70000]
+        })
+        city_records['state_geoid'] = city_records['city_geoid'].str.slice(0, 8)
+        self.assertTrue((city_records['state_geoid'] == 'geoId/06').all())
+
+        # Formula: SUM(pop * percent) / SUM(pop)
+        total_weighted = (city_records['city_pop'] * city_records['city_percent']).sum()
+        total_pop = city_records['city_pop'].sum()
+        weighted_avg = total_weighted / total_pop
+
+        # Expected: (10000*20 + 20000*30 + 70000*40) / 100000 = (200000 + 600000 + 2800000) / 100000 = 36.0
+        self.assertEqual(total_pop, 100000)
+        self.assertAlmostEqual(weighted_avg, 36.0, places=4)
+
+    def test_run_process_success(self):
+        mock_client = mock.MagicMock()
+        sample_data = pd.DataFrame({
+            'statvar': ['Percent_Person_18OrMoreYears_WithAnyDisability'],
+            'observation_about': ['geoId/06'],
+            'observation_date': ['2022'],
+            'measurement_method': ['dcAggregate/CrudePrevalence'],
+            'population_statvar': ['Count_Person_18OrMoreYears'],
+            '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, 'CDC500State_Output.csv')
+            result = process.run_process(mock_client, output_file)
+            self.assertTrue(result)
+            mock_client.query.assert_called_once()
+            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_runtime_error(self):
+        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, 'CDC500State_Output.csv')
+            with self.assertRaises(RuntimeError):
+                process.run_process(mock_client, output_file)
+
+    def test_run_process_query_error(self):
+        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, 'CDC500State_Output.csv')
+            with self.assertRaises(RuntimeError):
+                process.run_process(mock_client, output_file)
+
+    def test_run_process_dataframe_error(self):
+        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, 'CDC500State_Output.csv')
+            with self.assertRaises(RuntimeError):
+                process.run_process(mock_client, output_file)
+
+    @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):
+        mock_client_instance = mock.MagicMock()
+        mock_bq_client_cls.return_value = mock_client_instance
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            FLAGS(['test_process', f'--output_dir={tmp_dir}'])
+            process.main([])
+            expected_output_file = os.path.join(tmp_dir, 'CDC500State_Output.csv')
+            mock_run_process.assert_called_once_with(mock_client_instance,
+                                                     expected_output_file)
+
+
+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..1a18f40b33
--- /dev/null
+++ b/scripts/us_cdc/cdc500_state/validation_config.json
@@ -0,0 +1,39 @@
+{
+    "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
+            }
+        },
+        {
+            "rule_id": "check_min_value_percentage",
+            "description": "Checks that all percentage StatVars are not below 0%.",
+            "validator": "MIN_VALUE_CHECK",
+            "params": {
+                "minimum": 0
+            }
+        },
+        {
+            "rule_id": "check_num_places_state_count",
+            "description": "Checks that state-level observations cover all 50-52 US state entities.",
+            "validator": "NUM_PLACES_COUNT",
+            "params": {
+                "minimum": 50,
+                "maximum": 52
+            }
+        },
+        {
+            "rule_id": "check_statvar_max_dates",
+            "description": "Verifies that MaxDate meets expected vintage freshness per StatVar cohort.",
+            "validator": "SQL_VALIDATOR",
+            "params": {
+                "query": "SELECT StatVar, TRY_CAST(MaxDate AS INTEGER) AS max_year FROM stats",
+                "condition": "CASE WHEN StatVar = 'Percent_Person_WithAllTeethLoss' THEN max_year >= 2016 WHEN StatVar LIKE '%CorePreventiveServices%' THEN max_year >= 2020 WHEN StatVar IN ('Percent_Person_WithHighBloodPressure', 'Percent_Person_18OrMoreYears_WithHighBloodPressure_ReceivedTakingBloodPressureMedication', 'Percent_Person_WithHighCholesterol', 'Percent_Person_ReceivedCholesterolScreening', 'Percent_Person_WithChronicKidneyDisease') THEN max_year >= 2021 ELSE max_year >= 2022 END"
+            }
+        }
+    ]
+}