0pt/tests/integration/data-quality-clickhouse.integration.test.ts

/**
 * tests/integration/data-quality-clickhouse.integration.test.ts
 * ==============================================================
 * BI-P025 — the native quality provider against a live ClickHouse.
 *
 * The unit suite proves the provider's LOGIC (an empty dataset is a failure, a
 * missing dbt result is a failure). This suite proves the part a fake runner
 * cannot: that the compiled expressions are valid ClickHouse and count the
 * right rows, including the empty-dataset case where a real engine returns
 * zero violations over zero rows and every naive implementation reports green.
 *
 * It also re-proves BI-P011's forced tenant predicate on this path: a quality
 * run executes on an ADMINISTRATIVE credential (it must read the base table),
 * so if the predicate were missing here, one tenant's dirty rows would
 * quarantine another tenant's projection.
 *
 * Gated on CLICKHOUSE_URL. With the variable set and the server down, this
 * suite FAILS rather than skipping.
 */
import fs from 'node:fs';
import path from 'node:path';
import { beforeAll, describe, expect, it } from 'vitest';
import { NativeQualityProvider, type QualityRuleSpec } from '@/src/analytics/quality/quality-provider';
import { ClickHouseQualityRunner } from '@/src/analytics/quality/clickhouse-quality-runner';
import { InMemoryQuarantineStore, QualityGate } from '@/src/analytics/quality/quarantine';

const BASE_URL = process.env.CLICKHOUSE_URL ?? '';
const AUTH = process.env.CLICKHOUSE_AUTH ?? 'default:';
const suite = BASE_URL ? describe : describe.skip;
const DB = 'opt_quality_test';
const EVIDENCE_PATH = path.join(process.cwd(), 'evidence/bi-p025/native-quality-live.json');

async function admin(sql: string): Promise<{ ok: boolean; body: string }> {
  const url = new URL(BASE_URL);
  url.searchParams.set('query', sql);
  const res = await fetch(url.toString(), {
    method: 'POST',
    headers: { Authorization: `Basic ${Buffer.from(AUTH).toString('base64')}` },
    body: '',
    signal: AbortSignal.timeout(60_000),
  });
  return { ok: res.ok, body: await res.text() };
}

const runner = new ClickHouseQualityRunner({
  baseUrl: BASE_URL || 'http://unset',
  basicAuth: AUTH,
  database: DB,
  tables: { ds_sales: 'sales', ds_empty: 'empty_sales' },
});
const provider = new NativeQualityProvider(runner);
const evidence: Record<string, unknown> = {};

function rule(over: Partial<QualityRuleSpec>): QualityRuleSpec {
  return {
    rule_id: 'q1',
    tenant_id: 't_alpha',
    dataset_id: 'ds_sales',
    rule: 'not_null',
    field: 'customer',
    severity: 'block',
    params: {},
    ...over,
  };
}

beforeAll(async () => {
  if (!BASE_URL) return;
  const v = await admin('SELECT version()');
  if (!v.ok) throw new Error('CLICKHOUSE_URL is set but ClickHouse is not answering');
  evidence.clickhouse_version = v.body.trim();
  evidence.recorded_at = new Date().toISOString();

  await admin(`DROP DATABASE IF EXISTS ${DB}`);
  await admin(`CREATE DATABASE ${DB}`);
  await admin(
    `CREATE TABLE ${DB}.sales (tenant_id String, dataset_id String, customer String, amount Float64, ` +
      `channel String, event_time DateTime) ENGINE = MergeTree ORDER BY (tenant_id, event_time)`,
  );
  await admin(
    `CREATE TABLE ${DB}.empty_sales (tenant_id String, dataset_id String, customer String, amount Float64, ` +
      `channel String, event_time DateTime) ENGINE = MergeTree ORDER BY (tenant_id, event_time)`,
  );

  // t_alpha: 100 clean rows, then deliberate defects.
  await admin(
    `INSERT INTO ${DB}.sales SELECT 't_alpha', 'ds_sales', 'cust_' || toString(number), number, 'web', now() ` +
      `FROM numbers(100)`,
  );
  // 5 rows with an empty customer (the not_null violation).
  await admin(
    `INSERT INTO ${DB}.sales SELECT 't_alpha', 'ds_sales', '', number, 'web', now() FROM numbers(5)`,
  );
  // 3 rows out of range (amount > 1000).
  await admin(
    `INSERT INTO ${DB}.sales SELECT 't_alpha', 'ds_sales', 'big_' || toString(number), 5000, 'web', now() ` +
      `FROM numbers(3)`,
  );
  // 4 rows with an unaccepted channel.
  await admin(
    `INSERT INTO ${DB}.sales SELECT 't_alpha', 'ds_sales', 'ch_' || toString(number), 1, 'carrier_pigeon', now() ` +
      `FROM numbers(4)`,
  );
  // 6 rows sharing two duplicate keys (3 each) — the unique violation.
  await admin(
    `INSERT INTO ${DB}.sales SELECT 't_alpha', 'ds_sales', 'dup_' || toString(number % 2), number, 'web', now() ` +
      `FROM numbers(6)`,
  );

  // t_beta: FILTHY. Every rule fails here. If the tenant predicate is missing,
  // t_alpha's counts jump and this tenant quarantines the other one.
  await admin(
    `INSERT INTO ${DB}.sales SELECT 't_beta', 'ds_sales', '', 999999, 'carrier_pigeon', now() FROM numbers(500)`,
  );
}, 120_000);

