fix: add reports_bucket_name as a parameter used in gtfs validator prerelease generation

.github/workflows/api-deployer.yml

@@ -207,7 +207,7 @@ jobs:
           echo "POSTGRES_DB=${{ env.local_postgres_db }}" >> config/.env.local
           echo "POSTGRES_PORT=${{ env.local_postgres_port }}" >> config/.env.local
           echo "POSTGRES_HOST=localhost" >> config/.env.local
-          echo "ENV=dev" >> config/.env.local
+          echo "ENVIRONMENT=dev" >> config/.env.local
 
       # db models were generated and uploaded in api-build-test job above.
       - uses: actions/download-artifact@v4

.github/workflows/db-update-content.yml

@@ -95,7 +95,7 @@ jobs:
         echo "FEEDS_DATABASE_URL=postgresql://${{ secrets.DB_USER_NAME }}:${{ secrets.DB_USER_PASSWORD }}@localhost:5432/${{ inputs.DB_NAME }}" >> config/.env.local
         echo "POSTGRES_PORT=5432" >> config/.env.local
         echo "POSTGRES_HOST=localhost" >> config/.env.local
-        echo "ENV=${{ inputs.ENVIRONMENT }}" >> config/.env.local
+        echo "ENVIRONMENT=${{ inputs.ENVIRONMENT }}" >> config/.env.local
         cat config/.env.local
 
     - name: Load secrets from 1Password

api/src/scripts/load_dataset_on_create.py

@@ -29,7 +29,7 @@ def get_pubsub_client():
 
 
 def get_topic_path():
-    env = os.getenv("ENV", "dev")
+    env = os.getenv("ENVIRONMENT", "dev")
     pubsub_topic_name = f"datasets-batch-topic-{env}"
     project_id = f"mobility-feeds-{env}"  # Cannot use GOOGLE_CLOUD_PROJECT because it points to QA for DEV
     return get_pubsub_client().topic_path(project_id, pubsub_topic_name)

api/src/scripts/populate_db_gbfs.py

@@ -152,14 +152,14 @@ def trigger_downstream_tasks(self):
             f"Triggering downstream tasks for {len(self.added_feeds)} newly added feed(s): "
             f"{', '.join(f['stable_id'] for f in self.added_feeds)}"
         )
-        if os.getenv("ENV", "local") == "local":
+        if os.getenv("ENVIRONMENT", "local") == "local":
             self.logger.info("Skipping downstream tasks in local environment.")
             return
         if not self.added_feeds:
             self.logger.info("No feeds to trigger downstream tasks for.")
             return
 
-        env = os.getenv("ENV", "dev")
+        env = os.getenv("ENVIRONMENT", "dev")
         project_id = f"mobility-feeds-{env}"
         publisher = pubsub_v1.PublisherClient()
         topic_path = publisher.topic_path(project_id, GBFS_PUBSUB_TOPIC_NAME)

api/src/scripts/populate_db_gtfs.py

@@ -283,9 +283,9 @@ def trigger_downstream_tasks(self):
             f"{','.join([feed.stable_id for feed in self.added_gtfs_feeds] if self.added_gtfs_feeds else [])}"
         )
 
-        env = os.getenv("ENV")
-        self.logger.info(f"ENV = {env}")
-        if os.getenv("ENV", "local") != "local":
+        env = os.getenv("ENVIRONMENT")
+        self.logger.info(f"ENVIRONMENT = {env}")
+        if os.getenv("ENVIRONMENT", "local") != "local":
             # Publishes the new feeds to the Pub/Sub topic to download the datasets
             publish_all(self.added_gtfs_feeds)
 

config/.env.local

@@ -6,7 +6,7 @@ POSTGRES_PORT=5432
 POSTGRES_TEST_PORT=54320
 PGUSER=postgres
 POSTGRES_HOST=localhost
-ENV=local
+ENVIRONMENT=local
 SCHEMA_SPY_DOC=schemaspy-dev
 FEEDS_DATABASE_URL=postgresql://postgres:postgres@localhost:5432/MobilityDatabase
 FEEDS_DATABASE_URL_TEST=postgresql://postgres:postgres@localhost:54320/MobilityDatabaseTest

functions-python/backfill_dataset_service_date_range/.env.rename_me

@@ -1,3 +1,3 @@
 # Environment variables for the validation report information extraction to run locally
 export FEEDS_DATABASE_URL=${{FEEDS_DATABASE_URL}}
-export ENV=${{ENV}}
+export ENVIRONMENT=${{ENVIRONMENT}}

functions-python/backfill_dataset_service_date_range/src/main.py

@@ -25,7 +25,7 @@
 from google.cloud import storage
 import requests
 
