uscogdata/tests/testthat/test-views.R

test_that("all expected views register on session open", {
  skip_if_no_corpus()
  con <- cog_open()
  on.exit(cog_close())
  views <- DBI::dbGetQuery(con,
    "SELECT table_name FROM information_schema.tables
     WHERE table_schema = 'main' AND table_type = 'VIEW'"
  )
  expected <- c(
    "long", "spending_long", "revenue_long",
    "canonical_fips_xwalk", "summary_categories",
    "spending_annotated", "revenue_annotated",
    "ig_long", "ig_annotated",
    "ig_long_harmonized", "ig_annotated_harmonized"
  )
  expect_true(all(expected %in% views$table_name))
})

test_that("inst/sql/22- and 23- harmonized views enforce every WHERE predicate (real SQL text, synthetic parquet)", {
  # spending_long_harmonized / revenue_long_harmonized are three-predicate
  # views:
  #   SELECT * REPLACE (harmonized_code AS item_code)
  #   FROM long
  #   WHERE NOT is_aggregate
  #     AND harmonized_code IS NOT NULL
  #     AND LEFT(harmonized_code, 1) IN (<flow prefixes>)
  # None of the curated harmonization_map's `collapse` rulings land inside
  # the bundled fixture's 2011-2020 window for spending/revenue-prefixed
  # codes (see the "basis = 'harmonized' (default) matches 'raw'" test in
  # test-spending.R and docs/phase_r_harmonization_review.md § 0.2/§ 2), so
  # there is no real fixture row that exercises a nonzero fold or a
  # predicate-excluded row. Rather than re-implement the WHERE clause by
  # hand against an in-memory VALUES table (which would only prove the SQL
  # *pattern* works, not that the deployed inst/sql/22-/23- text actually
  # applies it), this test reads the real SQL files off disk, substitutes
  # {url} exactly as .register_views() does, and executes them -- plus
  # their 10-long.sql dependency -- against a synthetic hive-partitioned
  # parquet tree written to a temp dir. A regression in any predicate (e.g.
  # `NOT is_aggregate` dropped, the crosswalk-membership subquery changed,
  # the NULL guard removed) would change which of the rows below survive.
  #
  # The synthetic parquet is written with DuckDB's own COPY ... TO (FORMAT
  # PARQUET) rather than the arrow package: this package has no arrow
  # dependency (CLAUDE.md "No arrow dependency -- DuckDB reads parquet
  # natively"), and DuckDB can round-trip its own parquet writer/reader
  # without adding one for tests either.
  skip_if_no_corpus()

  tmp <- withr::local_tempdir()
  part_dir <- file.path(tmp, "data", "long", "year=2004")
  dir.create(part_dir, recursive = TRUE)
  part_path <- file.path(part_dir, "part-0.parquet")

  write_con <- DBI::dbConnect(duckdb::duckdb())
  on.exit(DBI::dbDisconnect(write_con, shutdown = TRUE), add = TRUE)
  DBI::dbExecute(write_con, sprintf("
    COPY (
      SELECT * FROM (VALUES
        -- Spending (E/F/G/K) rows, exercised against spending_long_harmonized:
        ('spend-A', 'E36', 100,    false, 'E36'), -- control: passes every predicate as-is
        ('spend-B', 'E38', 50,     false, 'E36'), -- collapse-fold: passes every predicate, renamed to E36
        ('spend-C', 'E05', 999999, true,  'E05'), -- excluded ONLY by `NOT is_aggregate`
        ('spend-D', 'E99', 888888, false, NULL),  -- excluded by `harmonized_code IS NOT NULL`
        -- 'S74' and 'Z61' are classified `balance` in the synthetic
        -- crosswalk below (mirroring the real corpus's own non-flow codes),
        -- so each is excluded from its view ONLY by the crosswalk-membership
        -- subquery -- the mechanism that replaced the prefix allowlists
        -- (uscogdata#11) and keeps balance stocks out of both flows
        -- (uscogdata#25).
        ('spend-E', 'S74', 777777, false, 'S74'), -- excluded ONLY by crosswalk membership (balance)
        -- Revenue rows, exercised against revenue_long_harmonized:
        ('rev-A',   'U11', 200,    false, 'U11'), -- control: passes every predicate as-is
        ('rev-B',   'U10', 25,     false, 'U11'), -- collapse-fold: passes every predicate, renamed to U11
        ('rev-C',   'T29', 555555, true,  'T29'), -- excluded ONLY by `NOT is_aggregate`
        ('rev-D',   'T88', 444444, false, NULL),  -- excluded by `harmonized_code IS NOT NULL`
        ('rev-E',   'Z61', 333333, false, 'Z61')  -- excluded ONLY by crosswalk membership (balance)
      ) AS t(canonical_govid, item_code, amt, is_aggregate, harmonized_code)
    ) TO %s (FORMAT PARQUET)
  ", uscogdata:::.sql_lit_chr(part_path)))

  # The flow views classify by membership in summary_categories, so the
  # synthetic corpus needs one too. Every flow code above is a member of its
  # own flow (so is_aggregate / NULL-harmonized exclusions stay the SOLE
  # excluder for those rows); S74/Z61 are members but classified balance, so
  # membership itself is what excludes them.
  DBI::dbExecute(write_con, sprintf("
    COPY (
      SELECT * FROM (VALUES
        ('E36', 'Water Utilities',  'expenditure', 'operations', NULL),
        ('E38', 'Water Utilities',  'expenditure', 'operations', NULL),
        ('E05', 'Corrections',      'expenditure', 'operations', NULL),
        ('E99', 'Other',            'expenditure', 'operations', NULL),
        ('S74', 'Fund Balances',    'balance',     NULL,         NULL),
        ('U11', 'Interest Earnings','revenue',     NULL,         'own_source'),
        ('U10', 'Interest Earnings','revenue',     NULL,         'own_source'),
        ('T29', 'Other Taxes',      'revenue',     NULL,         'own_source'),
        ('T88', 'Other Taxes',      'revenue',     NULL,         'own_source'),
        ('Z61', 'Fund Balances',    'balance',     NULL,         NULL)
      ) AS t(item_code, category, category_type, spend_subtype, revenue_subtype)
    ) TO %s (FORMAT PARQUET)
  ", uscogdata:::.sql_lit_chr(file.path(tmp, "data", "summary_categories.parquet"))))

  sql_dir <- system.file("sql", package = "uscogdata")
  .read_view_sql <- function(filename) {
    txt <- paste(readLines(file.path(sql_dir, filename), warn = FALSE), collapse = "\n")
    gsub("\\{url\\}", paste0(tmp, "/"), txt, fixed = FALSE)
  }

  con <- DBI::dbConnect(duckdb::duckdb())
  on.exit(DBI::dbDisconnect(con, shutdown = TRUE), add = TRUE)
  DBI::dbExecute(con, .read_view_sql("10-long.sql"))
  DBI::dbExecute(con, .read_view_sql("11-summary_categories.sql"))
  DBI::dbExecute(con, .read_view_sql("22-spending_long_harmonized.sql"))
  DBI::dbExecute(con, .read_view_sql("23-revenue_long_harmonized.sql"))

  spend <- DBI::dbGetQuery(con,
    "SELECT item_code, SUM(amt) AS amt FROM spending_long_harmonized
     GROUP BY item_code ORDER BY item_code"
  )
  # Exactly one surviving row: spend-C (aggregate), spend-D (NULL
  # harmonized_code), and spend-E (balance, not an expenditure member) must
  # all be gone, and spend-A + spend-B must be folded together under E36.
  expect_equal(nrow(spend), 1L)
  expect_equal(spend$item_code, "E36")
  expect_equal(spend$amt, 150)

  rev <- DBI::dbGetQuery(con,
    "SELECT item_code, SUM(amt) AS amt FROM revenue_long_harmonized
     GROUP BY item_code ORDER BY item_code"
  )
  expect_equal(nrow(rev), 1L)
  expect_equal(rev$item_code, "U11")
  expect_equal(rev$amt, 225)
})

test_that("inst/sql/24- and 25- IG views retain aggregates, COALESCE NULL harmonized_code, and exclude the L-- family total (real SQL text, synthetic parquet)", {
  # ig_long / ig_long_harmonized have the subtlest predicates in the package:
  # a deliberately ABSENT `NOT is_aggregate` (unlike every other *_long view),
  # and COALESCE(harmonized_code, item_code) instead of a plain
  # `harmonized_code IS NOT NULL` filter. The only end-to-end guard on this
  # today is bound to AL state / 2011 / Education K-12, where M12 happens to
  # be the sole IG code present -- regenerate the fixture without that one
  # row and the guard would die silently while staying green. As with the
  # 22-/23- test above, this reads the real inst/sql/24-/25- text off disk
  # and executes it against a synthetic hive-partitioned parquet tree, so a
  # regression in either predicate changes which rows survive.
  skip_if_no_corpus()

  tmp <- withr::local_tempdir()
  part_dir <- file.path(tmp, "data", "long", "year=2004")
  dir.create(part_dir, recursive = TRUE)
  part_path <- file.path(part_dir, "part-0.parquet")

  write_con <- DBI::dbConnect(duckdb::duckdb())
  on.exit(DBI::dbDisconnect(write_con, shutdown = TRUE), add = TRUE)
  DBI::dbExecute(write_con, sprintf("
    COPY (
      SELECT * FROM (VALUES
        ('ig-A', 'M04', 100,   false, 'M04'), -- control: passes through as-is
        ('ig-B', 'M38', 50,    false, 'M36'), -- fold control: real SB012 rule, renamed to M36 under harmonized basis
        ('ig-C', 'M47', 99999, true,  NULL),  -- legacy aggregate, NO harmonized_code: must survive BOTH views
        ('ig-D', 'L--', 55555, false, 'L--'), -- family total: deliberately NOT a crosswalk member, excluded from BOTH views
        ('ig-E', 'T29', 44444, false, 'T29')  -- revenue member, not intergovernmental: excluded from BOTH views
      ) AS t(canonical_govid, item_code, amt, is_aggregate, harmonized_code)
    ) TO %s (FORMAT PARQUET)
  ", uscogdata:::.sql_lit_chr(part_path)))

  # The IG views classify by summary_categories membership
  # (spend_subtype = 'intergovernmental'). L-- is deliberately absent --
  # exactly as it is from the real crosswalk -- which is what excludes it.
  DBI::dbExecute(write_con, sprintf("
    COPY (
      SELECT * FROM (VALUES
        ('M04', 'Corrections', 'expenditure', 'intergovernmental', NULL),
        ('M38', 'Health',      'expenditure', 'intergovernmental', NULL),
        ('M36', 'Health',      'expenditure', 'intergovernmental', NULL),
        ('M47', 'IG Other',    'expenditure', 'intergovernmental', NULL),
        ('T29', 'Other Taxes', 'revenue',     NULL,                'own_source')
      ) AS t(item_code, category, category_type, spend_subtype, revenue_subtype)
    ) TO %s (FORMAT PARQUET)
  ", uscogdata:::.sql_lit_chr(file.path(tmp, "data", "summary_categories.parquet"))))

  sql_dir <- system.file("sql", package = "uscogdata")
  .read_view_sql <- function(filename) {
    txt <- paste(readLines(file.path(sql_dir, filename), warn = FALSE), collapse = "\n")
    gsub("\\{url\\}", paste0(tmp, "/"), txt, fixed = FALSE)
  }

  con <- DBI::dbConnect(duckdb::duckdb())
  on.exit(DBI::dbDisconnect(con, shutdown = TRUE), add = TRUE)
  DBI::dbExecute(con, .read_view_sql("10-long.sql"))
  DBI::dbExecute(con, .read_view_sql("11-summary_categories.sql"))
  DBI::dbExecute(con, .read_view_sql("24-ig_long.sql"))
  DBI::dbExecute(con, .read_view_sql("25-ig_long_harmonized.sql"))

  raw <- DBI::dbGetQuery(con,
    "SELECT item_code, SUM(amt) AS amt FROM ig_long
     GROUP BY item_code ORDER BY item_code"
  )
  # L-- (family total, not a member) and T29 (revenue, not IG) are gone; the
  # aggregate row M47 survives -- proof `NOT is_aggregate` is absent from
  # ig_long.
  expect_equal(raw$item_code, c("M04", "M38", "M47"))
  expect_equal(raw$amt, c(100, 50, 99999))

  harmonized <- DBI::dbGetQuery(con,
    "SELECT item_code, SUM(amt) AS amt FROM ig_long_harmonized
     GROUP BY item_code ORDER BY item_code"
  )
  # M38 folds to M36 (real harmonized_code present); M47 keeps its raw code
  # via COALESCE(NULL, 'M47') -- proof the aggregate row is NOT dropped by
  # a plain `harmonized_code IS NOT NULL` filter. L-- and T29 stay excluded.
  expect_equal(harmonized$item_code, c("M04", "M36", "M47"))
  expect_equal(harmonized$amt, c(100, 50, 99999))
})

test_that(".build_series_break_refs matches fin_code + break_year window", {
  # No CODE-SPECIFIC series_breaks_pq row falls inside the bundled fixture's
  # 2011-2020 window (data-verified; see the "series_break_refs" test in
  # test-spending.R), so this proves the matching logic itself against the
  # live view + a synthetic year window that DOES hit a cataloged break
  # (SB075, fin_code E62, break_year 2005). The corpus-wide entries are a
  # separate path with its own coverage -- SB194 does sit at 2012, inside
  # the fixture window; see test-corpus-breaks.R.
  skip_if_no_corpus()
  con <- cog_open()
  on.exit(cog_close())
  refs <- uscogdata:::.build_series_break_refs(
    con, codes_observed = c("E62", "E04"), years = c(2003L, 2006L),
    schema_version = 5L
  )
  expect_true("SB075" %in% refs)
  expect_true("SB071" %in% refs)

  # Gated on schema_version >= 5 even when the codes/years would otherwise match.
  refs_v4 <- uscogdata:::.build_series_break_refs(
    con, codes_observed = c("E62"), years = c(2003L, 2006L), schema_version = 4L
  )
  expect_equal(refs_v4, character(0))
})

test_that("schema v5 harmonization views register when the corpus supports them", {
  skip_if_no_corpus()
  con <- cog_open()
  on.exit(cog_close())
  manifest <- uscogdata:::.uscogdata_env$manifest
  skip_if(as.integer(manifest$schema_version) < 5L, "fixture is schema_version < 5")

  views <- DBI::dbGetQuery(con,
    "SELECT table_name FROM information_schema.tables
     WHERE table_schema = 'main' AND table_type = 'VIEW'"
  )
  expected_v5 <- c(
    "spending_long_harmonized", "revenue_long_harmonized",
    "spending_annotated_harmonized", "revenue_annotated_harmonized",
    "harmonization_map", "harmonization_recipes", "series_breaks_pq"
  )
  expect_true(all(expected_v5 %in% views$table_name))
})

test_that(".harmonization_view_files guard is necessary: registration against a v4-shaped corpus (no harmonized_code column at all) succeeds only because the harmonized views are skipped", {
  # with_doctored_schema_version() (used elsewhere in this suite) only
  # rewrites manifest.json's schema_version -- the underlying `long` parquet
  # is still the bundled v6 fixture, which DOES have a harmonized_code
  # column, so it only proves the skip *happens*, not that it is *required*.
  # This test builds a genuinely v4-shaped corpus: `long` has no
  # harmonized_code column at all, matching a real pre-Phase-R2 publish
  # tree, and then shows two things: (1) the real .register_views(), gated
  # on manifest$schema_version, registers cleanly against it; (2) the exact
  # SQL text of a gated file (25-ig_long_harmonized.sql), executed directly
  # against the same corpus without the gate, fails -- proving the gate is
  # load-bearing, not incidental.
  tmp <- withr::local_tempdir()
  part_dir <- file.path(tmp, "data", "long", "year=2004")
  dir.create(part_dir, recursive = TRUE)
  part_path <- file.path(part_dir, "part-0.parquet")

  write_con <- DBI::dbConnect(duckdb::duckdb())
  on.exit(DBI::dbDisconnect(write_con, shutdown = TRUE), add = TRUE)
  DBI::dbExecute(write_con, sprintf("
    COPY (
      SELECT * FROM (VALUES
        ('gov-1', 'E36', 100, false, 500000, 2020)
      ) AS t(canonical_govid, item_code, amt, is_aggregate, population, popyear)
    ) TO %s (FORMAT PARQUET)
  ", uscogdata:::.sql_lit_chr(part_path)))

  xwalk_path <- file.path(tmp, "data", "canonical_fips_xwalk.parquet")
  DBI::dbExecute(write_con, sprintf("
    COPY (
      SELECT * FROM (VALUES
        ('gov-1', 'Test Gov', 1, 'County', '01', '001', NULL, 500000)
      ) AS t(canonical_govid, gov_name, govs_type, type_label, fips_state,
             fips_county, fips_place, population_acs)
    ) TO %s (FORMAT PARQUET)
  ", uscogdata:::.sql_lit_chr(xwalk_path)))

  cats_path <- file.path(tmp, "data", "summary_categories.parquet")
  DBI::dbExecute(write_con, sprintf("
    COPY (
      SELECT * FROM (VALUES
        ('E36', 'Test Category', 'expenditure', 'direct', NULL)
      ) AS t(item_code, category, category_type, spend_subtype, revenue_subtype)
    ) TO %s (FORMAT PARQUET)
  ", uscogdata:::.sql_lit_chr(cats_path)))

  # Confirm the synthetic `long` genuinely lacks harmonized_code (not just
  # NULL values -- the column itself must be absent) before trusting the
  # rest of this test.
  cols <- DBI::dbGetQuery(write_con, sprintf(
    "DESCRIBE SELECT * FROM read_parquet(%s)", uscogdata:::.sql_lit_chr(part_path)
  ))$column_name
  expect_false("harmonized_code" %in% cols)

  url <- paste0(tmp, "/")

  # (1) Full .register_views() against this v4-shaped corpus must succeed --
  # this is the behavior the guard exists to protect.
  con <- DBI::dbConnect(duckdb::duckdb())
  on.exit(DBI::dbDisconnect(con, shutdown = TRUE), add = TRUE)
  expect_no_error(
    uscogdata:::.register_views(con, url, manifest = list(schema_version = 4L))
  )
  views <- DBI::dbGetQuery(con,
    "SELECT table_name FROM information_schema.tables
     WHERE table_schema = 'main' AND table_type = 'VIEW'")$table_name
  expect_true(all(c("ig_long", "ig_annotated", "spending_annotated") %in% views))
  expect_false(any(c("ig_long_harmonized", "ig_annotated_harmonized",
                      "spending_long_harmonized") %in% views))

  # (2) Prove the gate is load-bearing: the exact SQL text of the skipped
  # file, executed directly (bypassing .register_views()'s schema_version
  # check) against the SAME corpus, fails because it references
  # long.harmonized_code, a column this corpus's `long` does not have.
  sql_dir <- system.file("sql", package = "uscogdata")
  .read_view_sql <- function(filename) {
    txt <- paste(readLines(file.path(sql_dir, filename), warn = FALSE), collapse = "\n")
    gsub("\\{url\\}", url, txt, fixed = FALSE)
  }
  con2 <- DBI::dbConnect(duckdb::duckdb())
  on.exit(DBI::dbDisconnect(con2, shutdown = TRUE), add = TRUE)
  DBI::dbExecute(con2, .read_view_sql("10-long.sql"))
  expect_error(DBI::dbExecute(con2, .read_view_sql("25-ig_long_harmonized.sql")))

  # Reconciling this test with the C2 guard (expenditure-concept review):
  # `ig_annotated`/`spending_annotated` registering cleanly above proves
  # only that CREATE VIEW binds against a `summary_categories` with no M/L
  # rows at all (this synthetic corpus's own summary_categories has a
  # single E36 row, see the COPY above) -- a LEFT JOIN never fails to
  # resolve regardless of what the joined-to table contains. It does NOT
  # mean querying expenditure_concept = "total" against this shape is safe:
  # exactly this corpus (schema_version reported as supported, but
  # summary_categories predates the M/L rows cog_pipeline PR #59 added) is
  # what .require_ig_categories() exists to catch at the *verb* level,
  # since PR #59 shipped those rows with no schema_version bump. Confirm
  # the new runtime guard actually fires against this same `con`.
  expect_error(
    uscogdata:::.require_ig_categories(con),
    class = "uscogdata_ig_categories_unsupported"
  )
})

test_that("spending_long carries exactly the non-IG expenditure crosswalk codes and excludes aggregates", {
  skip_if_no_corpus()
  con <- cog_open()
  on.exit(cog_close())

  # Classification is crosswalk membership, not prefixes (uscogdata#11):
  # every row's code must classify as expenditure and never as
  # intergovernmental (which lives in ig_long).
  stray <- DBI::dbGetQuery(con,
    "SELECT DISTINCT s.item_code
     FROM spending_long s
     LEFT JOIN summary_categories c USING (item_code)
     WHERE c.category_type IS DISTINCT FROM 'expenditure'
        OR c.spend_subtype = 'intergovernmental'"
  )$item_code
  expect_length(stray, 0L)

  # Balance codes are stocks, not flows -- they must never appear in a
  # spending result (uscogdata#25). Prefix filtering could not guarantee
  # this (W/X/Y/Z balance codes share letters with flow codes).
  balance_n <- DBI::dbGetQuery(con,
    "SELECT count(*) AS n FROM spending_long WHERE item_code IN (
       SELECT item_code FROM summary_categories WHERE category_type = 'balance'
     )"
  )$n
  expect_equal(balance_n, 0)

  agg_count <- DBI::dbGetQuery(con,
    "SELECT count(*) AS n FROM spending_long WHERE is_aggregate"
  )$n
  expect_equal(agg_count, 0)
})

test_that("revenue_long carries exactly the revenue crosswalk codes and excludes aggregates", {
  skip_if_no_corpus()
  con <- cog_open()
  on.exit(cog_close())

  # The view carries EVERY revenue subtype; which of Census's two published
  # concepts a query returns is decided per `revenue_concept` in R
  # (uscogdata#12), exactly as `expenditure_concept` narrows spending_long.
  stray <- DBI::dbGetQuery(con,
    "SELECT DISTINCT s.item_code
     FROM revenue_long s
     LEFT JOIN summary_categories c USING (item_code)
     WHERE c.category_type IS DISTINCT FROM 'revenue'"
  )$item_code
  expect_length(stray, 0L)

  # No balance stock ever appears in a revenue result (uscogdata#25).
  balance_n <- DBI::dbGetQuery(con,
    "SELECT count(*) AS n FROM revenue_long WHERE item_code IN (
       SELECT item_code FROM summary_categories WHERE category_type = 'balance'
     )"
  )$n
  expect_equal(balance_n, 0)

  agg_count <- DBI::dbGetQuery(con,
    "SELECT count(*) AS n FROM revenue_long WHERE is_aggregate"
  )$n
  expect_equal(agg_count, 0)
})

test_that("spending_annotated carries category + xwalk columns", {
  skip_if_no_corpus()
  con <- cog_open()
  on.exit(cog_close())
  row <- DBI::dbGetQuery(con,
    "SELECT * FROM spending_annotated LIMIT 1"
  )
  for (nm in c("canonical_govid", "item_code", "amt",
               "xwalk_gov_name", "govs_type", "population_acs",
               "category", "spend_subtype")) {
    expect_true(nm %in% names(row), info = paste("missing column:", nm))
  }
})

test_that("gov_population_yearly exposes one row per (year, canonical_govid)", {
  skip_if_no_corpus()
  with_fixture_corpus({
    con <- uscogdata:::.ensure_session()
    df <- DBI::dbGetQuery(
      con,
      "SELECT year, canonical_govid, population, popyear
       FROM gov_population_yearly
       WHERE canonical_govid = '121011212191'
       ORDER BY year"
    )
    expect_setequal(df$year, c(2011L, 2012L, 2019L, 2020L))
    expect_equal(nrow(df), 4L)
    expect_true(all(!is.na(df$population)))
    # Hardcoded values are from the bundled fixture (regenerated 2026-07-18
    # against cog_pipeline publish tree, pipeline_commit ece9b32, Phase R2
    # schema_version 5, years 2011/2012/2019/2020). Update if the fixture is
    # rebuilt against a different source vintage.
    expect_equal(df$population[df$year == 2011L], 1759591L)
    expect_equal(df$population[df$year == 2012L], 1819773L)
    expect_equal(df$population[df$year == 2019L], 1935878L)
    expect_equal(df$population[df$year == 2020L], 1952778L)
    # Uniqueness on (year, canonical_govid) across the whole view.
    dup <- DBI::dbGetQuery(
      con,
      "SELECT year, canonical_govid, COUNT(*) AS n
       FROM gov_population_yearly
       GROUP BY year, canonical_govid HAVING n > 1"
    )
    expect_equal(nrow(dup), 0L)
  })
})