//! **GTFS** (General Transit Feed Specification) static feeds — feature `gtfs`.
//!
//! A GTFS feed is a **ZIP archive of CSV `.txt` files** (`stops.txt`, `routes.txt`,
//! `trips.txt`, `agency.txt`, `calendar.txt`, …). This collector reads it with a
//! **pure-Rust ZIP reader** ([`unzip`]) — a central-directory walk that handles the
//! STORED (0) and DEFLATE (8) methods via flate2's miniz_oxide backend, so the leaf
//! needs no `zip` crate — then parses the chosen member with the same light `csv`
//! reader the CSV collector uses, mapping the GTFS spec columns into a **typed**
//! Arrow batch (a fixed per-file schema, not the CSV two-pass sniff: GTFS columns
//! are named + typed by the spec).
//!
//! Three presets ship — [`GtfsCollector::stops`], [`GtfsCollector::routes`],
//! [`GtfsCollector::trips`] — each picking one file + its column set. The parse
//! core [`parse_gtfs_zip`] is pure (bytes in, Arrow out — no I/O), so the fixture
//! test asserts the round-tripped Arrow as DATA.
use std::io::{Read, Write};
use anyhow::{Result, anyhow};
use arrow_array::{ArrayRef, RecordBatch};
use arrow_schema::{DataType, Schema};
use super::arrow_util::{batch, f64_col, i64_col, schema as build_schema, str_col};
use super::{
AuthKind, BatchSink, Bounds, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
// ── pure-Rust ZIP reader (central-directory walk) ───────────────────────────────
const EOCD_SIG: u32 = 0x0605_4b50; // PK\x05\x06 — End Of Central Directory
const CDH_SIG: u32 = 0x0201_4b50; // PK\x01\x02 — Central Directory Header
const LFH_SIG: u32 = 0x0403_4b50; // PK\x03\x04 — Local File Header
fn u16le(b: &[u8], o: usize) -> Option<u16> {
b.get(o..o + 2).map(|s| u16::from_le_bytes([s[0], s[1]]))
}
fn u32le(b: &[u8], o: usize) -> Option<u32> {
b.get(o..o + 4)
.map(|s| u32::from_le_bytes([s[0], s[1], s[2], s[3]]))
}
/// One extracted ZIP member.
#[derive(Debug, Clone)]
pub struct ZipEntry {
pub name: String,
pub data: Vec<u8>,
}
/// Raw-DEFLATE inflate (ZIP stores raw deflate, no zlib wrapper).
fn inflate(comp: &[u8]) -> Result<Vec<u8>> {
let mut d = flate2::read::DeflateDecoder::new(comp);
let mut out = Vec::new();
d.read_to_end(&mut out)?;
Ok(out)
}
/// Extract every member of a ZIP archive via the central directory. Supports the
/// STORED (0) and DEFLATE (8) methods; a ZIP64 / unsupported-method entry is a
/// hard error (named), never a silent skip.
pub fn unzip(zip: &[u8]) -> Result<Vec<ZipEntry>> {
// Find the EOCD by scanning backwards for its signature (comment is usually
// empty; cap the scan at the max comment length + record size).
let mut eocd = None;
let start = zip.len().saturating_sub(22 + 0xFFFF);
for i in (start..=zip.len().saturating_sub(22)).rev() {
if u32le(zip, i) == Some(EOCD_SIG) {
eocd = Some(i);
break;
}
}
let eocd = eocd.ok_or_else(|| anyhow!("not a ZIP archive (no EOCD record)"))?;
let total = u16le(zip, eocd + 10).ok_or_else(|| anyhow!("truncated EOCD"))? as usize;
let cd_offset = u32le(zip, eocd + 16).ok_or_else(|| anyhow!("truncated EOCD"))? as usize;
let mut out = Vec::with_capacity(total);
let mut p = cd_offset;
for _ in 0..total {
if u32le(zip, p) != Some(CDH_SIG) {
return Err(anyhow!("bad central-directory header at offset {p}"));
}
let method = u16le(zip, p + 10).ok_or_else(|| anyhow!("truncated CDH"))?;
let comp_size = u32le(zip, p + 20).ok_or_else(|| anyhow!("truncated CDH"))? as usize;
let name_len = u16le(zip, p + 28).ok_or_else(|| anyhow!("truncated CDH"))? as usize;
let extra_len = u16le(zip, p + 30).ok_or_else(|| anyhow!("truncated CDH"))? as usize;
let comment_len = u16le(zip, p + 32).ok_or_else(|| anyhow!("truncated CDH"))? as usize;
let local_off = u32le(zip, p + 42).ok_or_else(|| anyhow!("truncated CDH"))? as usize;
let name = String::from_utf8_lossy(
zip.get(p + 46..p + 46 + name_len)
.ok_or_else(|| anyhow!("truncated CDH name"))?,
)
.into_owned();
if comp_size == 0xFFFF_FFFF {
return Err(anyhow!("ZIP64 entry `{name}` not supported"));
}
// Seek to the local file header — its name/extra lengths can differ from
// the central directory's, so read them from the local header itself.
if u32le(zip, local_off) != Some(LFH_SIG) {
return Err(anyhow!("bad local header for `{name}` at {local_off}"));
}
let l_name_len =
u16le(zip, local_off + 26).ok_or_else(|| anyhow!("truncated LFH"))? as usize;
let l_extra_len =
u16le(zip, local_off + 28).ok_or_else(|| anyhow!("truncated LFH"))? as usize;
let data_start = local_off + 30 + l_name_len + l_extra_len;
let raw = zip
.get(data_start..data_start + comp_size)
.ok_or_else(|| anyhow!("truncated data for `{name}`"))?;
let data = match method {
0 => raw.to_vec(),
8 => inflate(raw)?,
m => return Err(anyhow!("unsupported ZIP method {m} for `{name}`")),
};
out.push(ZipEntry { name, data });
p += 46 + name_len + extra_len + comment_len;
}
Ok(out)
}
/// Find one member's bytes by (base)name — matches either the exact stored name or
/// its trailing path component (GTFS files may sit under a folder in the zip).
pub fn zip_entry<'a>(entries: &'a [ZipEntry], name: &str) -> Option<&'a [u8]> {
entries
.iter()
.find(|e| e.name == name || e.name.rsplit('/').next() == Some(name))
.map(|e| e.data.as_slice())
}
/// **Build a ZIP archive** from `(name, method, raw_bytes)` members (method `0`
/// STORED, `8` DEFLATE). This is the fixture/round-trip helper the tests + the
/// offline GTFS sample use to synthesise a feed with no committed binary blob; it
/// writes crc=0 (this crate's reader does not verify CRC).
pub fn build_zip(files: &[(&str, u8, &[u8])]) -> Vec<u8> {
fn deflate(raw: &[u8]) -> Vec<u8> {
let mut e = flate2::write::DeflateEncoder::new(Vec::new(), flate2::Compression::default());
e.write_all(raw).expect("deflate write");
e.finish().expect("deflate finish")
}
let mut out = Vec::new();
let mut central = Vec::new();
let mut offsets = Vec::new();
for (name, method, raw) in files {
let stored = if *method == 8 {
deflate(raw)
} else {
raw.to_vec()
};
let local_off = out.len() as u32;
offsets.push((
local_off,
*method,
stored.clone(),
name.to_string(),
raw.len() as u32,
));
// Local file header.
out.extend_from_slice(&LFH_SIG.to_le_bytes());
out.extend_from_slice(&20u16.to_le_bytes()); // version needed
out.extend_from_slice(&0u16.to_le_bytes()); // flags
out.extend_from_slice(&(*method as u16).to_le_bytes());
out.extend_from_slice(&0u16.to_le_bytes()); // mod time
out.extend_from_slice(&0u16.to_le_bytes()); // mod date
out.extend_from_slice(&0u32.to_le_bytes()); // crc32 (unverified)
out.extend_from_slice(&(stored.len() as u32).to_le_bytes());
out.extend_from_slice(&(raw.len() as u32).to_le_bytes());
out.extend_from_slice(&(name.len() as u16).to_le_bytes());
out.extend_from_slice(&0u16.to_le_bytes()); // extra len
out.extend_from_slice(name.as_bytes());
out.extend_from_slice(&stored);
}
let cd_offset = out.len() as u32;
for (local_off, method, stored, name, raw_len) in &offsets {
central.extend_from_slice(&CDH_SIG.to_le_bytes());
central.extend_from_slice(&20u16.to_le_bytes()); // version made
central.extend_from_slice(&20u16.to_le_bytes()); // version needed
central.extend_from_slice(&0u16.to_le_bytes()); // flags
central.extend_from_slice(&(*method as u16).to_le_bytes());
central.extend_from_slice(&0u16.to_le_bytes()); // mod time
central.extend_from_slice(&0u16.to_le_bytes()); // mod date
central.extend_from_slice(&0u32.to_le_bytes()); // crc32
central.extend_from_slice(&(stored.len() as u32).to_le_bytes());
central.extend_from_slice(&raw_len.to_le_bytes());
central.extend_from_slice(&(name.len() as u16).to_le_bytes());
central.extend_from_slice(&0u16.to_le_bytes()); // extra
central.extend_from_slice(&0u16.to_le_bytes()); // comment
central.extend_from_slice(&0u16.to_le_bytes()); // disk start
central.extend_from_slice(&0u16.to_le_bytes()); // internal attr
central.extend_from_slice(&0u32.to_le_bytes()); // external attr
central.extend_from_slice(&local_off.to_le_bytes());
central.extend_from_slice(name.as_bytes());
}
let cd_size = central.len() as u32;
out.extend_from_slice(¢ral);
// EOCD.
out.extend_from_slice(&EOCD_SIG.to_le_bytes());
out.extend_from_slice(&0u16.to_le_bytes()); // disk
out.extend_from_slice(&0u16.to_le_bytes()); // cd disk
out.extend_from_slice(&(files.len() as u16).to_le_bytes()); // entries this disk
out.extend_from_slice(&(files.len() as u16).to_le_bytes()); // entries total
out.extend_from_slice(&cd_size.to_le_bytes());
out.extend_from_slice(&cd_offset.to_le_bytes());
out.extend_from_slice(&0u16.to_le_bytes()); // comment len
out
}
// ── typed GTFS CSV → Arrow ──────────────────────────────────────────────────────
/// The Arrow type a GTFS column maps to.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum GtfsColType {
Str,
F64,
I64,
}
impl GtfsColType {
fn data_type(self) -> DataType {
match self {
GtfsColType::Str => DataType::Utf8,
GtfsColType::F64 => DataType::Float64,
GtfsColType::I64 => DataType::Int64,
}
}
}
/// One GTFS column: the spec header name + its Arrow type.
#[derive(Debug, Clone, Copy)]
pub struct GtfsCol {
pub name: &'static str,
pub ty: GtfsColType,
}
const fn col(name: &'static str, ty: GtfsColType) -> GtfsCol {
GtfsCol { name, ty }
}
/// Parse a GTFS `.txt` CSV (header row + records) into a typed Arrow batch,
/// selecting the `cols` by header name. A column absent from the header yields an
/// all-null column (GTFS marks many fields optional); an empty / unparseable cell
/// is a null.
pub fn parse_gtfs_csv(csv_bytes: &[u8], cols: &[GtfsCol]) -> Result<RecordBatch> {
// GTFS is UTF-8 (spec); strip a BOM if a producer added one.
let bytes = csv_bytes
.strip_prefix(&[0xEF, 0xBB, 0xBF])
.unwrap_or(csv_bytes);
let mut rdr = ::csv::ReaderBuilder::new()
.has_headers(true)
.flexible(true)
.from_reader(bytes);
let header = rdr.headers()?.clone();
let index_of = |want: &str| header.iter().position(|h| h.trim() == want);
let idxs: Vec<Option<usize>> = cols.iter().map(|c| index_of(c.name)).collect();
// Collect raw cells per selected column.
let mut raw: Vec<Vec<Option<String>>> = vec![Vec::new(); cols.len()];
for rec in rdr.records() {
let rec = rec?;
for (ci, idx) in idxs.iter().enumerate() {
let cell = idx
.and_then(|i| rec.get(i))
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_string);
raw[ci].push(cell);
}
}
let columns: Vec<ArrayRef> = cols
.iter()
.zip(&raw)
.map(|(c, cells)| match c.ty {
GtfsColType::Str => str_col(cells),
GtfsColType::F64 => f64_col(
&cells
.iter()
.map(|v| v.as_ref().and_then(|s| s.parse().ok()))
.collect::<Vec<_>>(),
),
GtfsColType::I64 => i64_col(
&cells
.iter()
.map(|v| v.as_ref().and_then(|s| s.parse().ok()))
.collect::<Vec<_>>(),
),
})
.collect();
let fdefs: Vec<(&str, DataType, bool)> = cols
.iter()
.map(|c| (c.name, c.ty.data_type(), true))
.collect();
batch(&build_schema(&fdefs), columns)
}
/// Unzip a GTFS feed and parse one member (`file`, e.g. `"stops.txt"`) into a typed
/// Arrow batch via `cols`. The pure parse core (bytes in, Arrow out — no I/O).
pub fn parse_gtfs_zip(zip: &[u8], file: &str, cols: &[GtfsCol]) -> Result<RecordBatch> {
let entries = unzip(zip)?;
let csv =
zip_entry(&entries, file).ok_or_else(|| anyhow!("GTFS feed has no `{file}` member"))?;
parse_gtfs_csv(csv, cols)
}
// ── collector ─────────────────────────────────────────────────────────────────
/// The GTFS static-feed collector. `cfg.base_url` is the feed `.zip` (a local path
/// or `http(s)://` URL); the preset selects which member + columns to land.
pub struct GtfsCollector {
name: String,
file: String,
cols: Vec<GtfsCol>,
table: String,
}
impl Default for GtfsCollector {
fn default() -> Self {
Self::stops()
}
}
impl GtfsCollector {
/// `stops.txt` → a stop-point table (id, name, WGS84 lat/lon, zone, type).
pub fn stops() -> Self {
Self {
name: "gtfs-stops".into(),
file: "stops.txt".into(),
table: "gtfs_stops".into(),
cols: vec![
col("stop_id", GtfsColType::Str),
col("stop_name", GtfsColType::Str),
col("stop_lat", GtfsColType::F64),
col("stop_lon", GtfsColType::F64),
col("zone_id", GtfsColType::Str),
col("location_type", GtfsColType::I64),
col("parent_station", GtfsColType::Str),
],
}
}
/// `routes.txt` → a route table (id, agency, short/long name, type).
pub fn routes() -> Self {
Self {
name: "gtfs-routes".into(),
file: "routes.txt".into(),
table: "gtfs_routes".into(),
cols: vec![
col("route_id", GtfsColType::Str),
col("agency_id", GtfsColType::Str),
col("route_short_name", GtfsColType::Str),
col("route_long_name", GtfsColType::Str),
col("route_type", GtfsColType::I64),
col("route_color", GtfsColType::Str),
],
}
}
/// `trips.txt` → a trip table (route, service, trip, headsign, direction).
pub fn trips() -> Self {
Self {
name: "gtfs-trips".into(),
file: "trips.txt".into(),
table: "gtfs_trips".into(),
cols: vec![
col("route_id", GtfsColType::Str),
col("service_id", GtfsColType::Str),
col("trip_id", GtfsColType::Str),
col("trip_headsign", GtfsColType::Str),
col("direction_id", GtfsColType::I64),
col("shape_id", GtfsColType::Str),
],
}
}
/// The GTFS member file this preset reads.
pub fn file(&self) -> &str {
&self.file
}
/// The spec columns this preset lands.
pub fn cols(&self) -> &[GtfsCol] {
&self.cols
}
fn gtfs_schema(&self) -> Schema {
let fdefs: Vec<(&str, DataType, bool)> = self
.cols
.iter()
.map(|c| (c.name, c.ty.data_type(), true))
.collect();
build_schema(&fdefs)
}
}
#[async_trait::async_trait]
impl Collector for GtfsCollector {
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: self.name.clone(),
source_kind: SourceKind::Gtfs,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["gtfs-zip".into(), "text/csv".into()],
license: "per-source (typically CC-BY / open transit)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["stop_id".into(), "route_id".into(), "trip_id".into()],
description: "GTFS static transit feed (zip of CSV) → typed stops/routes/trips \
Arrow tables via a pure-Rust ZIP reader + csv parse."
.into(),
}
}
/// **One card for all three presets**, because all three read the same ZIP from the
/// same unknown publisher — only the member file differs, and that difference is a
/// quirk, not a different authority.
///
/// Like [`crate::gtfs_rt`] and [`crate::wfs`] this is a GENERIC reader: the
/// authority is whoever publishes the feed an operator configures. The useful thing
/// korp CAN name is the specification — and, better, the two files inside every
/// GTFS ZIP that carry the publisher's own name, licence page and contact address.
/// That is why `agency.txt` / `feed_info.txt` lead the quirks: they are the answer
/// to "who runs this" for the feed actually configured.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc, RateLimit, Registration};
Some(
HumanDoc::new(
"whoever publishes the GTFS feed you configure — this is a generic reader. \
The ZIP names them itself: `feed_info.txt` (`feed_publisher_name`, \
`feed_publisher_url`) and `agency.txt` (`agency_name`, `agency_url`).",
"the transit agency whose service the feed describes (`agency.txt`)",
"https://github.com/google/transit/blob/master/LICENSE",
"https://gtfs.org/documentation/schedule/reference/",
)
.with_contact(Contact::Unknown)
.with_registration(Registration::NotNeeded)
.with_rate_limit(RateLimit::Documented(
"not set by the GTFS specification — a static feed is one ZIP download and \
the publishing agency sets any limit. `feed_info.txt` carries \
`feed_version` and `feed_start_date`/`feed_end_date`, which is what tells \
you a re-fetch is actually due."
.into(),
))
.with_quirk(
"`terms_url` is the SPECIFICATION's licence, not the feed's. For the data \
licence and a human to ask, read the configured ZIP's own \
`feed_info.txt` — `feed_publisher_url`, `feed_contact_email`, \
`feed_contact_url` — which is the only contact korp can honestly promise \
exists for a feed it has not seen.",
)
.with_quirk(
"The declared source `https://transitfeeds.example/feed/latest.zip` is an \
RFC 2606 documentation placeholder, not a feed: until a real URL (or a \
local path) is configured this collector reports \
`Missing::PlaceholderSource` rather than running.",
)
.with_quirk(
"The pure-Rust ZIP reader supports the STORED (0) and DEFLATE (8) methods \
only, and REFUSES a ZIP64 entry outright (`ZIP64 entry … not supported`). \
Large national feeds are the ones that trip this; re-zip with a classic \
archive if you hit it.",
)
.with_quirk(
"Member lookup tolerates a directory prefix inside the archive — \
`gtfs/stops.txt` resolves for `stops.txt` — but a feed genuinely missing \
the member fails with `GTFS feed has no \\`…\\` member`. A trimmed \
extract that dropped the file is the usual cause.",
)
.with_quirk(
"The columns are FIXED by the preset, not sniffed, and the match is by \
exact header name. A column the preset declares but the feed does not \
carry lands as an all-NULL column rather than an error (`route_color` and \
`shape_id` are optional in GTFS and routinely absent), and a publisher's \
extension column is silently dropped. Check for an all-NULL column before \
concluding the data is missing upstream.",
),
)
}
fn source_target(&self) -> SourceTarget {
// stops carry WGS84 geometry → an osm3d point scene; routes/trips are plain
// warehouse tables that join back via route_id / trip_id.
if self.file == "stops.txt" {
SourceTarget::map_scene(
SourceEndpoint::Gtfs,
"https://transitfeeds.example/feed/latest.zip",
self.table.clone(),
"osm3d",
)
} else {
SourceTarget::warehouse(
SourceEndpoint::Gtfs,
"https://transitfeeds.example/feed/latest.zip",
self.table.clone(),
)
}
}
/// **One row per stop / route / trip** — each variant's OWN id.
///
/// The derived key was the UNION of all three files' ids
/// (`stop_id, route_id, trip_id`), so every variant named two columns it does not
/// emit: `gtfs-stops` has no `route_id`, `gtfs-routes` has no `stop_id`. A GTFS id
/// is unique within its file by the specification, which is what makes the single
/// column an identity rather than a guess.
fn business_key(&self) -> crate::history::BusinessKey {
let (col, what) = match self.file.as_str() {
"routes.txt" => ("route_id", "route"),
"trips.txt" => ("trip_id", "trip"),
_ => ("stop_id", "stop"),
};
crate::history::BusinessKey::of(
&[col],
format!("one row per {what} — `{col}` is unique within its GTFS file by the specification"),
)
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
Ok(self.gtfs_schema())
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
let zip = crate::http::read_source(cfg, &cfg.base_url).await?;
let rb = parse_gtfs_zip(&zip, &self.file, &self.cols)?;
let n = rb.num_rows() as u64;
if n == 0 {
return Ok(CollectStats {
reason: Some(format!("GTFS `{}` had no data rows", self.file)),
..Default::default()
});
}
sink.push(rb)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use arrow_array::Array;
const STOPS: &str = "stop_id,stop_name,stop_lat,stop_lon,zone_id,location_type\n\
9022001,T-Centralen,59.33107,18.05977,A,0\n\
9022002,Slussen,59.31970,18.07208,A,0\n";
const ROUTES: &str = "route_id,agency_id,route_short_name,route_long_name,route_type\n\
R1,SL,17,Åkeshov–Skarpnäck,1\n";
/// A DEFLATE-compressed `stops.txt` inside a hand-built GTFS zip round-trips
/// through the pure-Rust ZIP reader + typed CSV parse → a stop-point table
/// (WGS84 lat/lon → f64). Offline DATA contract + testmatrix cell.
#[test]
fn gtfs_zip_stops_deflate_round_trip_into_a_stop_table() {
// stops DEFLATEd (method 8) — exercises the real inflate path.
let zip = build_zip(&[("stops.txt", 8, STOPS.as_bytes())]);
let c = GtfsCollector::stops();
let rb = parse_gtfs_zip(&zip, c.file(), c.cols()).expect("parse gtfs zip");
assert_eq!(rb.num_rows(), 2, "two stops parsed");
let rb_schema = rb.schema();
let names: Vec<&str> = rb_schema
.fields()
.iter()
.map(|f| f.name().as_str())
.collect();
for want in [
"stop_id",
"stop_name",
"stop_lat",
"stop_lon",
"location_type",
] {
assert!(
names.contains(&want),
"stop table carries `{want}`: {names:?}"
);
}
let id = rb
.column_by_name("stop_id")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::StringArray>())
.expect("stop_id is utf8");
assert_eq!(id.value(0), "9022001");
let lat = rb
.column_by_name("stop_lat")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Float64Array>())
.expect("stop_lat is f64");
assert!(
(lat.value(0) - 59.33107).abs() < 1e-9,
"T-Centralen latitude → f64"
);
let loc = rb
.column_by_name("location_type")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Int64Array>())
.expect("location_type is i64");
assert_eq!(loc.value(0), 0, "location_type → i64");
// A column absent from the header (parent_station) is an all-null column.
let parent = rb
.column_by_name("parent_station")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::StringArray>())
.expect("parent_station present");
assert!(parent.is_null(0), "absent GTFS column → all-null column");
crate::functional_status(
"korp-collectors/gtfs-stops",
"gtfs_zip_stops_parse",
true,
&format!(
"{} GTFS stops → typed stop table (DEFLATE zip)",
rb.num_rows()
),
);
}
/// A STORED (uncompressed) member + a folder-prefixed path are both found by
/// basename, and the routes preset types route_type as i64. Two-member zip.
#[test]
fn gtfs_zip_stored_member_and_basename_lookup_for_routes() {
// routes STORED (method 0) under a folder prefix; a second file present.
let zip = build_zip(&[
("feed/routes.txt", 0, ROUTES.as_bytes()),
(
"feed/agency.txt",
0,
b"agency_id,agency_name\nSL,Storstockholm\n",
),
]);
let c = GtfsCollector::routes();
let rb = parse_gtfs_zip(&zip, c.file(), c.cols()).expect("parse routes");
assert_eq!(
rb.num_rows(),
1,
"one route parsed (found under feed/ prefix)"
);
let short = rb
.column_by_name("route_short_name")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::StringArray>())
.expect("route_short_name is utf8");
assert_eq!(short.value(0), "17");
let rtype = rb
.column_by_name("route_type")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Int64Array>())
.expect("route_type is i64");
assert_eq!(rtype.value(0), 1, "route_type → i64 (1 = metro/subway)");
// The raw ZIP reader sees both members.
let entries = unzip(&zip).expect("unzip");
assert_eq!(entries.len(), 2, "two members in the feed");
assert!(
zip_entry(&entries, "agency.txt").is_some(),
"basename lookup finds folder-nested file"
);
crate::functional_status(
"korp-collectors/gtfs-routes",
"gtfs_zip_stored_routes",
true,
&format!(
"{} GTFS route → route table (STORED zip, basename lookup)",
rb.num_rows()
),
);
}
/// A missing member is a hard error (not a silent empty), and a non-ZIP blob is
/// rejected — the RED-when-broken guards.
#[test]
fn gtfs_missing_member_and_non_zip_error() {
let zip = build_zip(&[("stops.txt", 8, STOPS.as_bytes())]);
let c = GtfsCollector::trips();
assert!(
parse_gtfs_zip(&zip, c.file(), c.cols()).is_err(),
"missing trips.txt → Err"
);
assert!(unzip(b"not a zip at all").is_err(), "non-ZIP bytes → Err");
}
}