diff --git a/api/src/feeds/impl/feeds_api_impl.py b/api/src/feeds/impl/feeds_api_impl.py index e255b21eb..bcc9b4b15 100644 --- a/api/src/feeds/impl/feeds_api_impl.py +++ b/api/src/feeds/impl/feeds_api_impl.py @@ -37,6 +37,7 @@ add_official_filter, get_gbfs_feeds_query, ) +from shared.common.availability_check_source import feed_availability_checks_filter from shared.common.seal_criteria import SealCriterionName from shared.common.error_handling import ( availability_from_after_to, @@ -367,7 +368,11 @@ def get_gtfs_feed_availability( if not feed: raise_http_error(404, gtfs_feed_not_found.format(id)) - query = db_session.query(GtfsFeedAvailabilityCheck).filter(GtfsFeedAvailabilityCheck.feed_id == feed.id) + # Only the daily check answers "was the feed available"; download diagnostics recorded in + # the same table are for the operations API. + query = db_session.query(GtfsFeedAvailabilityCheck).filter( + GtfsFeedAvailabilityCheck.feed_id == feed.id, feed_availability_checks_filter() + ) if from_dt: query = query.filter(GtfsFeedAvailabilityCheck.checked_at >= from_dt) if to_dt: diff --git a/api/src/scripts/populate_db_test_data.py b/api/src/scripts/populate_db_test_data.py index d3c9c3dc9..68f2e295a 100644 --- a/api/src/scripts/populate_db_test_data.py +++ b/api/src/scripts/populate_db_test_data.py @@ -332,6 +332,9 @@ def populate_test_datasets(self, filepath, db_session: "Session"): latency_ms=check.get("latency_ms"), error_type=check.get("error_type"), ) + # Left unset when absent so the column default applies; source is NOT NULL. + if "source" in check: + availability_check.source = check["source"] db_session.add(availability_check) db_session.commit() diff --git a/api/src/shared/common/availability_check_source.py b/api/src/shared/common/availability_check_source.py new file mode 100644 index 000000000..512ad5a35 --- /dev/null +++ b/api/src/shared/common/availability_check_source.py @@ -0,0 +1,41 @@ +"""Who wrote a `gtfs_feed_availability_check` row, and which rows count as feed availability. + +The table holds more than the daily availability check: failed nightly dataset downloads and +the download runs asked for by hand through the operations API are recorded there too, with +their request and response detail, so they can be debugged side by side. + +Only the daily check answers "was the feed available". Every public reader of the table - +the `/v1/gtfs_feeds/{id}/availability` endpoint and the seal's `available` criterion, both in +the nightly run and in the backfill - filters through `feed_availability_checks_filter()`, so +changing which sources count is a single edit to `FEED_AVAILABILITY_SOURCES`. The operations +API is the one reader that sees every source. + +This module is under `shared/common` so the API and every function that writes or reads the +table (`tasks_executor`, `batch_process_dataset`, `operations_api`) share one definition. +""" + +from enum import Enum +from typing import Final, Tuple + +from shared.database_gen.sqlacodegen_models import GtfsFeedAvailabilityCheck + + +class AvailabilityCheckSource(str, Enum): + """Values stored in `gtfs_feed_availability_check.source`.""" + + # The daily lightweight check run by tasks_executor. + AVAILABILITY_CHECK = "availability_check" + # A nightly dataset download that failed, recorded by batch_process_dataset. + DATASET_DOWNLOAD = "dataset_download" + # A run on the nightly job asked for through the operations API, download-only or full, + # success included. + DATASET_DOWNLOAD_MANUAL = "dataset_download_manual" + + +# The sources whose rows answer "was the feed available". +FEED_AVAILABILITY_SOURCES: Final[Tuple[AvailabilityCheckSource, ...]] = (AvailabilityCheckSource.AVAILABILITY_CHECK,) + + +def feed_availability_checks_filter(): + """SQLAlchemy clause keeping only the checks that count as feed availability.""" + return GtfsFeedAvailabilityCheck.source.in_([source.value for source in FEED_AVAILABILITY_SOURCES]) diff --git a/api/src/shared/db_models/gtfs_feed_availability_check_impl.py b/api/src/shared/db_models/gtfs_feed_availability_check_impl.py index ffc5fe2df..d168633db 100644 --- a/api/src/shared/db_models/gtfs_feed_availability_check_impl.py +++ b/api/src/shared/db_models/gtfs_feed_availability_check_impl.py @@ -27,4 +27,13 @@ def from_orm(cls, check: GtfsFeedAvailabilityCheckOrm | None) -> GtfsFeedAvailab status_code=check.status_code, latency_ms=float(check.latency_ms) if check.latency_ms is not None else None, error_type=check.error_type, + source=check.source, + error_message=check.error_message, + content_type=check.content_type, + is_zip=check.is_zip, + resolved_url=check.resolved_url, + request_headers=check.request_headers, + response_headers=check.response_headers, + redirect_urls=check.redirect_urls, + external_ip=check.external_ip, ) diff --git a/api/tests/integration/test_data/extra_test_data.json b/api/tests/integration/test_data/extra_test_data.json index 4f3e9f5ce..bee1d6d03 100644 --- a/api/tests/integration/test_data/extra_test_data.json +++ b/api/tests/integration/test_data/extra_test_data.json @@ -280,6 +280,17 @@ "status_code": null, "latency_ms": null, "error_type": "connection_timeout" + }, + { + "feed_stable_id": "mdb-1", + "checked_at": "2025-05-20T10:00:00+00:00", + "request_url": "https://example.com/gtfs.zip", + "request_type": "http_get", + "success": false, + "status_code": 403, + "latency_ms": 640, + "error_type": "http_error", + "source": "dataset_download" } ] } \ No newline at end of file diff --git a/api/tests/integration/test_feeds_api.py b/api/tests/integration/test_feeds_api.py index fe336b18c..cc345befd 100644 --- a/api/tests/integration/test_feeds_api.py +++ b/api/tests/integration/test_feeds_api.py @@ -989,7 +989,10 @@ def test_feeds_have_expected_license_info( # ---- GTFS Feed Availability endpoint tests ---- AVAILABILITY_FEED_ID = TEST_GTFS_FEED_STABLE_IDS[0] # mdb-1 — has availability_checks fixture data -AVAILABILITY_CHECKS_COUNT = 5 # number of checks added in extra_test_data.json for mdb-1 +# Daily availability checks added in extra_test_data.json for mdb-1. The fixture also has a +# failed dataset download for mdb-1, which the public endpoint must not return. +AVAILABILITY_CHECKS_COUNT = 5 +DATASET_DOWNLOAD_CHECKED_AT = "2025-05-20T10:00:00" def test_gtfs_feed_availability_basic(client: TestClient): @@ -1142,6 +1145,20 @@ def test_gtfs_feed_availability_pagination_offset(client: TestClient): assert first_timestamps.isdisjoint(second_timestamps) +def test_gtfs_feed_availability_excludes_download_diagnostics(client: TestClient): + """A failed dataset download recorded in the same table is not an availability check.""" + response = client.request( + "GET", + f"/v1/gtfs_feeds/{AVAILABILITY_FEED_ID}/availability", + headers=authHeaders, + params={"from": "2025-05-01T00:00:00Z"}, + ) + assert response.status_code == 200 + body = response.json() + assert body["total"] == 1 + assert all(not c["checked_at"].startswith(DATASET_DOWNLOAD_CHECKED_AT) for c in body["checks"]) + + def test_gtfs_feed_availability_not_found(client: TestClient): """Unknown feed ID returns 404.""" response = client.request( diff --git a/api/tests/unittest/models/test_availability_check_source_contract.py b/api/tests/unittest/models/test_availability_check_source_contract.py new file mode 100644 index 000000000..1df5ba7a4 --- /dev/null +++ b/api/tests/unittest/models/test_availability_check_source_contract.py @@ -0,0 +1,53 @@ +"""Contract tests pinning `gtfs_feed_availability_check.source` values. No database connection. + +`shared.common.availability_check_source` is the single definition of the sources the +functions write and of the ones that count as feed availability. The public availability +endpoint and the Seal of Reliability both filter through it; the operations API exposes every +source as an enum on its `source` filter. + +Adding a source means updating, in lockstep: `AvailabilityCheckSource` and the +`availability_source` parameter in `docs/OperationsAPI.yaml` plus a stub regen. +""" + +import unittest +from pathlib import Path + +import yaml + +from shared.common.availability_check_source import ( + FEED_AVAILABILITY_SOURCES, + AvailabilityCheckSource, + feed_availability_checks_filter, +) +from shared.database_gen.sqlacodegen_models import GtfsFeedAvailabilityCheck + +OPERATIONS_SPEC = Path(__file__).resolve().parents[4] / "docs" / "OperationsAPI.yaml" + + +class TestAvailabilityCheckSourceContract(unittest.TestCase): + def test_only_the_daily_check_counts_as_feed_availability(self): + """Changing this changes the public endpoint and the seal; it should be deliberate.""" + self.assertEqual(FEED_AVAILABILITY_SOURCES, (AvailabilityCheckSource.AVAILABILITY_CHECK,)) + + def test_filter_selects_the_feed_availability_sources(self): + clause = feed_availability_checks_filter() + compiled = clause.compile(compile_kwargs={"literal_binds": True}) + self.assertEqual(clause.left.table.name, GtfsFeedAvailabilityCheck.__tablename__) + self.assertEqual(clause.left.name, "source") + self.assertIn("'availability_check'", str(compiled)) + for source in set(AvailabilityCheckSource) - set(FEED_AVAILABILITY_SOURCES): + self.assertNotIn(f"'{source.value}'", str(compiled)) + + def test_operations_spec_lists_every_source(self): + spec = yaml.safe_load(OPERATIONS_SPEC.read_text()) + parameter = spec["components"]["parameters"]["availability_source"] + self.assertEqual( + set(parameter["schema"]["items"]["enum"]), + {source.value for source in AvailabilityCheckSource}, + ) + + def test_db_default_is_the_daily_check(self): + """A row written without a source must still count as an availability check.""" + column = GtfsFeedAvailabilityCheck.__table__.c.source + self.assertFalse(column.nullable) + self.assertIn(AvailabilityCheckSource.AVAILABILITY_CHECK.value, str(column.server_default.arg)) diff --git a/docs/OperationsAPI.yaml b/docs/OperationsAPI.yaml index d8f4c10a7..c99b783f4 100644 --- a/docs/OperationsAPI.yaml +++ b/docs/OperationsAPI.yaml @@ -189,6 +189,7 @@ paths: parameters: - $ref: "#/components/parameters/availability_from" - $ref: "#/components/parameters/availability_to" + - $ref: "#/components/parameters/availability_source" - $ref: "#/components/parameters/limit_query_param_availability_endpoint" - $ref: "#/components/parameters/offset" - $ref: "#/components/parameters/availability_sort" @@ -205,6 +206,47 @@ paths: description: GTFS feed not found. 500: description: Internal server error. + /v1/operations/gtfs_feeds/{id}/download_debug: + parameters: + - $ref: "#/components/parameters/feed_id_path_param" + post: + description: > + Downloads the GTFS feed through the same code path used by the nightly dataset download. Nothing is stored unless `dry_run` is false. `run_on` in the body picks where: `api` (the default) downloads within this request and answers 200 with the request and response detail; `nightly_job` queues the download on the nightly dataset download job itself - same function, instance size, network egress and request - and answers 202 at once; with `dry_run = false` that run also processes the dataset exactly as the nightly run does. That outcome, dry or full, is recorded as an availability check with `source = dataset_download_manual`, usually within a few minutes; read it from the availability endpoint with that source and `from` set to `queued_at`. `max_bytes` and `timeout_seconds` apply only to `api`. With no options, `api` behaves like the nightly download: the whole file is read and there is no read timeout. The file is hashed as it streams and never written to disk, so any size can be checked. Reading stops after 480 seconds so the diagnostics are returned before the request itself times out; `stop_reason` then reports `deadline`. Credential values in headers and URLs are redacted. + tags: + - "operations" + operationId: debugGtfsFeedDownload + security: + - Authentication: [] + parameters: + - $ref: "#/components/parameters/download_debug_max_bytes" + - $ref: "#/components/parameters/download_debug_timeout_seconds" + requestBody: + required: false + content: + application/json: + schema: + $ref: "#/components/schemas/GtfsFeedDownloadDebugRequest" + responses: + 200: + description: Diagnostics for the download attempt. + content: + application/json: + schema: + $ref: "#/components/schemas/GtfsFeedDownloadDebugResponse" + 202: + description: "`run_on = nightly_job`: the download was queued on the nightly job." + content: + application/json: + schema: + $ref: "#/components/schemas/GtfsFeedDownloadQueuedResponse" + 400: + description: Invalid request parameters, or the feed has no producer URL. + 404: + description: GTFS feed not found. + 500: + description: Internal server error. + 503: + description: "`run_on = nightly_job` while dataset processing is not configured in this environment." /v1/operations/gtfs_rt_feeds/{id}: parameters: - $ref: "#/components/parameters/feed_id_path_param" @@ -1830,6 +1872,205 @@ components: nullable: true description: Machine-readable error category when the check failed. example: timeout + source: + type: string + nullable: true + description: What wrote the check; see the `source` filter of the availability endpoint. + example: dataset_download + error_message: + type: string + nullable: true + description: Error detail when the check failed, with credentials redacted. + content_type: + type: string + nullable: true + description: Normalised Content-Type of the response. + example: application/zip + is_zip: + type: boolean + nullable: true + description: Whether the response content is a ZIP file. + resolved_url: + type: string + nullable: true + description: URL actually requested after credentials were applied, with credentials redacted. + request_headers: + type: object + nullable: true + additionalProperties: + type: string + description: Headers sent with the request, with credential values redacted. + response_headers: + type: object + nullable: true + additionalProperties: + type: string + description: Headers returned by the server, with credential values redacted. + redirect_urls: + type: array + nullable: true + items: + type: string + description: URLs the request was redirected through. + external_ip: + type: string + nullable: true + description: Egress IP address observed for the request, when it was captured. + example: 34.95.1.1 + GtfsFeedDownloadQueuedResponse: + type: object + description: A download queued on the nightly dataset download job (`run_on = nightly_job`). + required: + - feed_id + - execution_id + - queued_at + - source + - dry_run + properties: + dry_run: + type: boolean + description: Whether the queued run only downloads (true) or downloads and processes the dataset as the nightly run does (false). + example: true + feed_id: + type: string + description: Stable identifier of the GTFS feed. + example: mdb-123 + execution_id: + type: string + description: Execution identifier of the queued dry run, as it appears in the job's logs. + queued_at: + type: string + format: date-time + description: When the dry run was queued. Its result is the first availability check of `source` at or after this time. + source: + type: string + description: > + Availability check source the download outcome is recorded under: `dataset_download_manual`, for dry and full runs alike. + example: dataset_download_manual + GtfsFeedDownloadDebugRequest: + type: object + description: Options for a download debug attempt. + properties: + run_on: + type: string + enum: + - api + - nightly_job + default: api + description: > + Where the download runs: `api` within this request, answering with its diagnostics; `nightly_job` on the nightly dataset download job, answering 202 with where to read the result. + dry_run: + type: boolean + default: true + description: > + With `run_on = nightly_job`, false runs the complete nightly process instead of only the download: if the file changed, a new dataset is uploaded, stored and becomes the feed's latest, and its follow-up tasks run. The download outcome is recorded with `source = dataset_download_manual`. Only allowed with `run_on = nightly_job` and without `headers`, so a stored dataset is never produced with headers the nightly run would not send. + headers: + type: object + nullable: true + maxProperties: 50 + additionalProperties: + type: string + description: > + Request headers to send instead of the defaults (the per-feed override, or the default User-Agent and Referer), as the local download verifier's --header does. The feed's API key is still applied on top. Omit to use the same headers as the nightly download. + example: + User-Agent: curl/8.0 + GtfsFeedDownloadDebugResponse: + type: object + description: > + Request and response detail of a single feed download attempt. Credential values + in headers and URLs are replaced with a redaction marker. + required: + - feed_id + - producer_url + - success + properties: + feed_id: + type: string + description: Stable identifier of the GTFS feed. + example: mdb-123 + producer_url: + type: string + description: Feed URL that was requested, with credentials redacted. + example: https://example.com/gtfs.zip + resolved_url: + type: string + nullable: true + description: URL actually requested after credentials were applied, with credentials redacted. + success: + type: boolean + description: Whether the download returned a 2xx response. + example: false + status_code: + type: integer + nullable: true + description: HTTP status code of the final response. + example: 403 + latency_ms: + type: integer + nullable: true + description: Time to first response, in milliseconds. + example: 1840 + request_headers: + type: object + nullable: true + additionalProperties: + type: string + description: Headers sent with the request, with credential values redacted. + response_headers: + type: object + nullable: true + additionalProperties: + type: string + description: Headers returned by the server, with credential values redacted. + redirect_urls: + type: array + nullable: true + items: + type: string + description: URLs the request was redirected through. + external_ip: + type: string + nullable: true + description: Egress IP address observed for the request, when it could be determined. + example: 34.95.1.1 + content_type: + type: string + nullable: true + description: Normalised Content-Type of the response. + example: text/html + is_zip: + type: boolean + nullable: true + description: Whether the downloaded content is a ZIP file. + downloaded_bytes: + type: integer + nullable: true + description: Number of bytes read before the download completed or hit the byte cap. + example: 2038 + truncated: + type: boolean + description: Whether the download stopped before the end of the file, at the byte cap or the time limit. + example: false + stop_reason: + type: string + nullable: true + enum: [max_bytes, deadline] + description: > + Why reading stopped before the end of the file: `max_bytes` when the requested byte cap was reached, `deadline` when the 480 second endpoint limit was. Null when the whole file was read. + duration_ms: + type: integer + nullable: true + description: Time from sending the request to the last byte read, in milliseconds. + example: 12500 + error_type: + type: string + nullable: true + description: Machine-readable error category when the download failed. + example: ValueError + error_message: + type: string + nullable: true + description: Error detail when the download failed, with credentials redacted. GtfsFeedContinuousCoverageResponse: type: object description: > @@ -3931,6 +4172,27 @@ components: minimum: 0 default: 0 example: 0 + download_debug_max_bytes: + x-operation: true + name: max_bytes + in: query + description: Stop reading the feed after this many bytes, for a quicker check. Omit to read the whole file, as the nightly download does. + required: false + schema: + type: integer + minimum: 1024 + example: 1048576 + download_debug_timeout_seconds: + x-operation: true + name: timeout_seconds + in: query + description: Connect and read timeout for the download, in seconds. Omit for no timeout, as the nightly download does; reading still stops at the 480 second endpoint limit. + required: false + schema: + type: integer + minimum: 1 + maximum: 480 + example: 30 limit_query_param_availability_endpoint: x-operation: true name: limit @@ -3953,6 +4215,24 @@ components: type: string format: date-time example: "2026-04-01T00:00:00Z" + availability_source: + x-operation: true + name: source + in: query + description: > + Return only checks written by these sources; repeat the parameter to combine several (`?source=dataset_download&source=dataset_download_manual`). Without it, checks from every source are returned. Sources: `availability_check` (the daily lightweight check), `dataset_download` (failed nightly dataset downloads) or `dataset_download_manual` (download_debug runs on the nightly job, dry or full, success included). Only `availability_check` counts toward the public availability endpoint and the Seal of Reliability. + required: false + style: form + explode: true + schema: + type: array + items: + type: string + enum: + - availability_check + - dataset_download + - dataset_download_manual + example: [dataset_download] availability_to: x-operation: true name: to diff --git a/functions-python/batch_process_dataset/src/main.py b/functions-python/batch_process_dataset/src/main.py index 2ff117741..a561e7e88 100644 --- a/functions-python/batch_process_dataset/src/main.py +++ b/functions-python/batch_process_dataset/src/main.py @@ -32,6 +32,7 @@ from sqlalchemy import func from sqlalchemy.orm import Session +from shared.common.availability_check_source import AvailabilityCheckSource from shared.common.gcp_memory_utils import limit_gcp_memory from shared.common.gcp_utils import ( create_refresh_materialized_view_task, @@ -42,7 +43,10 @@ from shared.dataset_service.main import DatasetTraceService, DatasetTrace, Status from shared.helpers.logger import init_logger, get_logger from shared.helpers.utils import ( - download_and_get_hash, + FeedDownloadError, + build_availability_check, + download_feed, + get_external_ip, get_hash_from_file, download_from_gcs, ) @@ -83,8 +87,12 @@ def __init__( api_key_parameter_name, public_hosted_datasets_url, dataset_stable_id=None, + record_source: Optional[str] = None, ): self.logger = get_logger(DatasetProcessor.__name__, feed_stable_id) + # Set for runs someone asked for by hand: the download outcome is then recorded + # under this source whether it succeeds or fails, so they can read it back. + self.record_source = record_source self.producer_url = producer_url self.bucket_name = bucket_name self.latest_hash = latest_hash @@ -139,19 +147,91 @@ def download_content(self, temporary_file_path, feed_id): """ Downloads the content of a URL and return the hash of the file """ - file_hash = download_and_get_hash( - self.producer_url, - file_path=temporary_file_path, - feed_id=feed_id, - authentication_type=self.authentication_type, - api_key_parameter_name=self.api_key_parameter_name, - credentials=self.feed_credentials, - logger=self.logger, - ) + try: + result = download_feed( + self.producer_url, + file_path=temporary_file_path, + feed_id=feed_id, + authentication_type=self.authentication_type, + api_key_parameter_name=self.api_key_parameter_name, + credentials=self.feed_credentials, + logger=self.logger, + ) + except FeedDownloadError as exc: + if feed_id: + self.record_download_diagnostics( + feed_id, + exc.diagnostics, + source=self.record_source + or AvailabilityCheckSource.DATASET_DOWNLOAD.value, + ) + raise + if feed_id and self.record_source: + result.diagnostics.external_ip = get_external_ip() + self.record_download_diagnostics( + feed_id, result.diagnostics, source=self.record_source + ) + file_hash = result.file_hash self.logger.info("hash is: %s", file_hash) is_zip = zipfile.is_zipfile(temporary_file_path) return file_hash, is_zip + @with_db_session + def record_download_diagnostics( + self, + feed_id, + diagnostics, + source: str = AvailabilityCheckSource.DATASET_DOWNLOAD.value, + db_session: Session = None, + ): + """ + Stores the diagnostics of a download in gtfs_feed_availability_check: failed + nightly downloads, and every run asked for by hand (dry or full) whatever its + outcome. + Never raises: a storage failure must not mask the original download error. + """ + try: + db_session.add( + build_availability_check(diagnostics, feed_id=feed_id, source=source) + ) + db_session.commit() + except Exception as exc: + self.logger.error("Could not store download diagnostics: %s", exc) + + def dry_run_download(self, feed_id, headers: Optional[dict] = None) -> bool: + """ + Downloads the feed exactly as transfer_dataset does - same function, temp file + and request - and records the diagnostics as a manual download. Nothing is + uploaded and no dataset is created. headers, when given, replace the defaults. + :return: whether the download succeeded + """ + temp_file_path = self.generate_temp_filename() + try: + diagnostics = download_feed( + self.producer_url, + temp_file_path, + feed_id=feed_id, + authentication_type=self.authentication_type, + api_key_parameter_name=self.api_key_parameter_name, + credentials=self.feed_credentials, + logger=self.logger, + headers=headers, + ).diagnostics + # Failures already carry it; a success needs it too, to compare egress + # with a test run from another service. + diagnostics.external_ip = get_external_ip() + except FeedDownloadError as exc: + diagnostics = exc.diagnostics + finally: + if os.path.exists(temp_file_path): + os.remove(temp_file_path) + self.record_download_diagnostics( + feed_id, + diagnostics, + source=AvailabilityCheckSource.DATASET_DOWNLOAD_MANUAL.value, + ) + return diagnostics.success + def upload_dataset_zip_to_storage( self, source_file_path, @@ -553,6 +633,32 @@ def record_trace( trace_service.save(trace) +def process_dry_run(json_payload: dict) -> str: + """ + Handles a dry-run message: one download through the nightly code path, with its + diagnostics recorded and nothing else. Skips the execution trace on purpose, so a + dry run never counts against the nightly job's maximum executions. + """ + stable_id = json_payload["feed_stable_id"] + logger = get_logger("process_dataset", stable_id) + processor = DatasetProcessor( + json_payload["producer_url"], + json_payload["feed_id"], + stable_id, + json_payload["execution_id"], + None, + os.getenv("DATASETS_BUCKET_NAME"), + int(json_payload["authentication_type"]), + json_payload["api_key_parameter_name"], + os.getenv("PUBLIC_HOSTED_DATASETS_URL"), + ) + success = processor.dry_run_download( + json_payload["feed_id"], headers=json_payload.get("headers") + ) + logger.info("Dry run download %s", "succeeded" if success else "failed") + return "Dry run completed." + + @functions_framework.cloud_event def process_dataset(cloud_event: CloudEvent): """ @@ -596,6 +702,14 @@ def process_dataset(cloud_event: CloudEvent): logging.error(f"Function completed with error:{error_message}") return + if json_payload.get("dry_run"): + try: + return process_dry_run(json_payload) + except Exception as e: + error_message = f"[{stable_id}] Dry run failed: [{type(e).__name__}]: {e}" + logging.error(error_message) + return error_message + try: try: maximum_executions = int(os.getenv("MAXIMUM_EXECUTIONS", "1")) @@ -637,6 +751,7 @@ def process_dataset(cloud_event: CloudEvent): json_payload["api_key_parameter_name"], public_hosted_datasets_url, json_payload.get("dataset_stable_id"), + record_source=json_payload.get("record_source"), ) if json_payload.get("use_bucket_latest", False): dataset_file = processor.process_from_bucket() diff --git a/functions-python/batch_process_dataset/src/scripts/download_verifier.py b/functions-python/batch_process_dataset/src/scripts/download_verifier.py index 08a8be08c..7467168d8 100644 --- a/functions-python/batch_process_dataset/src/scripts/download_verifier.py +++ b/functions-python/batch_process_dataset/src/scripts/download_verifier.py @@ -1,51 +1,63 @@ +import argparse +import json import logging import os +import uuid from main import DatasetProcessor from gcp_storage_emulator.server import create_server +from shared.helpers.utils import FeedDownloadError, download_feed HOST = "localhost" PORT = 9023 BUCKET_NAME = "verifier" -PRODUCER_URL = "https://example.com/dataset.zip" # Replace with actual producer URL -def verify_download_content(producer_url: str): +def verify_download_content(producer_url: str, headers=None): """ - Verifies the download_content is able to retrieve the file - This is useful to simulate the download code locally and test issues related with user-agent and downloaded content. - Not supported authenticated feeds currently. - """ - logging.info("Verifying downloaded content... (not implemented)") + Verifies the shared download code is able to retrieve the file and logs the request + and response detail of the attempt, whether it succeeds or fails. + This is useful to simulate the download code locally and test issues related with + user-agent and downloaded content. It never reads from or writes to a database. + Authenticated feeds are not supported currently. - logging.info(f"Producer URL: {producer_url}") + Pass headers to reproduce a per-feed feed_download/http_headers override from + production; they replace the default headers, as the override does. + """ + logging.info("Producer URL: %s", producer_url) - processor = DatasetProcessor( - producer_url=producer_url, - feed_id=None, - feed_stable_id=None, - execution_id=None, - latest_hash=None, - bucket_name=None, - authentication_type=0, - api_key_parameter_name=None, - public_hosted_datasets_url=None, + tempfile = os.path.join(os.environ["WORKING_DIR"], f"{uuid.uuid4()}.zip") + logging.info("Temp filename: %s", tempfile) + try: + result = download_feed(producer_url, tempfile, headers=headers) + except FeedDownloadError as exc: + logging.error("Download failed") + logging.error(json.dumps(exc.diagnostics.as_dict(), indent=2, default=str)) + raise + logging.info(json.dumps(result.diagnostics.as_dict(), indent=2, default=str)) + logging.info( + "Downloaded file from %s is a valid ZIP file: %s", producer_url, result.is_zip ) - tempfile = processor.generate_temp_filename() - logging.info(f"Temp filename: {tempfile}") - file_hash, is_zip = processor.download_content(tempfile, "feed_id") - logging.info(f"Downloaded file from {producer_url} is a valid ZIP file: {is_zip}") - logging.info(f"File hash: {file_hash}") + logging.info("File hash: %s", result.file_hash) + + +def parse_header(value: str) -> tuple: + """Parse a "Name: value" command line header.""" + name, separator, header_value = value.partition(":") + if not separator or not name.strip(): + raise argparse.ArgumentTypeError( + f"Invalid header {value!r}, expected 'Name: value'" + ) + return name.strip(), header_value.strip() def verify_upload_dataset(producer_url: str): """ Verifies the upload_dataset is able to upload the dataset to the GCP storage emulator. - This is useful to simulate the upload code locally and test issues related with user-agent and uploaded content. - This function also tests the DatasetProcessor class methods for generating a temporary filename - and uploading the dataset. - :param producer_url: - :return: + This is useful to simulate the upload code locally and test issues related with + user-agent and uploaded content. + This function also tests the DatasetProcessor class methods for generating a temporary + filename and uploading the dataset. """ processor = DatasetProcessor( producer_url=producer_url, @@ -59,31 +71,60 @@ def verify_upload_dataset(producer_url: str): public_hosted_datasets_url=None, ) tempfile = processor.generate_temp_filename() - logging.info(f"Temp filename: {tempfile}") + logging.info("Temp filename: %s", tempfile) dataset_file = processor.transfer_dataset("feed_id_2126", False) - logging.info(f"Dataset File: {dataset_file}") + logging.info("Dataset File: %s", dataset_file) -if __name__ == "__main__": +def parse_args(): + parser = argparse.ArgumentParser( + description="Download a GTFS feed through the shared download code and print the " + "request and response detail of the attempt." + ) + parser.add_argument("--url", required=True, help="Producer URL of the feed.") + parser.add_argument( + "--header", + dest="headers", + action="append", + type=parse_header, + default=None, + help="Request header as 'Name: value', repeatable. Replaces the default headers, " + "e.g. to reproduce a production feed_download/http_headers override.", + ) + parser.add_argument( + "--skip-upload", + action="store_true", + help="Only verify the download, skipping the GCP storage emulator.", + ) + return parser.parse_args() + + +def main(): logging.basicConfig(level=logging.INFO) - # Replace with actual producer URL - try: - os.environ["STORAGE_EMULATOR_HOST"] = f"http://{HOST}:{PORT}" - os.environ["WORKING_DIR"] = "/tmp/verifier" - # create working dir if not exists - if not os.path.exists(os.environ["WORKING_DIR"]): - os.makedirs(os.environ["WORKING_DIR"]) - - server = create_server( - host=HOST, port=PORT, in_memory=False, default_bucket=BUCKET_NAME - ) - server.start() + args = parse_args() + headers = dict(args.headers) if args.headers else None + + os.environ["WORKING_DIR"] = "/tmp/verifier" + os.makedirs(os.environ["WORKING_DIR"], exist_ok=True) - verify_download_content(producer_url=PRODUCER_URL) + if args.skip_upload: + verify_download_content(producer_url=args.url, headers=headers) logging.info("Download content verification completed successfully.") - verify_upload_dataset(producer_url=PRODUCER_URL) - except Exception as e: - logging.error(f"Error verifying download content: {e}") + return + + os.environ["STORAGE_EMULATOR_HOST"] = f"http://{HOST}:{PORT}" + server = create_server( + host=HOST, port=PORT, in_memory=False, default_bucket=BUCKET_NAME + ) + server.start() + try: + verify_download_content(producer_url=args.url, headers=headers) + logging.info("Download content verification completed successfully.") + verify_upload_dataset(producer_url=args.url) finally: server.stop() logging.info("Verification completed.") + + +if __name__ == "__main__": + main() diff --git a/functions-python/batch_process_dataset/tests/test_download_failure_recording.py b/functions-python/batch_process_dataset/tests/test_download_failure_recording.py new file mode 100644 index 000000000..c774dd471 --- /dev/null +++ b/functions-python/batch_process_dataset/tests/test_download_failure_recording.py @@ -0,0 +1,169 @@ +import unittest +from unittest.mock import MagicMock, patch + +import pytest + +from main import DatasetProcessor +from shared.helpers.utils import ( + FeedDownloadError, + FeedDownloadResult, + FeedRequestDiagnostics, +) + + +def _processor(feed_id="feed-1"): + return DatasetProcessor( + producer_url="https://example.com/gtfs.zip", + feed_id=feed_id, + feed_stable_id="mdb-1", + execution_id=None, + latest_hash=None, + bucket_name=None, + authentication_type=0, + api_key_parameter_name=None, + public_hosted_datasets_url=None, + ) + + +def _diagnostics(): + return FeedRequestDiagnostics( + request_url="https://example.com/gtfs.zip", + resolved_url="https://example.com/gtfs.zip", + request_headers={"User-Agent": "agent"}, + response_headers={"Server": "cloudflare"}, + status_code=403, + latency_ms=1840, + redirect_urls=[], + external_ip="34.1.2.3", + content_type="text/html", + is_zip=False, + error_type="ValueError", + error_message="Invalid HTTP response code: [403]", + ) + + +class TestDownloadFailureRecording(unittest.TestCase): + def test_failure_records_one_row_and_reraises(self): + processor = _processor() + diagnostics = _diagnostics() + + with patch( + "main.download_feed", + side_effect=FeedDownloadError("boom", diagnostics), + ): + with patch.object(processor, "record_download_diagnostics") as mock_record: + with pytest.raises(FeedDownloadError): + processor.download_content("/tmp/does-not-matter", "feed-1") + + mock_record.assert_called_once() + args = mock_record.call_args[0] + self.assertEqual(args[0], "feed-1") + self.assertIs(args[1], diagnostics) + + def test_success_records_nothing(self): + processor = _processor() + + with patch( + "main.download_feed", + return_value=FeedDownloadResult( + file_hash="abc123", is_zip=True, diagnostics=_diagnostics() + ), + ): + with patch("main.zipfile.is_zipfile", return_value=True): + with patch.object( + processor, "record_download_diagnostics" + ) as mock_record: + file_hash, is_zip = processor.download_content( + "/tmp/does-not-matter", "feed-1" + ) + + self.assertEqual(file_hash, "abc123") + self.assertTrue(is_zip) + mock_record.assert_not_called() + + def test_no_feed_id_skips_recording(self): + """The local verifier runs without a feed id, so there is no row to attach.""" + processor = _processor(feed_id=None) + + with patch( + "main.download_feed", + side_effect=FeedDownloadError("boom", _diagnostics()), + ): + with patch.object(processor, "record_download_diagnostics") as mock_record: + with pytest.raises(FeedDownloadError): + processor.download_content("/tmp/does-not-matter", None) + + mock_record.assert_not_called() + + def test_storage_failure_does_not_mask_download_error(self): + processor = _processor() + + session = MagicMock() + session.commit.side_effect = Exception("db down") + + with patch( + "main.download_feed", + side_effect=FeedDownloadError("boom", _diagnostics()), + ): + with patch("main.build_availability_check", return_value=MagicMock()): + with pytest.raises(FeedDownloadError): + processor.download_content("/tmp/does-not-matter", "feed-1") + + def test_recorded_row_is_tagged_as_dataset_download(self): + processor = _processor() + diagnostics = _diagnostics() + session = MagicMock() + + with patch("main.build_availability_check") as mock_build: + processor.record_download_diagnostics( + "feed-1", diagnostics, db_session=session + ) + + mock_build.assert_called_once_with( + diagnostics, feed_id="feed-1", source="dataset_download" + ) + session.add.assert_called_once() + session.commit.assert_called_once() + + +class TestManualRunRecording(unittest.TestCase): + """A run asked for by hand records its download outcome, success included.""" + + def _manual_processor(self): + processor = _processor() + processor.record_source = "dataset_download_manual" + return processor + + def test_success_is_recorded_under_the_manual_source(self): + processor = self._manual_processor() + diagnostics = _diagnostics() + result = FeedDownloadResult( + file_hash="abc123", is_zip=True, diagnostics=diagnostics + ) + + with patch("main.download_feed", return_value=result), patch( + "main.zipfile.is_zipfile", return_value=True + ), patch("main.get_external_ip", return_value="34.1.2.3"), patch.object( + processor, "record_download_diagnostics" + ) as record: + file_hash, _ = processor.download_content("/tmp/x", "feed-1") + + self.assertEqual(file_hash, "abc123") + record.assert_called_once_with( + "feed-1", diagnostics, source="dataset_download_manual" + ) + self.assertEqual(diagnostics.external_ip, "34.1.2.3") + + def test_failure_is_recorded_under_the_manual_source(self): + processor = self._manual_processor() + diagnostics = _diagnostics() + + with patch( + "main.download_feed", side_effect=FeedDownloadError("boom", diagnostics) + ), patch.object(processor, "record_download_diagnostics") as record: + with pytest.raises(FeedDownloadError): + processor.download_content("/tmp/x", "feed-1") + + record.assert_called_once_with( + "feed-1", diagnostics, source="dataset_download_manual" + ) diff --git a/functions-python/batch_process_dataset/tests/test_dry_run_download.py b/functions-python/batch_process_dataset/tests/test_dry_run_download.py new file mode 100644 index 000000000..335b34335 --- /dev/null +++ b/functions-python/batch_process_dataset/tests/test_dry_run_download.py @@ -0,0 +1,136 @@ +import base64 +import json +import os +import unittest +from unittest.mock import patch + +from cloudevents.http import CloudEvent + +from main import DatasetProcessor, process_dataset +from shared.common.availability_check_source import AvailabilityCheckSource +from shared.helpers.utils import ( + FeedDownloadError, + FeedDownloadResult, + FeedRequestDiagnostics, +) + +MANUAL_SOURCE = AvailabilityCheckSource.DATASET_DOWNLOAD_MANUAL.value + + +def _processor(): + return DatasetProcessor( + producer_url="https://example.com/gtfs.zip", + feed_id="feed-1", + feed_stable_id="mdb-1", + execution_id="download-dry-run-1", + latest_hash=None, + bucket_name=None, + authentication_type=0, + api_key_parameter_name=None, + public_hosted_datasets_url=None, + ) + + +def _diagnostics(**overrides): + values = dict( + request_url="https://example.com/gtfs.zip", + status_code=200, + content_type="application/zip", + is_zip=True, + ) + values.update(overrides) + return FeedRequestDiagnostics(**values) + + +def _event(payload): + data = base64.b64encode(json.dumps(payload).encode()).decode() + return CloudEvent( + {"type": "event.type", "source": "event.source"}, + {"message": {"data": data}}, + ) + + +class TestDryRunDownload(unittest.TestCase): + def test_success_is_recorded_with_egress_ip_and_temp_file_removed(self): + processor = _processor() + diagnostics = _diagnostics() + written = {} + + def fake_download(url, file_path, **kwargs): + # The real job writes to disk; so does the dry run. + with open(file_path, "wb") as f: + f.write(b"PK\x03\x04") + written["path"] = file_path + written["headers"] = kwargs["headers"] + return FeedDownloadResult( + file_hash="abc", is_zip=True, diagnostics=diagnostics + ) + + with patch.dict(os.environ, {"WORKING_DIR": "/tmp"}), patch( + "main.download_feed", side_effect=fake_download + ), patch("main.get_external_ip", return_value="34.1.2.3"), patch.object( + processor, "record_download_diagnostics" + ) as record: + success = processor.dry_run_download( + "feed-1", headers={"User-Agent": "curl/8.0"} + ) + + self.assertTrue(success) + self.assertEqual(written["headers"], {"User-Agent": "curl/8.0"}) + self.assertFalse(os.path.exists(written["path"])) + self.assertEqual(diagnostics.external_ip, "34.1.2.3") + record.assert_called_once_with("feed-1", diagnostics, source=MANUAL_SOURCE) + + def test_failure_is_recorded_not_raised(self): + processor = _processor() + diagnostics = _diagnostics(status_code=403, error_type="ValueError") + + with patch( + "main.download_feed", + side_effect=FeedDownloadError("boom", diagnostics), + ), patch.object(processor, "record_download_diagnostics") as record: + success = processor.dry_run_download("feed-1") + + self.assertFalse(success) + record.assert_called_once_with("feed-1", diagnostics, source=MANUAL_SOURCE) + + +class TestProcessDatasetDryRun(unittest.TestCase): + payload = { + "execution_id": "download-dry-run-1", + "producer_url": "https://example.com/gtfs.zip", + "feed_stable_id": "mdb-1", + "feed_id": "feed-1", + "dataset_id": None, + "dataset_hash": None, + "authentication_type": 0, + "authentication_info_url": None, + "api_key_parameter_name": None, + "dry_run": True, + "headers": {"User-Agent": "curl/8.0"}, + } + + def test_dry_run_only_downloads(self): + """No trace, no upload, no dataset: only the download and its diagnostics.""" + with patch.object( + DatasetProcessor, "dry_run_download", return_value=True + ) as dry_run, patch("main.DatasetTraceService") as trace_service, patch.object( + DatasetProcessor, "process_from_producer_url" + ) as process, patch( + "main.record_trace" + ) as record_trace: + result = process_dataset(_event(self.payload)) + + self.assertEqual(result, "Dry run completed.") + dry_run.assert_called_once_with("feed-1", headers={"User-Agent": "curl/8.0"}) + trace_service.assert_not_called() + process.assert_not_called() + record_trace.assert_not_called() + + def test_dry_run_error_is_returned(self): + with patch.object( + DatasetProcessor, "dry_run_download", side_effect=RuntimeError("db down") + ): + result = process_dataset(_event(self.payload)) + + self.assertIn("Dry run failed", result) diff --git a/functions-python/helpers/pub_sub.py b/functions-python/helpers/pub_sub.py index 605fec7b7..3f95c6e69 100644 --- a/functions-python/helpers/pub_sub.py +++ b/functions-python/helpers/pub_sub.py @@ -17,7 +17,7 @@ import logging import os import uuid -from typing import Dict, List +from typing import Dict, List, Optional from google.cloud import pubsub_v1 from google.cloud.pubsub_v1 import PublisherClient @@ -76,9 +76,22 @@ def trigger_dataset_download( feed: Feed | Gtfsfeed, execution_id: str, topic_name: str = DATASET_BATCH_TOPIC, + dry_run: bool = False, + headers: Optional[dict] = None, + record_source: Optional[str] = None, + dataset_stable_id: Optional[str] = None, + dataset_hash: Optional[str] = None, ) -> None: """Publishes the feed to the configured Pub/Sub topic. + dry_run asks batch_process_dataset to only download the feed and record the + diagnostics, storing no dataset; headers replace the default request headers. + record_source has the job record the download outcome under that source even when + it succeeds, for runs asked for by hand. + dataset_stable_id and dataset_hash describe the feed's latest dataset, as the nightly + scheduler sends them; without the hash the job treats any download as a change and + stores a new dataset. Left out for a new feed, which has none. + No-ops when no topic is configured (``DATASET_PROCESSING_TOPIC_NAME`` unset), e.g. local development and tests, so the caller never constructs a Pub/Sub client or blocks on a publish. Mirrors ``create_web_revalidation_task``. @@ -99,11 +112,20 @@ def trigger_dataset_download( "feed_stable_id": feed.stable_id, "feed_id": feed.id, "dataset_id": None, - "dataset_hash": None, + "dataset_hash": dataset_hash, "authentication_type": feed.authentication_type, "authentication_info_url": feed.authentication_info_url, "api_key_parameter_name": feed.api_key_parameter_name, } + if dataset_stable_id: + message_data["dataset_stable_id"] = dataset_stable_id + # Only set when asked, so the nightly message is unchanged. + if dry_run: + message_data["dry_run"] = True + if headers is not None: + message_data["headers"] = headers + if record_source: + message_data["record_source"] = record_source try: # Convert to JSON string diff --git a/functions-python/helpers/tests/test_download_debug.py b/functions-python/helpers/tests/test_download_debug.py new file mode 100644 index 000000000..694d456b6 --- /dev/null +++ b/functions-python/helpers/tests/test_download_debug.py @@ -0,0 +1,388 @@ +import io +import itertools +import os +import unittest +import zipfile +from unittest.mock import MagicMock, patch + +import pytest + +import utils +from utils import ( + REDACTED, + FeedDownloadError, + FeedRequestDiagnostics, + build_availability_check, + download_feed, + get_external_ip, + sanitize_headers, + sanitize_url, +) + + +class TestSanitizeHeaders(unittest.TestCase): + def test_redacts_known_secret_headers(self): + result = sanitize_headers( + { + "User-Agent": "agent", + "Authorization": "Bearer abc", + "Cookie": "session=1", + "X-Api-Key": "secret", + } + ) + self.assertEqual(result["User-Agent"], "agent") + self.assertEqual(result["Authorization"], REDACTED) + self.assertEqual(result["Cookie"], REDACTED) + self.assertEqual(result["X-Api-Key"], REDACTED) + + def test_redacts_api_key_parameter_name_header(self): + result = sanitize_headers( + {"User-Agent": "agent", "my-feed-token": "secret"}, + api_key_parameter_name="my-feed-token", + ) + self.assertEqual(result["my-feed-token"], REDACTED) + self.assertEqual(result["User-Agent"], "agent") + + def test_header_names_are_preserved(self): + result = sanitize_headers({"Authorization": "Bearer abc"}) + self.assertIn("Authorization", result) + + def test_none_returns_none(self): + self.assertIsNone(sanitize_headers(None)) + + +class TestSanitizeUrl(unittest.TestCase): + def test_redacts_api_key_query_parameter(self): + result = sanitize_url( + "https://example.com/gtfs.zip?operator=CE&token=secret", + api_key_parameter_name="token", + ) + self.assertIn("operator=CE", result) + self.assertNotIn("secret", result) + + def test_redacts_known_secret_query_parameters(self): + result = sanitize_url("https://example.com/gtfs.zip?api_key=secret&x=1") + self.assertIn("x=1", result) + self.assertNotIn("secret", result) + + def test_strips_userinfo(self): + result = sanitize_url("https://user:password@example.com/gtfs.zip") + self.assertNotIn("password", result) + self.assertIn("example.com", result) + + def test_url_without_credentials_is_unchanged(self): + url = "https://example.com/gtfs.zip" + self.assertEqual(sanitize_url(url), url) + + def test_none_returns_none(self): + self.assertIsNone(sanitize_url(None)) + + +class TestGetExternalIp(unittest.TestCase): + def setUp(self): + utils._external_ip_cache = None + + def tearDown(self): + utils._external_ip_cache = None + + def test_returns_ip_on_success(self): + response = MagicMock(status=200, data=b"34.1.2.3\n") + pool = MagicMock() + pool.request.return_value = response + pool.__enter__ = lambda s: pool + pool.__exit__ = MagicMock(return_value=False) + + with patch("utils.urllib3.PoolManager", return_value=pool): + self.assertEqual(get_external_ip(), "34.1.2.3") + + def test_returns_none_when_lookup_fails(self): + with patch("utils.urllib3.PoolManager", side_effect=Exception("boom")): + self.assertIsNone(get_external_ip()) + + def test_result_is_cached(self): + response = MagicMock(status=200, data=b"34.1.2.3") + pool = MagicMock() + pool.request.return_value = response + pool.__enter__ = lambda s: pool + pool.__exit__ = MagicMock(return_value=False) + + with patch("utils.urllib3.PoolManager", return_value=pool) as mock_pool: + get_external_ip() + get_external_ip() + self.assertEqual(mock_pool.call_count, 1) + + +def _mock_response(status=200, headers=None, chunks=None): + response = MagicMock() + response.status = status + response.headers = headers if headers is not None else {} + response.retries = MagicMock(history=[]) + response.__enter__ = lambda s: response + response.__exit__ = MagicMock(return_value=False) + data = list(chunks if chunks is not None else [b""]) + + def read(_size): + return data.pop(0) if data else b"" + + response.read.side_effect = read + return response + + +def _zip_bytes() -> bytes: + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as archive: + archive.writestr("agency.txt", "agency_id,agency_name\n1,Test\n" * 50) + return buffer.getvalue() + + +def _mock_pool(response): + pool = MagicMock() + pool.request.return_value = response + pool.__enter__ = lambda s: pool + pool.__exit__ = MagicMock(return_value=False) + return pool + + +@patch("shared.common.config_reader.get_config_value", return_value=None) +class TestDownloadFeed(unittest.TestCase): + def setUp(self): + self.file_path = os.path.join( + os.path.dirname(__file__), "download_feed_test.bin" + ) + utils._external_ip_cache = None + + def tearDown(self): + if os.path.exists(self.file_path): + os.remove(self.file_path) + utils._external_ip_cache = None + + def test_diagnostics_populated_on_success(self, _mock_config): + response = _mock_response( + status=200, + headers={"Content-Type": "application/zip", "Server": "nginx"}, + chunks=[b"PK\x03\x04data", b""], + ) + with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)): + result = download_feed( + "https://example.com/gtfs.zip", self.file_path, feed_id="feed-1" + ) + + diagnostics = result.diagnostics + self.assertTrue(diagnostics.success) + self.assertEqual(diagnostics.status_code, 200) + self.assertEqual(diagnostics.content_type, "application/zip") + self.assertEqual(diagnostics.downloaded_bytes, 8) + self.assertEqual(diagnostics.response_headers["Server"], "nginx") + self.assertIn("User-Agent", diagnostics.request_headers) + self.assertIsNotNone(diagnostics.latency_ms) + self.assertIsNone(diagnostics.error_type) + self.assertIsNotNone(result.file_hash) + + def test_external_ip_not_looked_up_on_success(self, _mock_config): + response = _mock_response(chunks=[b"data", b""]) + with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)): + with patch("utils.get_external_ip") as mock_ip: + download_feed("https://example.com/gtfs.zip", self.file_path) + mock_ip.assert_not_called() + + def test_non_2xx_raises_with_diagnostics(self, _mock_config): + response = _mock_response( + status=403, headers={"Content-Type": "text/html", "Server": "cloudflare"} + ) + with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)): + with patch("utils.get_external_ip", return_value="34.1.2.3"): + with pytest.raises(FeedDownloadError) as exc_info: + download_feed("https://example.com/gtfs.zip", self.file_path) + + diagnostics = exc_info.value.diagnostics + self.assertFalse(diagnostics.success) + self.assertEqual(diagnostics.status_code, 403) + self.assertEqual(diagnostics.content_type, "text/html") + self.assertEqual(diagnostics.response_headers["Server"], "cloudflare") + self.assertEqual(diagnostics.external_ip, "34.1.2.3") + self.assertEqual(diagnostics.error_type, "ValueError") + + def test_network_error_raises_with_diagnostics(self, _mock_config): + with patch("utils.urllib3.PoolManager", side_effect=Exception("Network error")): + with patch("utils.get_external_ip", return_value=None): + with pytest.raises(FeedDownloadError) as exc_info: + download_feed("https://example.com/gtfs.zip", self.file_path) + + diagnostics = exc_info.value.diagnostics + self.assertFalse(diagnostics.success) + self.assertIsNone(diagnostics.status_code) + self.assertEqual(diagnostics.error_type, "Exception") + self.assertEqual(diagnostics.error_message, "Network error") + + def test_credentials_never_appear_in_diagnostics(self, _mock_config): + response = _mock_response(status=403) + with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)): + with patch("utils.get_external_ip", return_value=None): + with pytest.raises(FeedDownloadError) as exc_info: + download_feed( + "https://example.com/gtfs.zip", + self.file_path, + authentication_type=1, + api_key_parameter_name="token", + credentials="super-secret", + ) + + serialized = str(exc_info.value.diagnostics.as_dict()) + self.assertNotIn("super-secret", serialized) + self.assertIn(REDACTED, serialized) + + def test_max_bytes_truncates_and_returns_no_hash(self, _mock_config): + response = _mock_response(chunks=[b"a" * 10, b"a" * 10, b""]) + with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)): + result = download_feed( + "https://example.com/gtfs.zip", + self.file_path, + chunk_size=10, + max_bytes=10, + ) + + self.assertIsNone(result.file_hash) + self.assertEqual(result.diagnostics.downloaded_bytes, 10) + self.assertEqual(result.stop_reason, "max_bytes") + + def test_no_file_path_streams_without_writing(self, _mock_config): + content = _zip_bytes() + chunks = [content[i : i + 100] for i in range(0, len(content), 100)] + [b""] + response = _mock_response(chunks=chunks) + with patch( + "utils.urllib3.PoolManager", return_value=_mock_pool(response) + ), patch("builtins.open") as mock_open: + result = download_feed("https://example.com/gtfs.zip", None, chunk_size=100) + + mock_open.assert_not_called() + self.assertIsNotNone(result.file_hash) + self.assertIsNone(result.stop_reason) + self.assertTrue(result.is_zip) + self.assertEqual(result.diagnostics.downloaded_bytes, len(content)) + self.assertIsNotNone(result.duration_ms) + + def test_no_file_path_detects_non_zip(self, _mock_config): + response = _mock_response(chunks=[b"login", b""]) + with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)): + result = download_feed("https://example.com/gtfs.zip", None) + + self.assertFalse(result.is_zip) + + def test_truncated_zip_is_recognised_by_its_signature(self, _mock_config): + content = _zip_bytes() + response = _mock_response(chunks=[content[:50], content[50:], b""]) + with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)): + result = download_feed( + "https://example.com/gtfs.zip", None, chunk_size=50, max_bytes=50 + ) + + self.assertEqual(result.stop_reason, "max_bytes") + self.assertTrue(result.is_zip) + + def test_deadline_stops_reading(self, _mock_config): + response = _mock_response(chunks=[b"a" * 10] * 5 + [b""]) + clock = itertools.count(start=0, step=10) + with patch( + "utils.urllib3.PoolManager", return_value=_mock_pool(response) + ), patch("utils.time.monotonic", side_effect=lambda: next(clock)): + result = download_feed( + "https://example.com/gtfs.zip", + None, + chunk_size=10, + deadline_seconds=25, + ) + + self.assertEqual(result.stop_reason, "deadline") + self.assertIsNone(result.file_hash) + self.assertLess(result.diagnostics.downloaded_bytes, 50) + + def test_redirect_urls_are_recorded(self, _mock_config): + response = _mock_response(chunks=[b"data", b""]) + response.retries = MagicMock( + history=[MagicMock(redirect_location="https://cdn.example.com/gtfs.zip")] + ) + with patch("utils.urllib3.PoolManager", return_value=_mock_pool(response)): + result = download_feed("https://example.com/gtfs.zip", self.file_path) + + self.assertEqual( + result.diagnostics.redirect_urls, ["https://cdn.example.com/gtfs.zip"] + ) + + +class TestBuildAvailabilityCheck(unittest.TestCase): + def test_maps_diagnostics_onto_row(self): + diagnostics = FeedRequestDiagnostics( + request_url="https://example.com/gtfs.zip", + resolved_url="https://example.com/gtfs.zip", + request_headers={"User-Agent": "agent"}, + response_headers={"Server": "cloudflare"}, + status_code=403, + latency_ms=1840, + redirect_urls=["https://cdn.example.com/gtfs.zip"], + external_ip="34.1.2.3", + content_type="text/html", + is_zip=False, + error_type="ValueError", + error_message="Invalid HTTP response code: [403]", + ) + + row = build_availability_check( + diagnostics, feed_id="feed-1", source="dataset_download" + ) + + self.assertEqual(row.feed_id, "feed-1") + self.assertEqual(row.source, "dataset_download") + self.assertEqual(row.request_type, "http_get") + self.assertEqual(row.status_code, 403) + self.assertEqual(row.external_ip, "34.1.2.3") + self.assertEqual(row.response_headers, {"Server": "cloudflare"}) + self.assertFalse(row.success) + + def test_success_flag_follows_status_code(self): + diagnostics = FeedRequestDiagnostics(status_code=200) + row = build_availability_check( + diagnostics, feed_id="feed-1", source="dataset_download" + ) + self.assertTrue(row.success) + + +@patch("shared.common.config_reader.get_config_value", return_value=None) +class TestErrorMessageSanitization(unittest.TestCase): + def setUp(self): + self.file_path = os.path.join(os.path.dirname(__file__), "err_test.bin") + utils._external_ip_cache = None + + def tearDown(self): + if os.path.exists(self.file_path): + os.remove(self.file_path) + utils._external_ip_cache = None + + def test_userinfo_in_producer_url_is_redacted_in_error(self, _mock_config): + url = "https://user:password@example.com/gtfs.zip" + with patch( + "utils.urllib3.PoolManager", side_effect=Exception(f"failed for {url}") + ): + with patch("utils.get_external_ip", return_value=None): + with pytest.raises(FeedDownloadError) as exc_info: + download_feed(url, self.file_path) + + self.assertNotIn("password", exc_info.value.diagnostics.error_message) + + def test_credentials_in_resolved_url_are_redacted_in_error(self, _mock_config): + def boom(*_args, **_kwargs): + raise Exception( + "failed for https://example.com/gtfs.zip?token=super-secret" + ) + + with patch("utils.urllib3.PoolManager", side_effect=boom): + with patch("utils.get_external_ip", return_value=None): + with pytest.raises(FeedDownloadError) as exc_info: + download_feed( + "https://example.com/gtfs.zip", + self.file_path, + authentication_type=1, + api_key_parameter_name="token", + credentials="super-secret", + ) + + self.assertNotIn("super-secret", exc_info.value.diagnostics.error_message) diff --git a/functions-python/helpers/tests/test_feed_http.py b/functions-python/helpers/tests/test_feed_http.py index 0d8228c08..5fd49c680 100644 --- a/functions-python/helpers/tests/test_feed_http.py +++ b/functions-python/helpers/tests/test_feed_http.py @@ -113,6 +113,28 @@ def test_custom_headers_from_config_override_default(self, _): self.assertEqual(headers["User-Agent"], "CustomAgent/1.0") self.assertNotIn("Referer", headers) + @patch("shared.common.config_reader.get_config_value") + def test_explicit_headers_skip_config_and_default(self, mock_get_config): + explicit = {"User-Agent": "curl/8.0"} + headers, _ = build_feed_request_params( + "http://example.com/feed.zip", feed_id="feed-1", headers=explicit + ) + self.assertEqual(headers, {"User-Agent": "curl/8.0"}) + mock_get_config.assert_not_called() + + @patch("shared.common.config_reader.get_config_value", return_value=None) + def test_explicit_headers_are_not_mutated_by_auth(self, _): + explicit = {"User-Agent": "curl/8.0"} + headers, _ = build_feed_request_params( + "http://example.com/feed.zip", + authentication_type="2", + api_key_parameter_name="X-API-Key", + credentials="token123", + headers=explicit, + ) + self.assertEqual(headers["X-API-Key"], "token123") + self.assertEqual(explicit, {"User-Agent": "curl/8.0"}) + @patch("shared.common.config_reader.get_config_value", return_value=None) def test_none_authentication_type_treated_as_zero(self, _): headers, url = build_feed_request_params( diff --git a/functions-python/helpers/tests/test_pub_sub_dry_run.py b/functions-python/helpers/tests/test_pub_sub_dry_run.py new file mode 100644 index 000000000..b794d8237 --- /dev/null +++ b/functions-python/helpers/tests/test_pub_sub_dry_run.py @@ -0,0 +1,53 @@ +import json +import unittest +from unittest.mock import MagicMock, patch + +from pub_sub import trigger_dataset_download + + +def _feed(): + feed = MagicMock() + feed.producer_url = "https://example.com/gtfs.zip" + feed.stable_id = "mdb-1" + feed.id = "feed-1" + feed.authentication_type = 0 + feed.authentication_info_url = None + feed.api_key_parameter_name = None + return feed + + +def _published_message(**kwargs): + publisher = MagicMock() + with patch("pub_sub.get_pubsub_client", return_value=publisher): + trigger_dataset_download(_feed(), "exec-1", topic_name="topic", **kwargs) + return json.loads(publisher.publish.call_args.kwargs["data"].decode()) + + +class TestTriggerDatasetDownload(unittest.TestCase): + def test_nightly_message_is_unchanged(self): + message = _published_message() + self.assertNotIn("dry_run", message) + self.assertNotIn("headers", message) + self.assertNotIn("record_source", message) + + def test_dry_run_message_carries_flag_and_headers(self): + message = _published_message(dry_run=True, headers={"User-Agent": "curl/8.0"}) + self.assertTrue(message["dry_run"]) + self.assertEqual(message["headers"], {"User-Agent": "curl/8.0"}) + + def test_record_source_is_sent_when_set(self): + message = _published_message(record_source="dataset_download_manual") + self.assertEqual(message["record_source"], "dataset_download_manual") + self.assertNotIn("dry_run", message) + + def test_latest_dataset_is_sent_when_known(self): + message = _published_message( + dataset_stable_id="mdb-1-202610200000", dataset_hash="abc123" + ) + self.assertEqual(message["dataset_stable_id"], "mdb-1-202610200000") + self.assertEqual(message["dataset_hash"], "abc123") + + def test_dataset_hash_stays_null_for_a_new_feed(self): + message = _published_message() + self.assertIsNone(message["dataset_hash"]) + self.assertNotIn("dataset_stable_id", message) diff --git a/functions-python/helpers/utils.py b/functions-python/helpers/utils.py index 8d9f53dc3..16ba4c466 100644 --- a/functions-python/helpers/utils.py +++ b/functions-python/helpers/utils.py @@ -13,15 +13,19 @@ # See the License for the specific language governing permissions and # limitations under the License. # +import contextlib import hashlib import logging import os import ssl import time +import zipfile import urllib3.exceptions +from dataclasses import asdict, dataclass from datetime import date, datetime, timezone from logging import Logger from typing import Optional, TYPE_CHECKING +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit if TYPE_CHECKING: # pragma: no cover from shared.helpers.sizing import Size @@ -145,17 +149,53 @@ def create_feed_ssl_context(trusted_certs: bool = False): return ctx +def get_feed_credentials(stable_id: str) -> Optional[str]: + """Return the API credential for a feed from the FEEDS_CREDENTIALS env var, or None.""" + try: + import json + + feeds_credentials = json.loads(os.getenv("FEEDS_CREDENTIALS", "{}")) + return feeds_credentials.get(stable_id, None) + except Exception as exc: + logging.warning("Could not parse FEEDS_CREDENTIALS: %s", exc) + return None + + +def _get_http_headers_override(feed_id: Optional[str]) -> Optional[dict]: + """Return the feed_download/http_headers config override, or None. + + Returns None when the config database is unreachable so that callers fall back to + the default headers instead of failing. + """ + try: + from shared.common.config_reader import get_config_value + + return get_config_value( + namespace="feed_download", key="http_headers", feed_id=feed_id + ) + except Exception as exc: + logging.warning( + "Could not read feed_download/http_headers for feed %s, " + "falling back to default headers: %s", + feed_id, + exc, + ) + return None + + def build_feed_request_params( url: str, feed_id: Optional[str] = None, authentication_type=0, api_key_parameter_name: Optional[str] = None, credentials: Optional[str] = None, + headers: Optional[dict] = None, ) -> tuple: """ Build HTTP request headers and resolve the final URL for a feed request. Handles: + - Explicit headers passed by the caller, which skip the config DB lookup - Per-feed User-Agent overrides via config DB (feed_download/http_headers) - Default mobile browser User-Agent + Referer fallback - Auth type 1: API key appended as a URL query parameter @@ -164,11 +204,10 @@ def build_feed_request_params( Returns: (headers, resolved_url) ready to pass to any HTTP method. """ - from shared.common.config_reader import get_config_value - - headers = get_config_value( - namespace="feed_download", key="http_headers", feed_id=feed_id - ) + if headers is not None: + headers = dict(headers) + else: + headers = _get_http_headers_override(feed_id) if headers is None: headers = { "User-Agent": "Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) " @@ -252,6 +291,232 @@ def _sanitize_error_message( return message +REDACTED = "***REDACTED***" + +_SECRET_HEADER_NAMES = frozenset( + { + "authorization", + "proxy-authorization", + "cookie", + "set-cookie", + "x-api-key", + "api-key", + "apikey", + "token", + "access-token", + "x-access-token", + "x-auth-token", + } +) + + +def _is_secret_name(name: str, api_key_parameter_name: Optional[str] = None) -> bool: + """Return True when a header or query parameter name carries a credential. + + Underscores and hyphens are treated as equivalent so that both api_key and api-key + are matched. + """ + lowered = name.strip().lower() + if lowered.replace("_", "-") in _SECRET_HEADER_NAMES: + return True + if not api_key_parameter_name: + return False + return lowered.replace("_", "-") == api_key_parameter_name.strip().lower().replace( + "_", "-" + ) + + +def sanitize_headers( + headers: Optional[dict], api_key_parameter_name: Optional[str] = None +) -> Optional[dict]: + """Return a copy of headers with credential values replaced by REDACTED. + + Names are always preserved, only values are redacted. + """ + if headers is None: + return None + return { + name: ( + REDACTED if _is_secret_name(name, api_key_parameter_name) else str(value) + ) + for name, value in headers.items() + } + + +def sanitize_url( + url: Optional[str], api_key_parameter_name: Optional[str] = None +) -> Optional[str]: + """Return url with credential query-parameter values and userinfo redacted.""" + if not url: + return url + try: + parsed = urlsplit(url) + except ValueError: + return REDACTED + + netloc = parsed.netloc + if "@" in netloc: + netloc = f"{REDACTED}@{netloc.rsplit('@', 1)[1]}" + + query = parsed.query + if query: + query = urlencode( + [ + ( + name, + ( + REDACTED + if _is_secret_name(name, api_key_parameter_name) + else value + ), + ) + for name, value in parse_qsl(query, keep_blank_values=True) + ], + safe="*", + ) + + return urlunsplit((parsed.scheme, netloc, parsed.path, query, parsed.fragment)) + + +_EXTERNAL_IP_URL = "https://checkip.amazonaws.com" +_external_ip_cache: Optional[str] = None + + +def get_external_ip(timeout_seconds: int = 3) -> Optional[str]: + """Return the egress IP address of this process, or None if it cannot be determined. + + The result is cached for the lifetime of the process. Never raises. + """ + global _external_ip_cache + if _external_ip_cache is not None: + return _external_ip_cache + try: + with urllib3.PoolManager() as http: + response = http.request( + "GET", + _EXTERNAL_IP_URL, + timeout=urllib3.Timeout(connect=timeout_seconds, read=timeout_seconds), + retries=False, + ) + if response.status == 200: + _external_ip_cache = response.data.decode("utf-8").strip() or None + except Exception as exc: + logging.warning("Could not determine external IP address: %s", exc) + return _external_ip_cache + + +@dataclass +class FeedRequestDiagnostics: + """Request and response detail for a feed download attempt, safe to log and store.""" + + request_url: Optional[str] = None + resolved_url: Optional[str] = None + request_headers: Optional[dict] = None + response_headers: Optional[dict] = None + status_code: Optional[int] = None + latency_ms: Optional[int] = None + redirect_urls: Optional[list] = None + external_ip: Optional[str] = None + content_type: Optional[str] = None + is_zip: Optional[bool] = None + downloaded_bytes: Optional[int] = None + error_type: Optional[str] = None + error_message: Optional[str] = None + + @property + def success(self) -> bool: + return ( + self.error_type is None + and self.status_code is not None + and 200 <= self.status_code < 300 + ) + + def as_dict(self) -> dict: + return asdict(self) + + +class FeedDownloadError(Exception): + """Raised when a feed download fails, carrying the diagnostics of the attempt.""" + + def __init__(self, message: str, diagnostics: FeedRequestDiagnostics): + super().__init__(message) + self.diagnostics = diagnostics + + +@dataclass +class FeedDownloadResult: + """Outcome of a successful feed download.""" + + file_hash: Optional[str] + is_zip: Optional[bool] + diagnostics: FeedRequestDiagnostics + # Why reading stopped before the end of the file: "max_bytes" or "deadline". + stop_reason: Optional[str] = None + # Wall time from sending the request to the last byte read. + duration_ms: Optional[int] = None + + +# The ZIP end-of-central-directory record sits in the last 22 bytes plus at most a +# 64 KiB comment, so this much tail is enough to tell a complete ZIP without the file. +_ZIP_EOCD_SIGNATURE = b"PK\x05\x06" +_ZIP_TAIL_BYTES = 65536 + 22 + + +class _ZipSniffer: + """Tracks the first and last bytes of a stream to recognise a ZIP without storing it.""" + + def __init__(self): + self.head = b"" + self.tail = bytearray() + + def update(self, data: bytes) -> None: + if len(self.head) < len(_ZIP_MAGIC): + self.head += data[: len(_ZIP_MAGIC) - len(self.head)] + self.tail += data + if len(self.tail) > _ZIP_TAIL_BYTES: + del self.tail[: len(self.tail) - _ZIP_TAIL_BYTES] + + def is_zip(self, complete: bool) -> bool: + """A truncated stream can only be judged by its leading signature.""" + if self.head != _ZIP_MAGIC: + return False + return not complete or _ZIP_EOCD_SIGNATURE in self.tail + + +def build_availability_check( + diagnostics: FeedRequestDiagnostics, + feed_id: str, + source: str, + request_type: str = "http_get", + checked_at: Optional[datetime] = None, +): + """Map diagnostics onto a GtfsFeedAvailabilityCheck row. + + source is an AvailabilityCheckSource value. + """ + from shared.database_gen.sqlacodegen_models import GtfsFeedAvailabilityCheck + + return GtfsFeedAvailabilityCheck( + feed_id=feed_id, + checked_at=checked_at or datetime.now(timezone.utc), + request_url=diagnostics.request_url, + resolved_url=diagnostics.resolved_url, + request_type=request_type, + status_code=diagnostics.status_code, + latency_ms=diagnostics.latency_ms, + error_message=diagnostics.error_message, + error_type=diagnostics.error_type, + success=diagnostics.success, + content_type=diagnostics.content_type, + is_zip=diagnostics.is_zip, + source=source, + request_headers=diagnostics.request_headers, + response_headers=diagnostics.response_headers, + redirect_urls=diagnostics.redirect_urls, + external_ip=diagnostics.external_ip, + ) + + def _execute_http_request( method: str, url: str, @@ -329,6 +594,7 @@ def perform_request( Note: request_url is always the original producer_url (never the credential-bearing resolved URL) to avoid persisting secrets. """ + from shared.common.availability_check_source import AvailabilityCheckSource from shared.database_gen.sqlacodegen_models import GtfsFeedAvailabilityCheck checked_at = datetime.now(timezone.utc) @@ -413,6 +679,7 @@ def perform_request( feed_id=feed_id, checked_at=checked_at, request_url=producer_url, + resolved_url=sanitize_url(resolved_url, api_key_parameter_name), request_type=request_type, status_code=status_code, latency_ms=latency_ms, @@ -421,12 +688,18 @@ def perform_request( success=success, content_type=content_type, is_zip=is_zip, + source=AvailabilityCheckSource.AVAILABILITY_CHECK.value, + request_headers=sanitize_headers(headers, api_key_parameter_name), + response_headers=sanitize_headers(dict(resp_headers) if resp_headers else None), + redirect_urls=[ + sanitize_url(u, api_key_parameter_name) for u in (redirect_urls or []) + ], ) -def download_and_get_hash( +def download_feed( url, - file_path, + file_path: Optional[str], hash_algorithm="sha256", chunk_size=8192, feed_id=None, @@ -435,48 +708,172 @@ def download_and_get_hash( credentials=None, logger=None, trusted_certs=False, # If True, disables SSL verification -): - """ - Downloads the content of a URL and stores it in a file and returns the hash of the file + max_bytes: Optional[int] = None, + timeout_seconds: Optional[int] = None, + headers: Optional[dict] = None, + deadline_seconds: Optional[float] = None, +) -> FeedDownloadResult: + """Download a feed to file_path and return its hash along with request diagnostics. + + file_path None streams and hashes the download without storing it, so a feed of any + size can be checked without the disk (in-memory on Cloud Run) holding it. + + max_bytes stops the download once that many bytes have been read, and + deadline_seconds once that much time has passed, which lets callers inspect the + response without the whole dataset. The returned hash is None in either case because + the file is incomplete, and stop_reason says which limit was hit. + + headers, when given, replaces the default and config DB request headers. + + Raises FeedDownloadError carrying a FeedRequestDiagnostics on any failure. """ logger = logger or logging.getLogger(__name__) - try: - hash_object = hashlib.new(hash_algorithm) + producer_url = url + resolved_url = url + hash_object = hashlib.new(hash_algorithm) + diagnostics = FeedRequestDiagnostics( + request_url=sanitize_url(producer_url, api_key_parameter_name) + ) + stop_reason = None + sniffer = _ZipSniffer() + duration_ms = None + try: ctx = create_feed_ssl_context(trusted_certs=trusted_certs) - headers, url = build_feed_request_params( - url, + headers, resolved_url = build_feed_request_params( + producer_url, feed_id=feed_id, authentication_type=authentication_type, api_key_parameter_name=api_key_parameter_name, credentials=credentials, + headers=headers, + ) + diagnostics.resolved_url = sanitize_url(resolved_url, api_key_parameter_name) + diagnostics.request_headers = sanitize_headers(headers, api_key_parameter_name) + + timeout = ( + urllib3.Timeout(connect=timeout_seconds, read=timeout_seconds) + if timeout_seconds + else None ) + request_kwargs = {"timeout": timeout} if timeout else {} with urllib3.PoolManager(ssl_context=ctx) as http: + start = time.monotonic() with http.request( - "GET", url, preload_content=False, headers=headers, redirect=True - ) as r, open(file_path, "wb") as out_file: - if 200 <= r.status < 300: - logger.info(f"HTTP response code: [{r.status}]") - while True: - data = r.read(chunk_size) - if not data: - break - hash_object.update(data) - out_file.write(data) - r.release_conn() - else: + "GET", + resolved_url, + preload_content=False, + headers=headers, + redirect=True, + **request_kwargs, + ) as r, ( + open(file_path, "wb") if file_path else contextlib.nullcontext() + ) as out_file: + diagnostics.latency_ms = int((time.monotonic() - start) * 1000) + diagnostics.status_code = r.status + diagnostics.response_headers = sanitize_headers(dict(r.headers)) + diagnostics.redirect_urls = [ + sanitize_url(h.redirect_location, api_key_parameter_name) + for h in (r.retries.history or []) + if h.redirect_location + ] + diagnostics.content_type = _parse_content_type( + r.headers.get("Content-Type") + ) + + if not 200 <= r.status < 300: raise ValueError(f"Invalid HTTP response code: [{r.status}]") - return hash_object.hexdigest() - except Exception as e: - if os.path.exists(file_path): + + logger.info(f"HTTP response code: [{r.status}]") + total = 0 + while True: + data = r.read(chunk_size) + if not data: + break + hash_object.update(data) + sniffer.update(data) + if out_file is not None: + out_file.write(data) + total += len(data) + if max_bytes is not None and total >= max_bytes: + stop_reason = "max_bytes" + break + if ( + deadline_seconds is not None + and time.monotonic() - start >= deadline_seconds + ): + stop_reason = "deadline" + break + r.release_conn() + duration_ms = int((time.monotonic() - start) * 1000) + diagnostics.downloaded_bytes = total + + complete = stop_reason is None + if file_path and complete: + is_zip = zipfile.is_zipfile(file_path) + else: + is_zip = sniffer.is_zip(complete) + diagnostics.is_zip = is_zip or ( + _is_zip_from_content_type(diagnostics.content_type) or False + ) + return FeedDownloadResult( + file_hash=hash_object.hexdigest() if complete else None, + is_zip=diagnostics.is_zip, + diagnostics=diagnostics, + stop_reason=stop_reason, + duration_ms=duration_ms, + ) + except Exception as exc: + diagnostics.error_type = type(exc).__name__ + # Collapse the credential-bearing resolved URL onto the producer URL, then the + # producer URL onto its sanitized form, so neither can leak through the message. + message = _sanitize_error_message(str(exc), resolved_url, producer_url) + if message and producer_url != diagnostics.request_url: + message = message.replace(producer_url, diagnostics.request_url) + diagnostics.error_message = message + diagnostics.external_ip = get_external_ip() + logger.warning( + "Feed download failed for %s: %s", + diagnostics.request_url, + diagnostics.as_dict(), + ) + if file_path and os.path.exists(file_path): try: os.remove(file_path) except Exception: logger.error(f"Delete file: [{file_path}]") + raise FeedDownloadError(str(exc), diagnostics) from exc - raise e + +def download_and_get_hash( + url, + file_path, + hash_algorithm="sha256", + chunk_size=8192, + feed_id=None, + authentication_type=0, + api_key_parameter_name=None, + credentials=None, + logger=None, + trusted_certs=False, # If True, disables SSL verification +): + """ + Downloads the content of a URL and stores it in a file and returns the hash of the file + """ + return download_feed( + url, + file_path, + hash_algorithm=hash_algorithm, + chunk_size=chunk_size, + feed_id=feed_id, + authentication_type=authentication_type, + api_key_parameter_name=api_key_parameter_name, + credentials=credentials, + logger=logger, + trusted_certs=trusted_certs, + ).file_hash def create_http_task( diff --git a/functions-python/operations_api/.openapi-generator/FILES b/functions-python/operations_api/.openapi-generator/FILES index c30bf8a8e..730dd4650 100644 --- a/functions-python/operations_api/.openapi-generator/FILES +++ b/functions-python/operations_api/.openapi-generator/FILES @@ -49,6 +49,9 @@ src/feeds_gen/models/gtfs_feed_continuous_coverage.py src/feeds_gen/models/gtfs_feed_continuous_coverage_boundary.py src/feeds_gen/models/gtfs_feed_continuous_coverage_file.py src/feeds_gen/models/gtfs_feed_continuous_coverage_response.py +src/feeds_gen/models/gtfs_feed_download_debug_request.py +src/feeds_gen/models/gtfs_feed_download_debug_response.py +src/feeds_gen/models/gtfs_feed_download_queued_response.py src/feeds_gen/models/gtfs_feed_validation_notice.py src/feeds_gen/models/gtfs_feed_validation_report.py src/feeds_gen/models/gtfs_feed_validation_reports_response.py diff --git a/functions-python/operations_api/src/feeds_operations/impl/feeds_operations_impl.py b/functions-python/operations_api/src/feeds_operations/impl/feeds_operations_impl.py index 31c462abf..3d2f7353f 100644 --- a/functions-python/operations_api/src/feeds_operations/impl/feeds_operations_impl.py +++ b/functions-python/operations_api/src/feeds_operations/impl/feeds_operations_impl.py @@ -18,8 +18,9 @@ from fastapi.responses import JSONResponse import logging -from datetime import datetime -from typing import Annotated, Final, Optional +import re +from datetime import datetime, timezone +from typing import Annotated, Final, List, Optional from deepdiff import DeepDiff from fastapi import HTTPException @@ -35,6 +36,15 @@ from feeds_gen.models.gtfs_feed_availability_response import ( GtfsFeedAvailabilityResponse, ) +from feeds_gen.models.gtfs_feed_download_debug_request import ( + GtfsFeedDownloadDebugRequest, +) +from feeds_gen.models.gtfs_feed_download_debug_response import ( + GtfsFeedDownloadDebugResponse, +) +from feeds_gen.models.gtfs_feed_download_queued_response import ( + GtfsFeedDownloadQueuedResponse, +) from feeds_gen.models.operation_create_request_gtfs_feed import ( OperationCreateRequestGtfsFeed, ) @@ -65,6 +75,7 @@ Gtfsrealtimefeed, TaskExecutionAttempt, ) +from shared.common.availability_check_source import AvailabilityCheckSource from shared.common.config_reader import get_config_value from shared.helpers.parquet_policy import SIZE_CONFIG_KEY, SIZE_CONFIG_NAMESPACE from shared.common.license_utils import assign_license_by_url, propagate_license_by_url @@ -72,7 +83,16 @@ from shared.db_models.gtfs_feed_availability_check_impl import ( GtfsFeedAvailabilityCheckImpl, ) -from shared.helpers.pub_sub import get_execution_id, trigger_dataset_download +from shared.helpers.pub_sub import ( + DATASET_BATCH_TOPIC, + get_execution_id, + trigger_dataset_download, +) +from shared.helpers.utils import ( + FeedDownloadError, + download_feed, + get_feed_credentials, +) from shared.helpers.query_helper import ( query_feed_by_stable_id, get_feeds_query, @@ -200,6 +220,73 @@ def _execution_attempt(row) -> ExecutionAttempt: ) +# Availability-check source batch_process_dataset records runs asked for by hand under, +# dry or full. +DOWNLOAD_MANUAL_SOURCE: Final = AvailabilityCheckSource.DATASET_DOWNLOAD_MANUAL.value + +# Stop reading well inside the function's 540 second timeout, so a slow download still +# returns its diagnostics instead of the request being cut off with nothing. +_DOWNLOAD_DEBUG_DEADLINE_SECONDS: Final = 480 + +# RFC 9110 field-name token. Values only need to be free of line breaks, which would +# let a caller smuggle extra headers into the request. +_HEADER_NAME_RE: Final = re.compile(r"^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$") + + +def _validate_debug_headers(headers: Optional[dict]) -> None: + """Reject custom download headers that requests would refuse mid-download.""" + for name, value in (headers or {}).items(): + if not _HEADER_NAME_RE.match(name): + raise HTTPException(status_code=400, detail=f"Invalid header name {name!r}") + if "\r" in value or "\n" in value: + raise HTTPException( + status_code=400, detail=f"Header {name!r} contains a line break" + ) + + +def _queue_nightly_download( + gtfs_feed, + headers: Optional[dict], + dry_run: bool, + latest_dataset: Optional[Gtfsdataset] = None, +) -> JSONResponse: + """Queue one run on batch_process_dataset and say where its download outcome lands. + + A dry run only downloads. A full run is the nightly process itself for this feed, + with the download outcome recorded even on success so the caller can read it back. + """ + # trigger_dataset_download silently skips without a topic; say so instead of + # leaving the caller polling for a result that will never come. + if not DATASET_BATCH_TOPIC: + raise HTTPException( + status_code=503, + detail="Dataset processing is not configured in this environment", + ) + prefix = "download-dry-run" if dry_run else "download-manual" + execution_id = get_execution_id(get_request_context(), prefix) + queued_at = datetime.now(timezone.utc) + if dry_run: + trigger_dataset_download(gtfs_feed, execution_id, dry_run=True, headers=headers) + else: + # The latest dataset's hash, as the nightly scheduler sends it, so an unchanged + # file is recognised and no new dataset is stored. + trigger_dataset_download( + gtfs_feed, + execution_id, + record_source=DOWNLOAD_MANUAL_SOURCE, + dataset_stable_id=latest_dataset.stable_id if latest_dataset else None, + dataset_hash=latest_dataset.hash if latest_dataset else None, + ) + payload = GtfsFeedDownloadQueuedResponse( + feed_id=gtfs_feed.stable_id, + execution_id=execution_id, + queued_at=queued_at, + source=DOWNLOAD_MANUAL_SOURCE, + dry_run=dry_run, + ) + return JSONResponse(status_code=202, content=jsonable_encoder(payload)) + + class OperationsApiImpl(BaseOperationsApi): """Implementation of the operations API.""" @@ -327,12 +414,24 @@ def get_gtfs_feed_availability( ], var_from: Optional[datetime] = None, to: Optional[datetime] = None, + source: Optional[List[str]] = None, limit: Optional[int] = 100, offset: Optional[int] = 0, sort: Optional[str] = "desc", db_session: Session = None, ) -> GtfsFeedAvailabilityResponse: - """Get availability check history for the specified GTFS feed.""" + """Get availability check history for the specified GTFS feed. + + Unlike the public endpoint, every source is returned unless `source` narrows it. + """ + known_sources = {s.value for s in AvailabilityCheckSource} + unknown_sources = sorted(set(source or []) - known_sources) + if unknown_sources: + raise HTTPException( + status_code=400, + detail=f"Invalid source {', '.join(unknown_sources)}; " + f"expected one of {', '.join(sorted(known_sources))}", + ) gtfs_feed = ( db_session.query(Gtfsfeed).filter(Gtfsfeed.stable_id == id).one_or_none() ) @@ -346,6 +445,8 @@ def get_gtfs_feed_availability( query = query.filter(GtfsFeedAvailabilityCheck.checked_at >= var_from) if to is not None: query = query.filter(GtfsFeedAvailabilityCheck.checked_at <= to) + if source: + query = query.filter(GtfsFeedAvailabilityCheck.source.in_(source)) total = query.count() order = ( @@ -432,6 +533,99 @@ def get_gtfs_feed_execution_attempts( attempts=[_execution_attempt(row) for row in attempts], ) + @with_db_session + def debug_gtfs_feed_download( + self, + id: Annotated[ + StrictStr, Field(description="The feed ID of the requested feed.") + ], + max_bytes: Optional[int] = None, + timeout_seconds: Optional[int] = None, + gtfs_feed_download_debug_request: Optional[GtfsFeedDownloadDebugRequest] = None, + db_session: Session = None, + ) -> GtfsFeedDownloadDebugResponse: + """Download the feed through the shared download path, here or on the nightly job.""" + request = gtfs_feed_download_debug_request + headers = request.headers if request else None + run_on = (request.run_on if request else None) or "api" + dry_run = request.dry_run if request and request.dry_run is not None else True + _validate_debug_headers(headers) + if not dry_run and run_on != "nightly_job": + raise HTTPException( + status_code=400, detail="dry_run=false requires run_on=nightly_job" + ) + if not dry_run and headers is not None: + raise HTTPException( + status_code=400, + detail="headers are only allowed on dry runs, so a stored dataset " + "always comes from the nightly request", + ) + if run_on == "nightly_job" and ( + max_bytes is not None or timeout_seconds is not None + ): + raise HTTPException( + status_code=400, + detail="max_bytes and timeout_seconds apply only to run_on=api", + ) + gtfs_feed = ( + db_session.query(Gtfsfeed).filter(Gtfsfeed.stable_id == id).one_or_none() + ) + if gtfs_feed is None: + raise HTTPException(status_code=404, detail="GTFS feed not found") + if not gtfs_feed.producer_url: + raise HTTPException(status_code=400, detail="GTFS feed has no producer URL") + if run_on == "nightly_job": + latest_dataset = ( + db_session.get(Gtfsdataset, gtfs_feed.latest_dataset_id) + if gtfs_feed.latest_dataset_id + else None + ) + return _queue_nightly_download(gtfs_feed, headers, dry_run, latest_dataset) + + # Streamed and hashed, never written: Cloud Run's disk is memory, and a full-size + # feed would otherwise exhaust the instance serving every other request. + stop_reason = None + duration_ms = None + try: + result = download_feed( + gtfs_feed.producer_url, + file_path=None, + feed_id=gtfs_feed.id, + authentication_type=gtfs_feed.authentication_type or 0, + api_key_parameter_name=gtfs_feed.api_key_parameter_name, + credentials=get_feed_credentials(gtfs_feed.stable_id), + max_bytes=max_bytes, + timeout_seconds=timeout_seconds, + headers=headers, + deadline_seconds=_DOWNLOAD_DEBUG_DEADLINE_SECONDS, + ) + diagnostics = result.diagnostics + stop_reason = result.stop_reason + duration_ms = result.duration_ms + except FeedDownloadError as exc: + diagnostics = exc.diagnostics + + return GtfsFeedDownloadDebugResponse( + feed_id=gtfs_feed.stable_id, + producer_url=diagnostics.request_url, + resolved_url=diagnostics.resolved_url, + success=diagnostics.success, + status_code=diagnostics.status_code, + latency_ms=diagnostics.latency_ms, + request_headers=diagnostics.request_headers, + response_headers=diagnostics.response_headers, + redirect_urls=diagnostics.redirect_urls, + external_ip=diagnostics.external_ip, + content_type=diagnostics.content_type, + is_zip=diagnostics.is_zip, + downloaded_bytes=diagnostics.downloaded_bytes, + truncated=stop_reason is not None, + stop_reason=stop_reason, + duration_ms=duration_ms, + error_type=diagnostics.error_type, + error_message=diagnostics.error_message, + ) + @with_db_session def get_gtfs_rt_feed( self, diff --git a/functions-python/operations_api/tests/conftest.py b/functions-python/operations_api/tests/conftest.py index 8ad40386e..4e6c9e061 100644 --- a/functions-python/operations_api/tests/conftest.py +++ b/functions-python/operations_api/tests/conftest.py @@ -76,6 +76,7 @@ status_code=200, latency_ms=120, error_type=None, + source="availability_check", ), GtfsFeedAvailabilityCheck( id=uuid.UUID("11111111-0000-0000-0000-000000000002"), @@ -87,6 +88,15 @@ status_code=503, latency_ms=450, error_type="http_error", + source="dataset_download", + error_message="Invalid HTTP response code: [503]", + content_type="text/html", + is_zip=False, + resolved_url="http://example.com/feed.zip", + request_headers={"User-Agent": "agent"}, + response_headers={"Server": "nginx"}, + redirect_urls=["http://cdn.example.com/feed.zip"], + external_ip="34.1.2.3", ), GtfsFeedAvailabilityCheck( id=uuid.UUID("11111111-0000-0000-0000-000000000003"), diff --git a/functions-python/operations_api/tests/feeds_operations/impl/test_availability_operations_impl.py b/functions-python/operations_api/tests/feeds_operations/impl/test_availability_operations_impl.py index 6e48f0cc9..87912031b 100644 --- a/functions-python/operations_api/tests/feeds_operations/impl/test_availability_operations_impl.py +++ b/functions-python/operations_api/tests/feeds_operations/impl/test_availability_operations_impl.py @@ -140,3 +140,81 @@ async def test_get_gtfs_feed_availability_latency_is_float(): for check in response.checks: if check.latency_ms is not None: assert isinstance(check.latency_ms, float) + + +@pytest.mark.asyncio +async def test_get_gtfs_feed_availability_filter_source(): + """source returns only the checks written by that source.""" + api = OperationsApiImpl() + response = api.get_gtfs_feed_availability( + id=feed_mdb_40.stable_id, source=["dataset_download"], limit=100, offset=0 + ) + assert response.total == 1 + assert [c.source for c in response.checks] == ["dataset_download"] + + +@pytest.mark.asyncio +async def test_get_gtfs_feed_availability_filter_several_sources(): + """Repeating source returns the checks of every source given.""" + api = OperationsApiImpl() + response = api.get_gtfs_feed_availability( + id=feed_mdb_40.stable_id, + source=["dataset_download", "dataset_download_manual"], + limit=100, + offset=0, + ) + assert [c.source for c in response.checks] == ["dataset_download"] + + response = api.get_gtfs_feed_availability( + id=feed_mdb_40.stable_id, + source=["availability_check", "dataset_download"], + limit=100, + offset=0, + ) + assert response.total == len(availability_checks_mdb_40) + + +@pytest.mark.asyncio +async def test_get_gtfs_feed_availability_without_source_returns_every_source(): + """Unlike the public endpoint, download diagnostics are included by default.""" + api = OperationsApiImpl() + response = api.get_gtfs_feed_availability( + id=feed_mdb_40.stable_id, limit=100, offset=0 + ) + sources = [c.source for c in response.checks] + assert response.total == len(availability_checks_mdb_40) + # Rows written without a source get the column default. + assert sources.count("availability_check") == len(availability_checks_mdb_40) - 1 + assert sources.count("dataset_download") == 1 + + +@pytest.mark.asyncio +async def test_get_gtfs_feed_availability_rejects_unknown_source(): + """An unknown source is a 400 naming it, not an empty page.""" + api = OperationsApiImpl() + with pytest.raises(HTTPException) as exc_info: + api.get_gtfs_feed_availability( + id=feed_mdb_40.stable_id, + source=["dataset_download", "nightly"], + limit=100, + offset=0, + ) + assert exc_info.value.status_code == 400 + assert "nightly" in exc_info.value.detail + + +@pytest.mark.asyncio +async def test_get_gtfs_feed_availability_returns_diagnostics(): + """Each check carries the request/response diagnostics recorded with it.""" + api = OperationsApiImpl() + response = api.get_gtfs_feed_availability( + id=feed_mdb_40.stable_id, source=["dataset_download"], limit=100, offset=0 + ) + check = response.checks[0] + assert check.error_message == "Invalid HTTP response code: [503]" + assert check.content_type == "text/html" + assert check.is_zip is False + assert check.request_headers == {"User-Agent": "agent"} + assert check.response_headers == {"Server": "nginx"} + assert check.redirect_urls == ["http://cdn.example.com/feed.zip"] + assert check.external_ip == "34.1.2.3" diff --git a/functions-python/operations_api/tests/feeds_operations/impl/test_download_debug_operations_impl.py b/functions-python/operations_api/tests/feeds_operations/impl/test_download_debug_operations_impl.py new file mode 100644 index 000000000..43f2b8704 --- /dev/null +++ b/functions-python/operations_api/tests/feeds_operations/impl/test_download_debug_operations_impl.py @@ -0,0 +1,395 @@ +import json +from unittest.mock import MagicMock, patch + +import pytest +from fastapi import HTTPException + +from conftest import feed_mdb_40 +from feeds_gen.models.gtfs_feed_download_debug_request import ( + GtfsFeedDownloadDebugRequest, +) +from feeds_operations.impl.feeds_operations_impl import ( + OperationsApiImpl, + _queue_nightly_download, +) +from shared.database.database import with_db_session +from shared.database_gen.sqlacodegen_models import GtfsFeedAvailabilityCheck +from shared.helpers.utils import FeedDownloadError, FeedDownloadResult +from shared.helpers.utils import FeedRequestDiagnostics + + +def _diagnostics(**overrides): + defaults = dict( + request_url="https://example.com/gtfs.zip", + resolved_url="https://example.com/gtfs.zip", + request_headers={"User-Agent": "agent"}, + response_headers={"Server": "cloudflare"}, + status_code=200, + latency_ms=120, + redirect_urls=[], + content_type="application/zip", + is_zip=True, + downloaded_bytes=2038, + ) + defaults.update(overrides) + return FeedRequestDiagnostics(**defaults) + + +@pytest.mark.asyncio +async def test_download_debug_success(): + """Returns the diagnostics of a successful download.""" + api = OperationsApiImpl() + diagnostics = _diagnostics() + result = FeedDownloadResult( + file_hash="abc123", is_zip=True, diagnostics=diagnostics + ) + + with patch( + "feeds_operations.impl.feeds_operations_impl.download_feed", + return_value=result, + ): + response = api.debug_gtfs_feed_download(id=feed_mdb_40.stable_id) + + assert response.feed_id == feed_mdb_40.stable_id + assert response.success is True + assert response.status_code == 200 + assert response.is_zip is True + assert response.downloaded_bytes == 2038 + assert response.truncated is False + assert response.response_headers == {"Server": "cloudflare"} + + +@pytest.mark.asyncio +async def test_download_debug_failure_returns_diagnostics(): + """A failed download returns 200 with the failure detail rather than raising.""" + api = OperationsApiImpl() + diagnostics = _diagnostics( + status_code=403, + content_type="text/html", + is_zip=False, + downloaded_bytes=None, + external_ip="34.1.2.3", + error_type="ValueError", + error_message="Invalid HTTP response code: [403]", + ) + + with patch( + "feeds_operations.impl.feeds_operations_impl.download_feed", + side_effect=FeedDownloadError("boom", diagnostics), + ): + response = api.debug_gtfs_feed_download(id=feed_mdb_40.stable_id) + + assert response.success is False + assert response.status_code == 403 + assert response.error_type == "ValueError" + assert response.external_ip == "34.1.2.3" + + +@pytest.mark.asyncio +async def test_download_debug_truncated_when_capped(): + """A download stopped at the byte cap is reported as truncated.""" + api = OperationsApiImpl() + result = FeedDownloadResult( + file_hash=None, + is_zip=None, + diagnostics=_diagnostics(downloaded_bytes=1024), + stop_reason="max_bytes", + ) + + with patch( + "feeds_operations.impl.feeds_operations_impl.download_feed", + return_value=result, + ): + response = api.debug_gtfs_feed_download( + id=feed_mdb_40.stable_id, max_bytes=1024 + ) + + assert response.truncated is True + assert response.stop_reason == "max_bytes" + + +@pytest.mark.asyncio +async def test_download_debug_defaults_match_the_nightly_download(): + """No options: whole file, no read timeout, nothing written, bounded by the deadline.""" + api = OperationsApiImpl() + result = FeedDownloadResult( + file_hash="abc123", is_zip=True, diagnostics=_diagnostics(), duration_ms=900 + ) + + with patch( + "feeds_operations.impl.feeds_operations_impl.download_feed", + return_value=result, + ) as download: + response = api.debug_gtfs_feed_download(id=feed_mdb_40.stable_id) + + kwargs = download.call_args.kwargs + assert kwargs["file_path"] is None + assert kwargs["max_bytes"] is None + assert kwargs["timeout_seconds"] is None + assert kwargs["deadline_seconds"] == 480 + assert response.truncated is False + assert response.stop_reason is None + assert response.duration_ms == 900 + + +@pytest.mark.asyncio +async def test_download_debug_not_found(): + """Returns 404 for an unknown feed ID.""" + api = OperationsApiImpl() + with pytest.raises(HTTPException) as exc_info: + api.debug_gtfs_feed_download(id="mdb-9999") + assert exc_info.value.status_code == 404 + + +@with_db_session +def _count_availability_checks(db_session=None) -> int: + return db_session.query(GtfsFeedAvailabilityCheck).count() + + +@pytest.mark.asyncio +async def test_download_debug_does_not_persist(): + """The endpoint stores nothing in gtfs_feed_availability_check.""" + api = OperationsApiImpl() + result = FeedDownloadResult( + file_hash="abc123", is_zip=True, diagnostics=_diagnostics() + ) + + before = _count_availability_checks() + + with patch( + "feeds_operations.impl.feeds_operations_impl.download_feed", + return_value=result, + ): + api.debug_gtfs_feed_download(id=feed_mdb_40.stable_id) + + assert _count_availability_checks() == before + + +@pytest.mark.asyncio +async def test_download_debug_passes_custom_headers(): + """Headers in the request body replace the defaults for the download.""" + api = OperationsApiImpl() + result = FeedDownloadResult( + file_hash="abc123", is_zip=True, diagnostics=_diagnostics() + ) + headers = {"User-Agent": "curl/8.0", "Accept": "*/*"} + + with patch( + "feeds_operations.impl.feeds_operations_impl.download_feed", + return_value=result, + ) as download: + api.debug_gtfs_feed_download( + id=feed_mdb_40.stable_id, + gtfs_feed_download_debug_request=GtfsFeedDownloadDebugRequest( + headers=headers + ), + ) + + assert download.call_args.kwargs["headers"] == headers + + +@pytest.mark.asyncio +async def test_download_debug_without_body_uses_default_headers(): + """No request body leaves header selection to the shared download path.""" + api = OperationsApiImpl() + result = FeedDownloadResult( + file_hash="abc123", is_zip=True, diagnostics=_diagnostics() + ) + + with patch( + "feeds_operations.impl.feeds_operations_impl.download_feed", + return_value=result, + ) as download: + api.debug_gtfs_feed_download(id=feed_mdb_40.stable_id) + + assert download.call_args.kwargs["headers"] is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "headers", + [{"Bad Name": "x"}, {"": "x"}, {"X-Test": "a\r\nX-Injected: b"}], +) +async def test_download_debug_rejects_invalid_headers(headers): + """Malformed header names and values with line breaks are a 400, not a download.""" + api = OperationsApiImpl() + with patch("feeds_operations.impl.feeds_operations_impl.download_feed") as download: + with pytest.raises(HTTPException) as exc_info: + api.debug_gtfs_feed_download( + id=feed_mdb_40.stable_id, + gtfs_feed_download_debug_request=GtfsFeedDownloadDebugRequest( + headers=headers + ), + ) + assert exc_info.value.status_code == 400 + download.assert_not_called() + + +def _nightly(headers=None): + return GtfsFeedDownloadDebugRequest(run_on="nightly_job", headers=headers) + + +@pytest.mark.asyncio +async def test_nightly_job_queues_a_dry_run(): + """run_on=nightly_job publishes a dry-run message and answers 202.""" + api = OperationsApiImpl() + with patch( + "feeds_operations.impl.feeds_operations_impl.DATASET_BATCH_TOPIC", "topic" + ), patch( + "feeds_operations.impl.feeds_operations_impl.trigger_dataset_download" + ) as trigger, patch( + "feeds_operations.impl.feeds_operations_impl.get_request_context", + return_value={}, + ), patch( + "feeds_operations.impl.feeds_operations_impl.download_feed" + ) as download: + response = api.debug_gtfs_feed_download( + id=feed_mdb_40.stable_id, + gtfs_feed_download_debug_request=_nightly({"User-Agent": "curl/8.0"}), + ) + + assert response.status_code == 202 + body = json.loads(response.body) + assert body["feed_id"] == feed_mdb_40.stable_id + assert body["source"] == "dataset_download_manual" + assert body["execution_id"].startswith("download-dry-run") + download.assert_not_called() + trigger.assert_called_once() + assert body["dry_run"] is True + assert trigger.call_args.kwargs == { + "dry_run": True, + "headers": {"User-Agent": "curl/8.0"}, + } + + +@pytest.mark.asyncio +async def test_nightly_job_without_topic_is_503(): + """Without a dataset topic nothing would run, so the caller is told.""" + api = OperationsApiImpl() + with patch( + "feeds_operations.impl.feeds_operations_impl.DATASET_BATCH_TOPIC", None + ), patch( + "feeds_operations.impl.feeds_operations_impl.trigger_dataset_download" + ) as trigger: + with pytest.raises(HTTPException) as exc_info: + api.debug_gtfs_feed_download( + id=feed_mdb_40.stable_id, gtfs_feed_download_debug_request=_nightly() + ) + assert exc_info.value.status_code == 503 + trigger.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("params", [{"max_bytes": 1024}, {"timeout_seconds": 30}]) +async def test_nightly_job_rejects_api_only_options(params): + """The nightly job has no size cap or timeout to set, so they are refused.""" + api = OperationsApiImpl() + with patch( + "feeds_operations.impl.feeds_operations_impl.trigger_dataset_download" + ) as trigger: + with pytest.raises(HTTPException) as exc_info: + api.debug_gtfs_feed_download( + id=feed_mdb_40.stable_id, + gtfs_feed_download_debug_request=_nightly(), + **params, + ) + assert exc_info.value.status_code == 400 + trigger.assert_not_called() + + +@pytest.mark.asyncio +async def test_nightly_job_rejects_invalid_headers(): + api = OperationsApiImpl() + with patch( + "feeds_operations.impl.feeds_operations_impl.trigger_dataset_download" + ) as trigger: + with pytest.raises(HTTPException) as exc_info: + api.debug_gtfs_feed_download( + id=feed_mdb_40.stable_id, + gtfs_feed_download_debug_request=_nightly({"Bad Name": "x"}), + ) + assert exc_info.value.status_code == 400 + trigger.assert_not_called() + + +@pytest.mark.asyncio +async def test_full_run_queues_the_nightly_process(): + """dry_run=false sends the normal nightly message, recording the outcome.""" + api = OperationsApiImpl() + with patch( + "feeds_operations.impl.feeds_operations_impl.DATASET_BATCH_TOPIC", "topic" + ), patch( + "feeds_operations.impl.feeds_operations_impl.trigger_dataset_download" + ) as trigger, patch( + "feeds_operations.impl.feeds_operations_impl.get_request_context", + return_value={}, + ): + response = api.debug_gtfs_feed_download( + id=feed_mdb_40.stable_id, + gtfs_feed_download_debug_request=GtfsFeedDownloadDebugRequest( + run_on="nightly_job", dry_run=False + ), + ) + + assert response.status_code == 202 + body = json.loads(response.body) + assert body["dry_run"] is False + assert body["source"] == "dataset_download_manual" + assert body["execution_id"].startswith("download-manual") + # mdb-40 has no dataset yet, so there is no hash to compare against. + assert trigger.call_args.kwargs == { + "record_source": "dataset_download_manual", + "dataset_stable_id": None, + "dataset_hash": None, + } + + +@pytest.mark.asyncio +async def test_full_run_sends_the_latest_dataset_hash(): + """An unchanged file must be recognised, so the latest hash goes with the message.""" + feed = MagicMock(stable_id="mdb-1") + latest = MagicMock(stable_id="mdb-1-202610200000", hash="abc123") + with patch( + "feeds_operations.impl.feeds_operations_impl.DATASET_BATCH_TOPIC", "topic" + ), patch( + "feeds_operations.impl.feeds_operations_impl.trigger_dataset_download" + ) as trigger, patch( + "feeds_operations.impl.feeds_operations_impl.get_request_context", + return_value={}, + ): + _queue_nightly_download(feed, None, dry_run=False, latest_dataset=latest) + + assert trigger.call_args.kwargs == { + "record_source": "dataset_download_manual", + "dataset_stable_id": "mdb-1-202610200000", + "dataset_hash": "abc123", + } + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "request_body", + [ + {"dry_run": False}, + {"run_on": "api", "dry_run": False}, + {"run_on": "nightly_job", "dry_run": False, "headers": {"X-Test": "1"}}, + ], +) +async def test_full_run_rejects_invalid_combinations(request_body): + """A full run needs the nightly job and the nightly request headers.""" + api = OperationsApiImpl() + with patch( + "feeds_operations.impl.feeds_operations_impl.trigger_dataset_download" + ) as trigger, patch( + "feeds_operations.impl.feeds_operations_impl.download_feed" + ) as download: + with pytest.raises(HTTPException) as exc_info: + api.debug_gtfs_feed_download( + id=feed_mdb_40.stable_id, + gtfs_feed_download_debug_request=GtfsFeedDownloadDebugRequest( + **request_body + ), + ) + assert exc_info.value.status_code == 400 + trigger.assert_not_called() + download.assert_not_called() diff --git a/functions-python/tasks_executor/src/tasks/feed_availability/check_gtfs_feed_availability.py b/functions-python/tasks_executor/src/tasks/feed_availability/check_gtfs_feed_availability.py index 8e54d6577..d7adaf847 100644 --- a/functions-python/tasks_executor/src/tasks/feed_availability/check_gtfs_feed_availability.py +++ b/functions-python/tasks_executor/src/tasks/feed_availability/check_gtfs_feed_availability.py @@ -14,7 +14,6 @@ # limitations under the License. # import logging -import os import time from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone @@ -22,13 +21,14 @@ from sqlalchemy.orm import Session +from shared.common.availability_check_source import AvailabilityCheckSource from shared.database.database import with_db_session from shared.database_gen.sqlacodegen_models import ( Feed, Gtfsfeed, GtfsFeedAvailabilityCheck, ) -from shared.helpers.utils import perform_request +from shared.helpers.utils import get_feed_credentials, perform_request DEFAULT_CONCURRENCY: int = 15 DEFAULT_TIMEOUT_SECONDS: int = 20 @@ -36,18 +36,6 @@ DEFAULT_FALLBACK_TO_GET: bool = True -def get_feed_credentials(stable_id: str) -> Optional[str]: - """Return the API credential for a feed from the FEEDS_CREDENTIALS env var, or None.""" - try: - import json - - feeds_credentials = json.loads(os.getenv("FEEDS_CREDENTIALS", "{}")) - return feeds_credentials.get(stable_id, None) - except Exception as exc: - logging.warning("Could not parse FEEDS_CREDENTIALS: %s", exc) - return None - - def get_parameters(payload: dict): dry_run = payload.get("dry_run", True) skip_db_update = payload.get("skip_db_update", False) @@ -286,6 +274,7 @@ def check_gtfs_feed_availability( error_message=str(exc), error_type=type(exc).__name__, success=False, + source=AvailabilityCheckSource.AVAILABILITY_CHECK.value, ) results.append(check) batch.append(check) diff --git a/functions-python/tasks_executor/src/tasks/seal_of_reliability/context.py b/functions-python/tasks_executor/src/tasks/seal_of_reliability/context.py index baad8cfc8..31bf87f76 100644 --- a/functions-python/tasks_executor/src/tasks/seal_of_reliability/context.py +++ b/functions-python/tasks_executor/src/tasks/seal_of_reliability/context.py @@ -38,6 +38,7 @@ from sqlalchemy import distinct, func, select from sqlalchemy.orm import Session +from shared.common.availability_check_source import feed_availability_checks_filter from shared.common.continuous_coverage import CALENDAR_FILES from shared.common.seal_criteria import AVAILABILITY_LOOKBACK, CriterionNameStr from tasks.seal_of_reliability.history import ( @@ -391,6 +392,9 @@ def _load_availability( ) -> Dict[FeedIdStr, AvailabilityCheck]: """feed_id -> its latest availability check in the 24 hours up to `now`. + Only the sources that count as feed availability are read, so a download failure + recorded in the same table does not stand in for the daily check. + A rolling 24-hour window rather than "the UTC day of `now`", so a check still counts when the availability job (02:00 UTC) and the seal run (04:00 UTC) drift apart, or when one of them runs late. @@ -407,6 +411,7 @@ def _load_availability( GtfsFeedAvailabilityCheck.feed_id.in_(list(feed_ids)), GtfsFeedAvailabilityCheck.checked_at > now - AVAILABILITY_LOOKBACK, GtfsFeedAvailabilityCheck.checked_at <= now, + feed_availability_checks_filter(), ) .distinct(GtfsFeedAvailabilityCheck.feed_id) .order_by( diff --git a/functions-python/tasks_executor/src/tasks/seal_of_reliability/evaluators/available.py b/functions-python/tasks_executor/src/tasks/seal_of_reliability/evaluators/available.py index 4353b8de7..f68f43e9a 100644 --- a/functions-python/tasks_executor/src/tasks/seal_of_reliability/evaluators/available.py +++ b/functions-python/tasks_executor/src/tasks/seal_of_reliability/evaluators/available.py @@ -21,6 +21,7 @@ from sqlalchemy import select from sqlalchemy.orm import Session +from shared.common.availability_check_source import feed_availability_checks_filter from shared.common.seal_criteria import ( AVAILABILITY_LOOKBACK, CriterionStatus, @@ -54,7 +55,11 @@ def load_history( feeds: Sequence, days: Sequence[date], ) -> Optional[AvailabilityHistory]: - """Load every availability check for all the feeds and days, in one query.""" + """Load every availability check for all the feeds and days, in one query. + + Same source filter as the nightly run, so a backfilled day reads what that day's run + would have read. + """ if not feeds or not days: return AvailabilityHistory({}) @@ -78,6 +83,7 @@ def load_history( GtfsFeedAvailabilityCheck.feed_id.in_(feed_ids), GtfsFeedAvailabilityCheck.checked_at > range_start, GtfsFeedAvailabilityCheck.checked_at <= range_end, + feed_availability_checks_filter(), ) ).all() diff --git a/functions-python/tasks_executor/tests/tasks/seal_of_reliability/backfill/test_seal_backfill.py b/functions-python/tasks_executor/tests/tasks/seal_of_reliability/backfill/test_seal_backfill.py index b3cd18c20..693bafcf7 100644 --- a/functions-python/tasks_executor/tests/tasks/seal_of_reliability/backfill/test_seal_backfill.py +++ b/functions-python/tasks_executor/tests/tasks/seal_of_reliability/backfill/test_seal_backfill.py @@ -31,6 +31,7 @@ from sqlalchemy import delete, insert, select +from shared.common.availability_check_source import AvailabilityCheckSource from shared.database.database import with_db_session from shared.database_gen.sqlacodegen_models import ( Feedinfo, @@ -1513,9 +1514,19 @@ def test_an_uncollapsed_trace_reports_every_marched_day(self): # day's start, so a check taken during day D falls in the window of day D+1, never day D. OK_CHECK = datetime(2026, 5, 11, 14, 0, tzinfo=timezone.utc) BAD_CHECK = datetime(2026, 5, 13, 14, 0, tzinfo=timezone.utc) - - -def _seed_check(db_session, feed_id, checked_at, success): +# Download diagnostics recorded in the same table. Each is the newest row of its day, so a +# march that read them would turn 2026-05-12 into a FAIL and 2026-05-16 into a PASS. +FAILED_DOWNLOAD = datetime(2026, 5, 11, 16, 0, tzinfo=timezone.utc) +MANUAL_DOWNLOAD = datetime(2026, 5, 15, 14, 0, tzinfo=timezone.utc) + + +def _seed_check( + db_session, + feed_id, + checked_at, + success, + source=AvailabilityCheckSource.AVAILABILITY_CHECK, +): db_session.execute( insert(GtfsFeedAvailabilityCheck.__table__).values( feed_id=feed_id, @@ -1523,6 +1534,7 @@ def _seed_check(db_session, feed_id, checked_at, success): request_url=f"https://example.com/{feed_id}.zip", request_type="http_head", success=success, + source=source.value, ) ) @@ -1539,6 +1551,9 @@ class TestAvailabilityIsMarchedFromHistory(unittest.TestCase): one becomes a PASS, the failed one a FAIL, and the days whose 24-hour window holds neither stay UNKNOWN. A run that ignores the table fails on the first two; one that ignores the window and reuses a single check for the whole march fails on the third. + + Two download diagnostics are seeded too and must change nothing: only the daily check + answers Available, in a march as in the nightly run. """ @with_db_session(db_url=default_db_url) @@ -1547,6 +1562,20 @@ def setUp(self, db_session): _seed_feed(db_session, AVAIL, OLD_CREATED) _seed_check(db_session, AVAIL, OK_CHECK, True) _seed_check(db_session, AVAIL, BAD_CHECK, False) + _seed_check( + db_session, + AVAIL, + FAILED_DOWNLOAD, + False, + AvailabilityCheckSource.DATASET_DOWNLOAD, + ) + _seed_check( + db_session, + AVAIL, + MANUAL_DOWNLOAD, + True, + AvailabilityCheckSource.DATASET_DOWNLOAD_MANUAL, + ) db_session.commit() @with_db_session(db_url=default_db_url) diff --git a/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_updater_db.py b/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_updater_db.py index 2c23c3531..92ba7eb80 100644 --- a/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_updater_db.py +++ b/functions-python/tasks_executor/tests/tasks/seal_of_reliability/test_seal_updater_db.py @@ -20,6 +20,7 @@ from datetime import datetime, timedelta, timezone from unittest.mock import MagicMock, patch +from shared.common.availability_check_source import AvailabilityCheckSource from shared.common.seal_criteria import ( AVAILABILITY_LOOKBACK, PROBATION_PERIOD, @@ -267,7 +268,13 @@ def _seed_previous_dataset( db_session.commit() -def _seed_availability_check(db_session, feed_id: str, success, checked_at=None): +def _seed_availability_check( + db_session, + feed_id: str, + success, + checked_at=None, + source=AvailabilityCheckSource.AVAILABILITY_CHECK, +): """One `gtfs_feed_availability_check` row for the feed.""" db_session.add( GtfsFeedAvailabilityCheck( @@ -277,6 +284,7 @@ def _seed_availability_check(db_session, feed_id: str, success, checked_at=None) request_type="http_head", status_code=200 if success else 503, success=success, + source=source.value, ) ) db_session.commit() @@ -440,8 +448,14 @@ def seed_previous_dataset( @staticmethod @with_db_session(db_url=default_db_url) - def seed_availability_check(feed_id, success, checked_at=None, db_session=None): - _seed_availability_check(db_session, feed_id, success, checked_at) + def seed_availability_check( + feed_id, + success, + checked_at=None, + source=AvailabilityCheckSource.AVAILABILITY_CHECK, + db_session=None, + ): + _seed_availability_check(db_session, feed_id, success, checked_at, source) @staticmethod @with_db_session(db_url=default_db_url) @@ -1675,6 +1689,49 @@ def test_a_recovery_later_in_the_window_wins(self): row = self.criterion_rows(TRACKED)[SealCriterionName.AVAILABLE.value] self.assertEqual(row.observed_status, CriterionStatus.PASS.value) + def test_download_diagnostics_do_not_stand_in_for_the_daily_check(self): + """Failed and hand-run downloads share the table but never decide Available. + + Each is the latest row in the window, so without the source filter it would be the + verdict and fail a feed whose daily check passed. + """ + self.satisfy_everything() + for hours, source in enumerate( + [ + AvailabilityCheckSource.DATASET_DOWNLOAD, + AvailabilityCheckSource.DATASET_DOWNLOAD_MANUAL, + ], + start=1, + ): + self.seed_availability_check( + TRACKED, + success=False, + checked_at=NOW + timedelta(hours=hours), + source=source, + ) + + update_seals( + dry_run=False, stable_feed_ids=[TRACKED], now=NOW + timedelta(hours=4) + ) + + row = self.criterion_rows(TRACKED)[SealCriterionName.AVAILABLE.value] + self.assertEqual(row.observed_status, CriterionStatus.PASS.value) + self.assertTrue(self.seal_row(TRACKED).has_seal) + + def test_download_diagnostics_alone_leave_available_unknown(self): + """With only download rows in the window, the daily check did not run: UNKNOWN.""" + self.seed_dataset(TRACKED, self.FAR_FUTURE) + self.seed_availability_check( + TRACKED, + success=True, + source=AvailabilityCheckSource.DATASET_DOWNLOAD_MANUAL, + ) + + update_seals(dry_run=False, stable_feed_ids=[TRACKED], now=NOW) + + row = self.criterion_rows(TRACKED)[SealCriterionName.AVAILABLE.value] + self.assertEqual(row.observed_status, CriterionStatus.UNKNOWN.value) + def test_a_check_still_answers_a_second_run_inside_the_window(self): """The window is a rolling 24h, not "checks this run has not seen yet".""" self.satisfy_everything() diff --git a/infra/batch/main.tf b/infra/batch/main.tf index 51cfe1feb..81cb6cfd2 100644 --- a/infra/batch/main.tf +++ b/infra/batch/main.tf @@ -150,20 +150,23 @@ resource "google_secret_manager_secret_iam_member" "secret_iam_member" { member = "serviceAccount:${google_service_account.functions_service_account.email}" } -# Grant permissions to the service account to publish to the pubsub topic -resource "google_pubsub_topic_iam_binding" "functions_publisher" { +# Grant permissions to the service account to publish to the pubsub topic. +# A member, not a binding: infra/functions-python grants the same role on this topic to +# its own service account (operations-api publishes here), and an authoritative binding +# would remove that grant on every batch deploy. +resource "google_pubsub_topic_iam_member" "functions_publisher" { project = var.project_id role = "roles/pubsub.publisher" topic = google_pubsub_topic.pubsub_topic.name - members = ["serviceAccount:${google_service_account.functions_service_account.email}"] + member = "serviceAccount:${google_service_account.functions_service_account.email}" } # Grant permissions to the service account to subscribe to the pubsub topic -resource "google_pubsub_topic_iam_binding" "functions_subscriber" { +resource "google_pubsub_topic_iam_member" "functions_subscriber" { project = var.project_id role = "roles/pubsub.subscriber" topic = google_pubsub_topic.pubsub_topic.name - members = ["serviceAccount:${google_service_account.functions_service_account.email}"] + member = "serviceAccount:${google_service_account.functions_service_account.email}" } # Batch datasets function diff --git a/liquibase/changelog.xml b/liquibase/changelog.xml index aea96aac0..cf37786bf 100644 --- a/liquibase/changelog.xml +++ b/liquibase/changelog.xml @@ -154,6 +154,10 @@ + + + + diff --git a/liquibase/changes/feat_download_debug.sql b/liquibase/changes/feat_download_debug.sql new file mode 100644 index 000000000..55a9893f5 --- /dev/null +++ b/liquibase/changes/feat_download_debug.sql @@ -0,0 +1,20 @@ +-- Add request/response debugging columns to gtfs_feed_availability_check. +-- source distinguishes rows written by the daily availability task from rows written +-- when a dataset download fails in batch_process_dataset. +-- request_headers, response_headers and resolved_url are sanitized before being stored: +-- credential-bearing header values and query parameters are replaced with a redaction marker. +-- external_ip is the egress IP observed for the request, captured best-effort on failures only. +ALTER TABLE gtfs_feed_availability_check + ADD COLUMN IF NOT EXISTS source TEXT, + ADD COLUMN IF NOT EXISTS request_headers JSONB, + ADD COLUMN IF NOT EXISTS response_headers JSONB, + ADD COLUMN IF NOT EXISTS redirect_urls JSONB, + ADD COLUMN IF NOT EXISTS external_ip TEXT, + ADD COLUMN IF NOT EXISTS resolved_url TEXT; + +UPDATE gtfs_feed_availability_check +SET source = 'availability_check' +WHERE source IS NULL; + +CREATE INDEX IF NOT EXISTS idx_gtfs_feed_availability_check_source_checked_at + ON gtfs_feed_availability_check (source, checked_at); diff --git a/liquibase/changes/feat_download_debug_source_default.sql b/liquibase/changes/feat_download_debug_source_default.sql new file mode 100644 index 000000000..5e4cc7719 --- /dev/null +++ b/liquibase/changes/feat_download_debug_source_default.sql @@ -0,0 +1,18 @@ +-- Make gtfs_feed_availability_check.source always set, and fold dry runs into manual downloads. +-- The public availability endpoint and the Seal of Reliability keep only rows whose source +-- is 'availability_check' (see api/src/shared/common/availability_check_source.py). A row +-- written without a source would silently drop out of both, so a missing source means the +-- daily availability check, which is what every row before the column existed was. +UPDATE gtfs_feed_availability_check +SET source = 'availability_check' +WHERE source IS NULL; + +-- Dry runs asked for through the operations API are recorded as manual downloads, like full +-- runs; the separate dry-run source is retired. +UPDATE gtfs_feed_availability_check +SET source = 'dataset_download_manual' +WHERE source = 'dataset_download_dry_run'; + +ALTER TABLE gtfs_feed_availability_check + ALTER COLUMN source SET DEFAULT 'availability_check', + ALTER COLUMN source SET NOT NULL; diff --git a/scripts/download-verifier.sh b/scripts/download-verifier.sh new file mode 100755 index 000000000..f4ee8853f --- /dev/null +++ b/scripts/download-verifier.sh @@ -0,0 +1,124 @@ +#!/bin/bash +# +# +# MobilityData 2026 +# +# 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 +# +# http://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. +# +# + +# Downloads a GTFS feed using the same shared code as the nightly dataset download and +# prints the request and response detail of the attempt, with credentials redacted. +# No database is used. To reproduce a production feed_download/http_headers override, pass +# its headers with --header; they replace the default headers. + +# Usage: +# download-verifier.sh --url [--header ""]... [--with_upload] [--no_install_venv] +# Example: +# download-verifier.sh --url https://example.com/gtfs.zip --header "User-Agent: curl/8.0" + +# relative path +SCRIPT_PATH="$(dirname -- "${BASH_SOURCE[0]}")" +FUNCTION_PATH="$SCRIPT_PATH/../functions-python/batch_process_dataset" + +# function printing usage +display_usage() { + printf "\nThis script downloads a GTFS feed and prints the request and response detail" + printf "\nScript Usage:\n" + echo "Usage: $0 [options]" + echo "Options:" + echo " -h|--help Display help content." + echo " --url Producer URL of the feed to download. Required." + echo " --header Request header, repeatable. Replaces the default headers." + echo " --with_upload Also verify the upload path using the GCP storage emulator." + echo " --no_install_venv Do not reinstall the python virtual environment." + exit 1 +} + +url='' +headers=() +with_upload=false +no_install_venv=false + +while [[ $# -gt 0 ]]; do + key="$1" + + case $key in + -h | --help) + display_usage + exit 0 + ;; + --url) + url="$2" + shift # past argument + shift # past value + ;; + --header) + headers+=("$2") + shift # past argument + shift # past value + ;; + --with_upload) + with_upload=true + shift # past argument + ;; + --no_install_venv) + no_install_venv=true + shift # past argument + ;; + *) # unknown option + shift # past argument + ;; + esac +done + +if [ -z "$url" ]; then + printf "\nERROR: --url is required\n" + display_usage + exit 1 +fi + +if [ ! -d "$FUNCTION_PATH/src" ]; then + printf "\nERROR: function's folder not found at location: %s\n" "$FUNCTION_PATH/src" + exit 1 +fi + +# Link the shared code into src/shared so the function's imports resolve. +printf "\nINFO: linking shared code\n" +"$SCRIPT_PATH"/function-python-setup.sh --function_name batch_process_dataset + +if [ "$no_install_venv" != "true" ]; then + printf "\nINFO: installing python virtual environment\n" + pushd "$FUNCTION_PATH" >/dev/null + rm -rf venv + pip3 install --disable-pip-version-check virtualenv >/dev/null + python3 -m virtualenv venv >/dev/null + venv/bin/python -m pip install --disable-pip-version-check \ + -r requirements.txt -r requirements_dev.txt >/dev/null + popd >/dev/null +else + printf "\nINFO: skipping python virtual environment installation\n" +fi + +export PYTHONPATH="$FUNCTION_PATH/src" + +args=(--url "$url") +for header in "${headers[@]}"; do + args+=(--header "$header") +done +if [ "$with_upload" != "true" ]; then + args+=(--skip-upload) +fi + +printf "\nINFO: downloading %s\n" "$url" +"$FUNCTION_PATH"/venv/bin/python "$FUNCTION_PATH/src/scripts/download_verifier.py" "${args[@]}"