-env = os.getenv("ENV", "dev").lower()
+env = os.getenv("ENVIRONMENT", "dev").lower()
 bucket_name = f"mobilitydata-datasets-{env}"
 
 init_logger()

functions-python/helpers/validation_report/validation_report_update.py

@@ -7,7 +7,7 @@
 from google.cloud.workflows import executions_v1
 from google.cloud.workflows.executions_v1 import Execution
 
-env = os.getenv("ENV", "dev").lower()
+env = os.getenv("ENVIRONMENT", "dev").lower()
 bucket_name = f"mobilitydata-datasets-{env}"
 
 

functions-python/tasks_executor/src/tasks/data_import/transportdatagouv/import_tdg_feeds.py

@@ -804,7 +804,7 @@ def commit_changes(
         logger.info("Commit after processing items (count=%d)", total_processed)
         db_session.commit()
         execution_id = str(uuid.uuid4())
-        if os.getenv("ENV", "").lower() == "local":
+        if os.getenv("ENVIRONMENT", "").lower() == "local":
             return
         for feed in feeds_to_publish:
             trigger_dataset_download(feed, execution_id)

functions-python/tasks_executor/src/tasks/validation_reports/README.md

@@ -77,7 +77,8 @@ that were already triggered (tracked in `task_execution_log`).
     "filter_statuses": ["active"],
     "filter_op_statuses": ["published"],
     "force_update": false,
-    "limit": 10
+    "limit": 10,
+    "reports_bucket_name": "stg-gtfs-validator-results"
 }
 ```
 
@@ -91,6 +92,7 @@ that were already triggered (tracked in `task_execution_log`).
 | `filter_op_statuses` | list[str] | `["published"]` | Filter feeds by operational status. Accepted values: `"published"`, `"unpublished"`, `"wip"` |
 | `force_update` | bool | `false` | Re-trigger even when a current report already exists |
 | `limit` | int | `null` | Cap the number of workflows triggered per call — useful for end-to-end testing |
+| `reports_bucket_name` | string | env-derived | Override the GCS bucket where validator results are stored. Use when running in prod but pointing to the staging validator (e.g. `"stg-gtfs-validator-results"`) |
 
 ---
 

functions-python/tasks_executor/src/tasks/validation_reports/rebuild_missing_validation_reports.py

@@ -68,6 +68,9 @@ def rebuild_missing_validation_reports_handler(payload) -> dict:
             Default: False
         "force_update": bool,        # [optional] Re-trigger even if a report already exists. Default: False
         "limit": int,                # [optional] Max datasets to trigger per call (for testing). Default: unlimited
+        "reports_bucket_name": str,  # [optional] Override the GCS bucket for validator results.
+                                     #   Defaults to the env-derived bucket (prod or staging).
+                                     #   Use when running in prod but pointing to the staging validator.
     }
     """
     (
@@ -80,6 +83,7 @@ def rebuild_missing_validation_reports_handler(payload) -> dict:
         bypass_db_update,
         force_update,
         limit,
+        reports_bucket_name,
     ) = get_parameters(payload)
 
     return rebuild_missing_validation_reports(
@@ -92,6 +96,7 @@ def rebuild_missing_validation_reports_handler(payload) -> dict:
         prod_env=prod_env,
         force_update=force_update,
         limit=limit,
+        reports_bucket_name=reports_bucket_name,
     )
 
 
@@ -106,6 +111,7 @@ def rebuild_missing_validation_reports(
     prod_env: bool = False,
     force_update: bool = False,
     limit: Optional[int] = None,
+    reports_bucket_name: Optional[str] = None,
     db_session: Session | None = None,
 ) -> dict:
     """
@@ -124,6 +130,8 @@ def rebuild_missing_validation_reports(
         prod_env: True if targeting the production environment. Default: False
         force_update: Re-trigger even if a report already exists. Default: False
         limit: Max datasets to trigger per call (for end-to-end testing). Default: unlimited
+        reports_bucket_name: Override the GCS bucket for validator results.
+            Defaults to the env-derived bucket (prod or staging).
         db_session: DB session (injected by @with_db_session)
     """
     validator_version = _get_validator_version(validator_endpoint)
@@ -192,7 +200,8 @@ def rebuild_missing_validation_reports(
             datasets_to_trigger,
             validator_endpoint=validator_endpoint,
             bypass_db_update=bypass_db_update,
-            reports_bucket_name=get_gtfs_validator_results_bucket(prod_env),
+            reports_bucket_name=reports_bucket_name
+            or get_gtfs_validator_results_bucket(prod_env),
             tracker=tracker,
         )
         total_triggered = len(triggered_ids)
@@ -222,6 +231,8 @@ def rebuild_missing_validation_reports(
             "prod_env": prod_env,
             "force_update": force_update,
             "limit": limit,
+            "reports_bucket_name": reports_bucket_name
+            or get_gtfs_validator_results_bucket(prod_env),
         },
     }
     logging.info(result)
@@ -322,8 +333,9 @@ def get_parameters(payload):
     Args:
         payload (dict): Task payload dict.
     Returns:
-        Tuple of (dry_run, filter_after_in_days, filter_statuses, prod_env,
-                  validator_endpoint, force_update, limit)
+        Tuple of (dry_run, filter_after_in_days, filter_statuses, filter_op_statuses,
+                  prod_env, validator_endpoint, bypass_db_update, force_update, limit,
+                  reports_bucket_name)
     """
     prod_env = os.getenv("ENVIRONMENT", "").lower() == "prod"
     default_endpoint = get_gtfs_validator_url(prod_env)
@@ -363,6 +375,8 @@ def get_parameters(payload):
     if limit is not None:
         limit = int(limit)
 
+    reports_bucket_name = payload.get("reports_bucket_name", None)
+
     return (
         dry_run,
         filter_after_in_days,
@@ -373,4 +387,5 @@ def get_parameters(payload):
         bypass_db_update,
         force_update,
         limit,
+        reports_bucket_name,
     )

functions-python/tasks_executor/tests/tasks/data_import/test_tdg_import.py

@@ -207,7 +207,7 @@ def test_import_creates_gtfs_and_rt(self, db_session: Session):
             mock_trigger,
         ), patch.dict(
             os.environ,
-            {"COMMIT_BATCH_SIZE": "1", "ENV": "test"},
+            {"COMMIT_BATCH_SIZE": "1", "ENVIRONMENT": "test"},
             clear=False,
         ):
             result = import_tdg_handler({"dry_run": False})

functions-python/tasks_executor/tests/tasks/validation_reports/test_rebuild_missing_validation_reports.py

@@ -39,6 +39,7 @@ def test_defaults(self):
             bypass_db_update,
             force_update,
             limit,
+            reports_bucket_name,
         ) = get_parameters({})
         self.assertTrue(dry_run)
         self.assertIsNone(filter_after_in_days)
@@ -49,6 +50,7 @@ def test_defaults(self):
         self.assertFalse(bypass_db_update)
         self.assertFalse(force_update)
         self.assertIsNone(limit)
+        self.assertIsNone(reports_bucket_name)
 
     def test_all_params(self):
         payload = {
@@ -60,6 +62,7 @@ def test_all_params(self):
             "bypass_db_update": True,
             "force_update": True,
             "limit": 10,
+            "reports_bucket_name": "my-custom-bucket",
         }
         (
             dry_run,
@@ -71,6 +74,7 @@ def test_all_params(self):
             bypass_db_update,
             force_update,
             limit,
+            reports_bucket_name,
         ) = get_parameters(payload)
         self.assertFalse(dry_run)
         self.assertEqual(filter_after_in_days, 30)
@@ -80,6 +84,7 @@ def test_all_params(self):
         self.assertTrue(bypass_db_update)
         self.assertTrue(force_update)
         self.assertEqual(limit, 10)
+        self.assertEqual(reports_bucket_name, "my-custom-bucket")
 
     def test_string_coercion(self):
         payload = {
@@ -88,9 +93,18 @@ def test_string_coercion(self):
             "force_update": "true",
             "limit": "5",
         }
-        dry_run, _, _, _, _, _, bypass_db_update, force_update, limit = get_parameters(
-            payload
-        )
+        (
+            dry_run,
+            _,
+            _,
+            _,
+            _,
+            _,
+            bypass_db_update,
+            force_update,
+            limit,
+            _,
+        ) = get_parameters(payload)
         self.assertFalse(dry_run)
         self.assertTrue(bypass_db_update)
         self.assertTrue(force_update)
@@ -186,6 +200,45 @@ def test_limit_slices_datasets(
         self.assertEqual(result["total_candidates"], 20)
         self.assertEqual(result["total_in_call"], 5)
 
+    @patch(f"{_MODULE}._get_validator_version", return_value="7.0.0")
+    @patch(f"{_MODULE}._filter_out_datasets_without_blob", return_value=[("f", "ds-1")])
+    @patch(f"{_MODULE}.execute_workflows", return_value=["ds-1"])
+    @patch(f"{_MODULE}.TaskExecutionTracker")
+    def test_reports_bucket_name_override_is_used(
+        self, tracker_cls, exec_mock, filter_blob_mock, version_mock
+    ):
+        """When reports_bucket_name is provided, it overrides the env-derived bucket."""
+        session = self._make_session_mock(datasets=[("f", "ds-1")])
+        rebuild_missing_validation_reports(
+            validator_endpoint="https://staging.example.com/api",
+            dry_run=False,
+            reports_bucket_name="stg-gtfs-validator-results",
+            prod_env=True,
+            db_session=session,
+        )
+        _, call_kwargs = exec_mock.call_args
+        self.assertEqual(
+            call_kwargs["reports_bucket_name"], "stg-gtfs-validator-results"
+        )
+
+    @patch(f"{_MODULE}._get_validator_version", return_value="7.0.0")
+    @patch(f"{_MODULE}._filter_out_datasets_without_blob", return_value=[("f", "ds-1")])
+    @patch(f"{_MODULE}.execute_workflows", return_value=["ds-1"])
+    @patch(f"{_MODULE}.TaskExecutionTracker")
+    def test_reports_bucket_name_defaults_to_env_derived(
+        self, tracker_cls, exec_mock, filter_blob_mock, version_mock
+    ):
+        """When reports_bucket_name is not set, the env-derived prod bucket is used."""
+        session = self._make_session_mock(datasets=[("f", "ds-1")])
+        rebuild_missing_validation_reports(
+            validator_endpoint="https://staging.example.com/api",
+            dry_run=False,
+            prod_env=True,
+            db_session=session,
+        )
+        _, call_kwargs = exec_mock.call_args
+        self.assertEqual(call_kwargs["reports_bucket_name"], "gtfs-validator-results")
+
     @patch(f"{_MODULE}._get_validator_version", return_value="7.0.0")
     @patch(f"{_MODULE}._filter_out_datasets_without_blob", return_value=[("f", "ds-1")])
     @patch(f"{_MODULE}.execute_workflows", return_value=["ds-1"])
@@ -229,6 +282,7 @@ def test_handler_passes_all_params(self, rebuild_mock):
             "force_update": True,
             "limit": 10,
             "filter_op_statuses": ["published", "wip"],
+            "reports_bucket_name": "my-custom-bucket",
         }
         rebuild_missing_validation_reports_handler(payload)
         rebuild_mock.assert_called_once_with(
@@ -241,6 +295,7 @@ def test_handler_passes_all_params(self, rebuild_mock):
             prod_env=False,
             force_update=True,
             limit=10,
+            reports_bucket_name="my-custom-bucket",
         )
 
     @patch(f"{_MODULE}._get_validator_version", return_value="7.0.0")

infra/functions-python/main.tf

@@ -340,8 +340,6 @@ resource "google_cloudfunctions2_function" "process_validation_report" {
     vpc_connector_egress_settings = "PRIVATE_RANGES_ONLY"
 
     environment_variables = {
-      # FIXME, duplicate variable
-      ENV = var.environment
       ENVIRONMENT = var.environment
       PROJECT_ID = var.project_id
       GCP_REGION = var.gcp_region
@@ -395,7 +393,7 @@ resource "google_cloudfunctions2_function" "compute_validation_report_counters"
     vpc_connector_egress_settings = "PRIVATE_RANGES_ONLY"
 
     environment_variables = {
-      ENV = var.environment
+      ENVIRONMENT = var.environment
       PYTHONNODEBUGRANGES = 0
     }
 
@@ -596,7 +594,7 @@ resource "google_cloudfunctions2_function" "gbfs_validator_pubsub" {
     vpc_connector = data.google_vpc_access_connector.vpc_connector.id
     vpc_connector_egress_settings = "PRIVATE_RANGES_ONLY"
     environment_variables = {
-      ENV = var.environment
+      ENVIRONMENT = var.environment
       BUCKET_NAME = google_storage_bucket.gbfs_snapshots_bucket.name
       PROJECT_ID = var.project_id
       GCP_REGION = var.gcp_region
@@ -688,7 +686,7 @@ resource "google_cloudfunctions2_function" "backfill_dataset_service_date_range"
 
     environment_variables = {
       # prevents multiline logs from being truncated on GCP console
-      ENV = var.environment
+      ENVIRONMENT = var.environment
       PYTHONNODEBUGRANGES = 0
     }
     dynamic "secret_environment_variables" {
@@ -1151,7 +1149,7 @@ resource "google_cloudfunctions2_function" "pmtiles_builder" {
   service_config {
     environment_variables = {
       PROJECT_ID  = var.project_id
-      ENV = var.environment
+      ENVIRONMENT = var.environment
       PUBSUB_TOPIC_NAME = "rebuild-bounding-boxes-topic"
       MATERIALIZED_VIEW_QUEUE = google_cloud_tasks_queue.refresh_materialized_view_task_queue.name
       DATASETS_BUCKET_NAME = "${var.datasets_bucket_name}-${var.environment}"