`src/ingest/parcels/provenance_selftest.ts`

102 lines
/**
 * TK-50 field-level-provenance self-test (run: `tsx src/ingest/parcels/provenance_selftest.ts`).
 *
 * No test framework is installed in this repo, so this ships as a deterministic
 * self-test that exits non-zero on any failure. It:
 *   (a) upserts a synthetic parcel carrying record-level provenance
 *       (source_url / source_key / fetched_at / raw_source) + registers a
 *       source_field_map for the synthetic source,
 *   (b) asserts all four provenance columns persisted,
 *   (c) asserts a "where did field X come from?" resolution =
 *         parcel.source_url + source_field_map[source_key][X] + raw_source[source_attr],
 *   (d) purges every synthetic row it created.
 *
 * Uses a clearly-synthetic county_fips ('99999') + source_key ('selftest-tk50')
 * so it can never collide with, or leave residue in, real ingested data.
 */
import { pool, query } from '../../../db/pool.ts';
import { upsertParcels, registerSourceFieldMap, type ParcelUpsertRow } from './upsert.ts';

const FIPS = '99999';
const APN = 'TK50-SELFTEST-APN';
const KEY = 'selftest-tk50';

let failed = 0;
function check(ok: boolean, label: string, detail?: unknown) {
  if (ok) { console.log(`  ok   ${label}`); return; }
  failed++;
  console.error(`  FAIL ${label}${detail !== undefined ? `\n       ${JSON.stringify(detail)}` : ''}`);
}

async function purge() {
  await query(`DELETE FROM parcel WHERE county_fips=$1 AND source_id=$2`, [FIPS, APN]);
  await query(`DELETE FROM source_field_map WHERE source_key=$1`, [KEY]);
}

async function main() {
  await purge(); // clean slate in case a prior run aborted before its own purge

  const fetchedAt = new Date().toISOString();
  // The raw source record we "obtained" the data from. owner_name maps from
  // OWNER_NAME1, last_sale_price from PRICE — declared in the field map below.
  const raw = { PARCEL_ID: APN, OWNER_NAME1: 'JANE Q SYNTHETIC', PRICE: 725000, SITE_ADDR: '1 TEST WAY', CITY: 'TESTVILLE' };
  const sourceUrl = `https://example.invalid/arcgis/query?where=PARCEL_ID='${APN}'&f=html`;

  const row: ParcelUpsertRow = {
    county_fips: FIPS, source_id: APN,
    address: '1 TEST WAY', norm_address: '1 TEST WAY', city: 'TESTVILLE', zip: '00000',
    lat: null, lng: null, year_built: null, sqft: null, beds: null, baths: null,
    units: null, use_desc: null, land_value: null, improvement_value: null, total_value: null,
    tax_year: null, owner_name: 'JANE Q SYNTHETIC', zoning: null,
    last_sale_date: '2025-01-15', last_sale_price: 725000,
    extra: JSON.stringify({ source: KEY }),
    sourceKey: KEY, sourceUrl, fetchedAt, rawSource: JSON.stringify(raw),
  };

  await upsertParcels([row]);
  await registerSourceFieldMap(KEY, { owner_name: 'OWNER_NAME1', last_sale_price: 'PRICE', address: 'SITE_ADDR', city: 'CITY' }, 'TK-50 self-test source');

  // (b) provenance columns persisted on the parcel row
  const p = await query<{ source_url: string; source_key: string; fetched_at: string; raw_source: Record<string, unknown> }>(
    `SELECT source_url, source_key, fetched_at, raw_source FROM parcel WHERE county_fips=$1 AND source_id=$2`, [FIPS, APN]);
  check(p.rows.length === 1, 'parcel row exists', p.rows.length);
  const pr = p.rows[0] || ({} as any);
  check(pr.source_url === sourceUrl, 'source_url persisted', pr.source_url);
  check(pr.source_key === KEY, 'source_key persisted', pr.source_key);
  check(pr.fetched_at != null, 'fetched_at persisted', pr.fetched_at);
  check(pr.raw_source != null && (pr.raw_source as any).OWNER_NAME1 === 'JANE Q SYNTHETIC', 'raw_source JSONB persisted', pr.raw_source);

  // (c) "where did field X come from?" resolution for X = owner_name.
  // A field traces to: parcel.source_url (the record link)
  //   + source_field_map[source_key][X].source_attr (which raw attribute)
  //   + raw_source[source_attr] (the raw value we mapped from).
  const X = 'owner_name';
  const res = await query<{ source_url: string; source_attr: string; raw_value: string }>(
    `SELECT p.source_url,
            m.source_attr,
            p.raw_source ->> m.source_attr AS raw_value
       FROM parcel p
       JOIN source_field_map m
         ON m.source_key = p.source_key AND m.our_field = $3
      WHERE p.county_fips = $1 AND p.source_id = $2`,
    [FIPS, APN, X]);
  check(res.rows.length === 1, `provenance resolves for field '${X}'`, res.rows.length);
  const rr = res.rows[0] || ({} as any);
  check(rr.source_url === sourceUrl, `  ${X} → source_url`, rr.source_url);
  check(rr.source_attr === 'OWNER_NAME1', `  ${X} → source_attr = OWNER_NAME1`, rr.source_attr);
  check(rr.raw_value === 'JANE Q SYNTHETIC', `  ${X} → raw value from that attr`, rr.raw_value);
  console.log(`  resolved: ${X} came from ${rr.source_url} :: attr '${rr.source_attr}' = '${rr.raw_value}'`);

  // (d) purge synthetic rows
  await purge();
  const gone = await query(`SELECT 1 FROM parcel WHERE county_fips=$1 AND source_id=$2`, [FIPS, APN]);
  const mapGone = await query(`SELECT 1 FROM source_field_map WHERE source_key=$1`, [KEY]);
  check(gone.rows.length === 0 && mapGone.rows.length === 0, 'synthetic rows purged', { parcel: gone.rows.length, map: mapGone.rows.length });

  await pool.end();
  if (failed) { console.error(`\n${failed} provenance self-test check(s) FAILED`); process.exit(1); }
  console.log('\nprovenance self-test: all checks passed');
}

main().catch(async (e) => { console.error(e); try { await purge(); await pool.end(); } catch {} process.exit(1); });