suite('BI-P025 live — the compiled expressions are valid SQL and count the right rows', () => {
  it("not_null finds exactly the 5 empty-customer rows, out of t_alpha's 118", async () => {
    const verdict = await provider.evaluate(rule({ rule: 'not_null', field: 'customer' }));
    expect(verdict.evaluated_rows).toBe(118);
    expect(verdict.violation_count).toBe(5);
    expect(verdict.passed).toBe(false);
    evidence.not_null = { evaluated: verdict.evaluated_rows, violations: verdict.violation_count };
  });

  it('range finds exactly the 3 rows over the maximum', async () => {
    const verdict = await provider.evaluate(
      rule({ rule: 'range', field: 'amount', params: { min: 0, max: 1000 } }),
    );
    expect(verdict.violation_count).toBe(3);
    evidence.range = { violations: verdict.violation_count };
  });

  it('distribution finds exactly the 4 rows outside the accepted values', async () => {
    const verdict = await provider.evaluate(
      rule({ rule: 'distribution', field: 'channel', params: { accepted_values: ['web', 'retail'] } }),
    );
    expect(verdict.violation_count).toBe(4);
    evidence.distribution = { violations: verdict.violation_count };
  });

  it('unique counts ROWS sitting on a duplicated key, not the number of keys', async () => {
    const verdict = await provider.evaluate(rule({ rule: 'unique', field: 'customer' }));
    // 11, and the arithmetic is worth writing down because the first version
    // of this test asserted 6 and the engine disagreed:
    //   * `dup_0` and `dup_1` appear 3 times each          -> 6 rows
    //   * the five EMPTY customers are all the same key ''  -> 5 rows
    // The empty strings are duplicates of one another, so `not_null` and
    // `unique` overlap on the same defect. That is the engine being right and
    // the fixture arithmetic being wrong, not a rule that over-counts.
    expect(verdict.violation_count).toBe(11);
    evidence.unique = {
      violations: verdict.violation_count,
      breakdown: '6 rows on dup_0/dup_1, plus 5 rows sharing the empty-string key',
    };
  });

  it('freshness finds no violation for rows written just now', async () => {
    const verdict = await provider.evaluate(
      rule({ rule: 'freshness', params: { timestamp_field: 'event_time', max_age_seconds: 3600 } }),
    );
    expect(verdict.violation_count).toBe(0);
    expect(verdict.passed).toBe(true);
  });

  it('a hostile accepted-value is bound, not executed', async () => {
    const verdict = await provider.evaluate(
      rule({
        rule: 'distribution',
        field: 'channel',
        params: { accepted_values: ['web', 'retail', "'); DROP TABLE sales; --"] },
      }),
    );
    expect(verdict.violation_count).toBe(4);
    // The table is still there, with its rows.
    const after = await admin(`SELECT count() FROM ${DB}.sales`);
    expect(Number(after.body.trim())).toBe(618);
  });
});

suite('BI-P025 live — the EMPTY dataset false green, against a real engine', () => {
  it('a real ClickHouse returns 0 violations over 0 rows, and the provider calls it a FAILURE', async () => {
    // The counterfactual first: the raw counts genuinely look clean.
    const raw = await runner.count({
      tenant_id: 't_alpha',
      dataset_id: 'ds_empty',
      violationExpression: "customer IS NULL OR customer = ''",
      params: {},
    });
    expect(raw).toEqual({ violations: 0, evaluated: 0 });

    // A naive implementation reports green here. This one does not.
    const verdict = await provider.evaluate(rule({ dataset_id: 'ds_empty' }));
    expect(verdict.passed).toBe(false);
    expect(verdict.detail).toContain('EMPTY');
    evidence.empty_dataset = {
      raw_counts_from_clickhouse: raw,
      verdict_passed: verdict.passed,
      note: 'a real engine returns zero violations over zero rows; the provider refuses to call that a pass',
    };
  });
});

suite("BI-P025 live — a quality run cannot read another tenant's rows", () => {
  it("t_alpha's counts are unaffected by t_beta's 500 filthy rows", async () => {
    const all = await admin(`SELECT count() FROM ${DB}.sales`);
    expect(Number(all.body.trim())).toBe(618); // 118 alpha + 500 beta

    const alpha = await provider.evaluate(
      rule({ tenant_id: 't_alpha', rule: 'not_null', field: 'customer' }),
    );
    const beta = await provider.evaluate(rule({ tenant_id: 't_beta', rule: 'not_null', field: 'customer' }));

    expect(alpha.evaluated_rows).toBe(118);
    expect(alpha.violation_count).toBe(5);
    expect(beta.evaluated_rows).toBe(500);
    expect(beta.violation_count).toBe(500);
    evidence.tenant_isolation = {
      base_table_rows: 618,
      alpha_evaluated: alpha.evaluated_rows,
      alpha_violations: alpha.violation_count,
      beta_evaluated: beta.evaluated_rows,
      beta_violations: beta.violation_count,
    };
  });

  it("t_beta's blocking failure quarantines t_beta only", async () => {
    const gate = new QualityGate(new InMemoryQuarantineStore());
    const beta = await provider.evaluate(rule({ tenant_id: 't_beta', rule: 'not_null', field: 'customer' }));
    await gate.apply('t_beta', 'ds_sales', [beta]);
    await expect(gate.assertProjectionReadable('t_beta', 'ds_sales')).rejects.toThrow(/QUARANTINED/);
    await expect(gate.assertProjectionReadable('t_alpha', 'ds_sales')).resolves.toBeUndefined();
  });

  it('refuses a dataset with no registered physical table rather than guessing one', async () => {
    await expect(provider.evaluate(rule({ dataset_id: 'ds_unregistered' }))).rejects.toThrow();
  });

  it('writes the evidence file', () => {
    fs.mkdirSync(path.dirname(EVIDENCE_PATH), { recursive: true });
    fs.writeFileSync(EVIDENCE_PATH, `${JSON.stringify(evidence, null, 2)}\n`);
    expect(fs.existsSync(EVIDENCE_PATH)).toBe(true);
  });
});