diff --git a/cli/index_run_e2e_test.ts b/cli/index_run_e2e_test.ts index 8cf900202..34de36143 100644 --- a/cli/index_run_e2e_test.ts +++ b/cli/index_run_e2e_test.ts @@ -3,6 +3,8 @@ import * as fs from "fs-extra"; import { dump as dumpYaml, load as loadYaml } from "js-yaml"; import * as path from "path"; +import * as dfapi from "df/cli/api"; +import { BigQueryDbAdapter } from "df/cli/api/dbadapters/bigquery"; import { alterWorkflowSettings, CREDENTIALS_PATH, @@ -14,9 +16,12 @@ import { } from "df/cli/index_test_base"; import { version } from "df/core/version"; import { suite, test, writeDefinitionFile } from "df/testing"; +import { dropDatasets, dropStaleDatasets, makeUniqueSuffix } from "df/testing/bigquery"; import { integrationSuite } from "df/testing/credentials"; import { TmpDirFixture } from "df/testing/fixtures"; +const ON_SCHEMA_CHANGE_DATASET = "dataform_e2e_osc"; + integrationSuite("run e2e", ({ afterEach }) => { const tmpDirFixture = new TmpDirFixture(afterEach); @@ -424,9 +429,22 @@ select 2 }); }); - suite("onSchemaChange", ({ beforeEach }) => { + suite("onSchemaChange", ({ before, beforeEach, after }) => { let projectDir: string; - const uniqueDataset = `dataform_e2e_osc_${Math.random().toString(36).substring(7)}`; + const dbadapter = new BigQueryDbAdapter(dfapi.credentials.read(CREDENTIALS_PATH)); + const uniqueDataset = `${ON_SCHEMA_CHANGE_DATASET}_${makeUniqueSuffix()}`; + + before("drop stale datasets", async () => { + await dropStaleDatasets(dbadapter, { + project: INTEGRATION_TEST_PROJECT, + location: INTEGRATION_TEST_LOCATION, + prefixes: [ON_SCHEMA_CHANGE_DATASET], + }); + }); + + after("drop datasets", async () => { + await dropDatasets(dbadapter, INTEGRATION_TEST_PROJECT, [uniqueDataset]); + }); beforeEach("setup test project", async () => { projectDir = tmpDirFixture.createNewTmpDir(); diff --git a/cli/index_test_base.ts b/cli/index_test_base.ts index 6da75ff48..a5ffa6c43 100644 --- a/cli/index_test_base.ts +++ b/cli/index_test_base.ts @@ -124,8 +124,9 @@ export function alterWorkflowSettings( export async function setupJitProject( tmpDirFixture: TmpDirFixture, projectDir: string, + workflowSettingsOverrides?: Partial, ): Promise { - await setupProject(tmpDirFixture, projectDir); + await setupProject(tmpDirFixture, projectDir, workflowSettingsOverrides); writeDefinitionFile( projectDir, "jit_table.js", diff --git a/cli/tests/jit/index_jit_advanced_test.ts b/cli/tests/jit/index_jit_advanced_test.ts index abf53293e..8d892ed06 100644 --- a/cli/tests/jit/index_jit_advanced_test.ts +++ b/cli/tests/jit/index_jit_advanced_test.ts @@ -1,12 +1,39 @@ import { expect } from "chai"; -import { CREDENTIALS_PATH, runCli, setupJitProject } from "df/cli/index_test_base"; +import * as dfapi from "df/cli/api"; +import { BigQueryDbAdapter } from "df/cli/api/dbadapters/bigquery"; +import { + CREDENTIALS_PATH, + INTEGRATION_TEST_LOCATION, + INTEGRATION_TEST_PROJECT, + runCli, + setupJitProject, +} from "df/cli/index_test_base"; import { test, writeDefinitionFile } from "df/testing"; +import { dropDatasets, dropStaleDatasets, makeUniqueSuffix } from "df/testing/bigquery"; import { integrationSuite } from "df/testing/credentials"; import { TmpDirFixture } from "df/testing/fixtures"; -integrationSuite("JiT support advanced", ({ afterEach }) => { +const JIT_INCREMENTAL_DATASET = "dataform_e2e_jit_inc"; + +integrationSuite("JiT support advanced", ({ afterEach, before, after }) => { const tmpDirFixture = new TmpDirFixture(afterEach); + const dbadapter = new BigQueryDbAdapter(dfapi.credentials.read(CREDENTIALS_PATH)); + const uniqueSuffix = makeUniqueSuffix(); + + before("drop stale datasets", async () => { + await dropStaleDatasets(dbadapter, { + project: INTEGRATION_TEST_PROJECT, + location: INTEGRATION_TEST_LOCATION, + prefixes: [JIT_INCREMENTAL_DATASET], + }); + }); + + after("drop datasets", async () => { + await dropDatasets(dbadapter, INTEGRATION_TEST_PROJECT, [ + `${JIT_INCREMENTAL_DATASET}_${uniqueSuffix}`, + ]); + }); test("JiT preOps and postOps support", async () => { const projectDir = tmpDirFixture.createNewTmpDir(); @@ -104,7 +131,9 @@ integrationSuite("JiT support advanced", ({ afterEach }) => { { name: "JiT incremental mode validation with consecutive runs", timeout: 60000 }, async () => { const projectDir = tmpDirFixture.createNewTmpDir(); - await setupJitProject(tmpDirFixture, projectDir); + await setupJitProject(tmpDirFixture, projectDir, { + defaultDataset: `${JIT_INCREMENTAL_DATASET}_${uniqueSuffix}`, + }); writeDefinitionFile( projectDir, "inc_jit.js", diff --git a/testing/BUILD b/testing/BUILD index 9059e2a89..3fc3ca4e0 100644 --- a/testing/BUILD +++ b/testing/BUILD @@ -17,6 +17,7 @@ js_library( ts_library( name = "testing", srcs = [ + "bigquery.ts", "child_process.ts", "credentials.ts", "fixtures.ts", diff --git a/testing/bigquery.ts b/testing/bigquery.ts new file mode 100644 index 000000000..cbcdc237b --- /dev/null +++ b/testing/bigquery.ts @@ -0,0 +1,77 @@ +import { randomBytes } from "crypto"; + +/** The minimal interface needed to run queries, satisfied by `BigQueryDbAdapter`. */ +export interface IQueryExecutor { + execute(statement: string): Promise<{ rows: any[] }>; +} + +const UNIQUE_SUFFIX_BYTES = 4; +const UNIQUE_SUFFIX_PATTERN = `[0-9a-f]{${UNIQUE_SUFFIX_BYTES * 2}}`; +const DATASET_PREFIX_REGEX = /^[A-Za-z0-9_]+$/; + +const DEFAULT_STALE_DATASET_MIN_AGE_HOURS = 6; + +export function makeUniqueSuffix(): string { + return randomBytes(UNIQUE_SUFFIX_BYTES).toString("hex"); +} + +/** Drops the given datasets, including their contents. Best effort: failures are only logged. */ +export async function dropDatasets( + executor: IQueryExecutor, + project: string, + datasets: string[], +): Promise { + await Promise.all( + datasets.map(async (dataset) => { + try { + await executor.execute(`drop schema if exists \`${project}.${dataset}\` cascade`); + } catch (e) { + console.warn(`Failed to drop dataset ${dataset}: ${e?.message || e}`); + } + }), + ); +} + +export interface IDropStaleDatasetsOptions { + project: string; + location: string; + prefixes: string[]; + minAgeHours?: number; +} + +export async function dropStaleDatasets( + executor: IQueryExecutor, + { + project, + location, + prefixes, + minAgeHours = DEFAULT_STALE_DATASET_MIN_AGE_HOURS, + }: IDropStaleDatasetsOptions, +): Promise { + if (prefixes.length === 0) { + return; + } + for (const prefix of prefixes) { + // Guards against SQL injection, and against an empty prefix matching every test's datasets. + if (!DATASET_PREFIX_REGEX.test(prefix)) { + throw new Error(`Invalid dataset name prefix: "${prefix}"`); + } + } + const namePattern = `^(${prefixes.join("|")})_${UNIQUE_SUFFIX_PATTERN}$`; + + let staleDatasets: string[]; + try { + const { rows } = await executor.execute( + `select schema_name + from \`${project}\`.\`region-${location.toLowerCase()}\`.INFORMATION_SCHEMA.SCHEMATA + where regexp_contains(schema_name, r'${namePattern}') + and creation_time < timestamp_sub(current_timestamp(), interval ${minAgeHours} hour)`, + ); + staleDatasets = rows.map((row: { schema_name: string }) => row.schema_name); + } catch (e) { + console.warn(`Failed to list stale datasets: ${e?.message || e}`); + return; + } + // Another run may be dropping the same datasets concurrently, which is fine. + await dropDatasets(executor, project, staleDatasets); +} diff --git a/testing/fixtures.ts b/testing/fixtures.ts index 8fec20a23..ef7ac0cda 100644 --- a/testing/fixtures.ts +++ b/testing/fixtures.ts @@ -5,6 +5,8 @@ import { IHookHandler } from "df/testing"; // TmpDirFixture can be used to create unique temporary directories which will be cleaned up // at the end of a test run. Intended for use within bazel tests. +// In a { parallel: true } suite, pass `after`, not `afterEach`: the cleanup deletes the directories +// of all tests, including ones that are still running. export class TmpDirFixture { private static dirCounter = 0; diff --git a/tests/integration/bigquery.spec.ts b/tests/integration/bigquery.spec.ts index 88ede2fd4..c10b83c77 100644 --- a/tests/integration/bigquery.spec.ts +++ b/tests/integration/bigquery.spec.ts @@ -3,15 +3,46 @@ import { expect } from "chai"; import * as dfapi from "df/cli/api"; import { BigQueryDbAdapter } from "df/cli/api/dbadapters/bigquery"; import { ExecutionSql } from "df/cli/api/dbadapters/execution_sql"; -import { CREDENTIALS_PATH } from "df/cli/index_test_base"; +import { + CREDENTIALS_PATH, + INTEGRATION_TEST_LOCATION, + INTEGRATION_TEST_PROJECT, +} from "df/cli/index_test_base"; import { dataform } from "df/protos/ts"; import { suite, test } from "df/testing"; +import { dropDatasets, dropStaleDatasets, makeUniqueSuffix } from "df/testing/bigquery"; import { integrationSuite } from "df/testing/credentials"; -import { cleanWarehouse, compile } from "df/tests/integration/utils"; +import { compile, datasetsOf } from "df/tests/integration/utils"; -integrationSuite("@dataform/integration/bigquery", { parallel: true }, () => { +const SEARCH_SCHEMA_SUFFIX = "search"; +const SEARCH_DATASET = `df_integration_test_${SEARCH_SCHEMA_SUFFIX}`; + +integrationSuite("@dataform/integration/bigquery", { parallel: true }, ({ before, after }) => { const credentials = dfapi.credentials.read(CREDENTIALS_PATH); const dbadapter = new BigQueryDbAdapter(credentials); + const uniqueSuffix = makeUniqueSuffix(); + + // The datasets created by the tests, without the unique suffix. + let datasets: string[]; + + before("drop stale datasets", async () => { + datasets = datasetsOf( + await compile("tests/integration/bigquery_project", SEARCH_SCHEMA_SUFFIX), + ); + await dropStaleDatasets(dbadapter, { + project: INTEGRATION_TEST_PROJECT, + location: INTEGRATION_TEST_LOCATION, + prefixes: datasets, + }); + }); + + after("drop datasets", async () => { + await dropDatasets( + dbadapter, + INTEGRATION_TEST_PROJECT, + datasets.map((dataset) => `${dataset}_${uniqueSuffix}`), + ); + }); test("run unit tests", async () => { const compiledGraph = await compile("tests/integration/bigquery_project", "unit_tests"); @@ -86,10 +117,11 @@ integrationSuite("@dataform/integration/bigquery", { parallel: true }, () => { }); test("search", async () => { - const compiledGraph = await compile("tests/integration/bigquery_project", "search"); - - // Drop all the tables before we do anything. - await cleanWarehouse(compiledGraph, dbadapter); + const compiledGraph = await compile( + "tests/integration/bigquery_project", + `${SEARCH_SCHEMA_SUFFIX}_${uniqueSuffix}`, + ); + const dataset = `${SEARCH_DATASET}_${uniqueSuffix}`; // Run the project. const executionGraph = await dfapi.build( @@ -106,8 +138,9 @@ integrationSuite("@dataform/integration/bigquery", { parallel: true }, () => { ); const [fullSearch, partialSearch, columnSearch] = await Promise.all([ - dbadapter.search("df_integration_test_search"), - dbadapter.search("test_sear"), + dbadapter.search(dataset), + // A partial match, that still only matches this run's dataset. + dbadapter.search(`test_search_${uniqueSuffix}`), dbadapter.search("val"), ]); diff --git a/tests/integration/evaluate.spec.ts b/tests/integration/evaluate.spec.ts index 249bcbc5f..68e62fc33 100644 --- a/tests/integration/evaluate.spec.ts +++ b/tests/integration/evaluate.spec.ts @@ -2,20 +2,57 @@ import { expect } from "chai"; import * as dfapi from "df/cli/api"; import { BigQueryDbAdapter } from "df/cli/api/dbadapters/bigquery"; -import { CREDENTIALS_PATH, INTEGRATION_TEST_PROJECT } from "df/cli/index_test_base"; +import { + CREDENTIALS_PATH, + INTEGRATION_TEST_LOCATION, + INTEGRATION_TEST_PROJECT, +} from "df/cli/index_test_base"; import { targetAsReadableString } from "df/core/targets"; import { dataform } from "df/protos/ts"; import { test } from "df/testing"; +import { dropDatasets, dropStaleDatasets, makeUniqueSuffix } from "df/testing/bigquery"; import { integrationSuite } from "df/testing/credentials"; -import { compile, keyBy } from "df/tests/integration/utils"; +import { compile, datasetsOf, keyBy } from "df/tests/integration/utils"; -integrationSuite("@dataform/integration/evaluate", () => { +const EVALUATE_SCHEMA_SUFFIX = "evaluate"; +const EVALUATE_DATASET = `df_integration_test_${EVALUATE_SCHEMA_SUFFIX}`; +const EVALUATE_ASSERTIONS_DATASET = `df_integration_test_assertions_${EVALUATE_SCHEMA_SUFFIX}`; + +integrationSuite("@dataform/integration/evaluate", ({ before, after }) => { const credentials = dfapi.credentials.read(CREDENTIALS_PATH); const dbadapter = new BigQueryDbAdapter(credentials); + const uniqueSuffix = makeUniqueSuffix(); + + // The datasets created by the tests, without the unique suffix. + let datasets: string[]; + + before("drop stale datasets", async () => { + datasets = datasetsOf( + await compile("tests/integration/bigquery_project", EVALUATE_SCHEMA_SUFFIX), + ); + await dropStaleDatasets(dbadapter, { + project: INTEGRATION_TEST_PROJECT, + location: INTEGRATION_TEST_LOCATION, + prefixes: datasets, + }); + }); + + after("drop datasets", async () => { + await dropDatasets( + dbadapter, + INTEGRATION_TEST_PROJECT, + datasets.map((dataset) => `${dataset}_${uniqueSuffix}`), + ); + }); test("evaluate from valid compiled graph as valid", async () => { // Create and run the project. - const compiledGraph = await compile("tests/integration/bigquery_project", "evaluate"); + const compiledGraph = await compile( + "tests/integration/bigquery_project", + `${EVALUATE_SCHEMA_SUFFIX}_${uniqueSuffix}`, + ); + const dataset = `${EVALUATE_DATASET}_${uniqueSuffix}`; + const assertionsDataset = `${EVALUATE_ASSERTIONS_DATASET}_${uniqueSuffix}`; const executionGraph = await dfapi.build(compiledGraph, {}, dbadapter); await dfapi.run(dbadapter, executionGraph).result(); @@ -27,44 +64,34 @@ integrationSuite("@dataform/integration/evaluate", () => { targetAsReadableString(t.target), ); - const view = - tablesByName[`${INTEGRATION_TEST_PROJECT}.df_integration_test_evaluate.example_view`]; + const view = tablesByName[`${INTEGRATION_TEST_PROJECT}.${dataset}.example_view`]; let evaluations = await dbadapter.evaluate(dataform.Table.create(view)); expect(evaluations.length).to.equal(1); expect(evaluations[0].status).to.equal(dataform.QueryEvaluation.QueryEvaluationStatus.SUCCESS); const materializedView = - tablesByName[ - `${INTEGRATION_TEST_PROJECT}.df_integration_test_evaluate.example_materialized_view` - ]; + tablesByName[`${INTEGRATION_TEST_PROJECT}.${dataset}.example_materialized_view`]; evaluations = await dbadapter.evaluate(dataform.Table.create(materializedView)); expect(evaluations.length).to.equal(1); expect(evaluations[0].status).to.equal(dataform.QueryEvaluation.QueryEvaluationStatus.SUCCESS); - const table = - tablesByName[`${INTEGRATION_TEST_PROJECT}.df_integration_test_evaluate.example_table`]; + const table = tablesByName[`${INTEGRATION_TEST_PROJECT}.${dataset}.example_table`]; evaluations = await dbadapter.evaluate(dataform.Table.create(table)); expect(evaluations.length).to.equal(1); expect(evaluations[0].status).to.equal(dataform.QueryEvaluation.QueryEvaluationStatus.SUCCESS); - const operation = - operationsByName[ - `${INTEGRATION_TEST_PROJECT}.df_integration_test_evaluate.example_operation` - ]; + const operation = operationsByName[`${INTEGRATION_TEST_PROJECT}.${dataset}.example_operation`]; evaluations = await dbadapter.evaluate(dataform.Operation.create(operation)); expect(evaluations.length).to.equal(1); expect(evaluations[0].status).to.equal(dataform.QueryEvaluation.QueryEvaluationStatus.SUCCESS); const assertion = - assertionsByName[ - `${INTEGRATION_TEST_PROJECT}.df_integration_test_assertions_evaluate.example_assertion_pass` - ]; + assertionsByName[`${INTEGRATION_TEST_PROJECT}.${assertionsDataset}.example_assertion_pass`]; evaluations = await dbadapter.evaluate(dataform.Assertion.create(assertion)); expect(evaluations.length).to.equal(1); expect(evaluations[0].status).to.equal(dataform.QueryEvaluation.QueryEvaluationStatus.SUCCESS); - const incremental = - tablesByName[`${INTEGRATION_TEST_PROJECT}.df_integration_test_evaluate.example_incremental`]; + const incremental = tablesByName[`${INTEGRATION_TEST_PROJECT}.${dataset}.example_incremental`]; evaluations = await dbadapter.evaluate(dataform.Table.create(incremental)); expect(evaluations.length).to.equal(2); expect(evaluations[0].status).to.equal(dataform.QueryEvaluation.QueryEvaluationStatus.SUCCESS); diff --git a/tests/integration/jit.spec.ts b/tests/integration/jit.spec.ts index 05f03ef74..d75f275db 100644 --- a/tests/integration/jit.spec.ts +++ b/tests/integration/jit.spec.ts @@ -1,5 +1,6 @@ import { expect } from "chai"; import * as fs from "fs-extra"; +import * as os from "os"; import * as path from "path"; import * as dfapi from "df/cli/api"; @@ -11,13 +12,27 @@ import { } from "df/cli/index_test_base"; import { dataform } from "df/protos/ts"; import { test, writeDefinitionFile } from "df/testing"; +import { dropDatasets, dropStaleDatasets, makeUniqueSuffix } from "df/testing/bigquery"; import { integrationSuite } from "df/testing/credentials"; -import { TmpDirFixture } from "df/testing/fixtures"; -integrationSuite("@dataform/integration/jit", { parallel: true }, ({ afterEach }) => { +// The unique suffix is appended to these names. +const JIT_DATASET = "df_integration_test_jit"; +const JIT_DRY_RUN_DATASET = "df_integration_test_jit_dry_run"; + +const DATASETS = [JIT_DATASET, JIT_DRY_RUN_DATASET]; + +integrationSuite("@dataform/integration/jit", { parallel: true }, ({ before }) => { const credentials = dfapi.credentials.read(CREDENTIALS_PATH); const dbadapter = new BigQueryDbAdapter(credentials); - const tmpDirFixture = new TmpDirFixture(afterEach); + const uniqueSuffix = makeUniqueSuffix(); + + before("drop stale datasets", async () => { + await dropStaleDatasets(dbadapter, { + project: INTEGRATION_TEST_PROJECT, + location: INTEGRATION_TEST_LOCATION, + prefixes: DATASETS, + }); + }); function setUpProject(projectDir: string, datasetName: string, tableName: string): void { fs.writeFileSync( @@ -47,67 +62,84 @@ defaultDataset: ${datasetName} ); } - test("JiT execution e2e", { timeout: 120000 }, async () => { - // Create a simple project with a JiT table - const projectDir = tmpDirFixture.createNewTmpDir(); - const datasetName = "df_integration_test_jit"; - const tableName = "jit_table"; - setUpProject(projectDir, datasetName, tableName); + // Each test owns its project directory and its dataset, and removes both when it ends. The tests + // run in parallel, so a cleanup shared by the suite (like TmpDirFixture with afterEach) would + // delete the project of a test that's still running. + async function withJitProject( + datasetPrefix: string, + fn: (projectDir: string, datasetName: string) => Promise, + ) { + // TEST_TMPDIR is set by Bazel. + const tmpDir = process.env.TEST_TMPDIR || os.tmpdir(); + const projectDir = fs.mkdtempSync(path.join(tmpDir, `${datasetPrefix}_`)); + const datasetName = `${datasetPrefix}_${uniqueSuffix}`; + try { + await fn(projectDir, datasetName); + } finally { + // Best effort: a cleanup error mustn't replace the error that failed the test, and one + // cleanup step failing mustn't skip the other. dropDatasets only logs its failures. + try { + fs.removeSync(projectDir); + } catch (e) { + process.stderr.write(`Could not remove ${projectDir}: ${e}\n`); + } + await dropDatasets(dbadapter, INTEGRATION_TEST_PROJECT, [datasetName]); + } + } - const compiledGraph = await dfapi.compile({ projectDir }); + function expectSuccessfulRun(runResult: dataform.IRunResult) { + // Include the action results, so that a failure shows the underlying error message. + expect( + dataform.RunResult.ExecutionStatus[runResult.status], + JSON.stringify(runResult.actions, null, 2), + ).eql(dataform.RunResult.ExecutionStatus[dataform.RunResult.ExecutionStatus.SUCCESSFUL]); + } - // Drop dataset to start fresh - await dbadapter.execute( - `drop schema if exists \`${INTEGRATION_TEST_PROJECT}.${datasetName}\` cascade`, - ); + test("JiT execution e2e", { timeout: 120000 }, async () => { + await withJitProject(JIT_DATASET, async (projectDir, datasetName) => { + // Create a simple project with a JiT table + const tableName = "jit_table"; + setUpProject(projectDir, datasetName, tableName); - const executionGraph = await dfapi.build(compiledGraph, {}, dbadapter); - const runResult = await dfapi.run(dbadapter, executionGraph, { projectDir }).result(); + const compiledGraph = await dfapi.compile({ projectDir }); + const executionGraph = await dfapi.build(compiledGraph, {}, dbadapter); + const runResult = await dfapi.run(dbadapter, executionGraph, { projectDir }).result(); - expect(dataform.RunResult.ExecutionStatus[runResult.status]).eql( - dataform.RunResult.ExecutionStatus[dataform.RunResult.ExecutionStatus.SUCCESSFUL], - ); + expectSuccessfulRun(runResult); - const rows = await dbadapter - .execute(`SELECT * FROM \`${INTEGRATION_TEST_PROJECT}.${datasetName}.${tableName}\``) - .then((res) => res.rows); - expect(rows).to.eql([{ id: 1 }]); + const rows = await dbadapter + .execute(`SELECT * FROM \`${INTEGRATION_TEST_PROJECT}.${datasetName}.${tableName}\``) + .then((res) => res.rows); + expect(rows).to.eql([{ id: 1 }]); + }); }); test("JiT dry run integration", { timeout: 120000 }, async () => { - // Create a simple project with a JiT table - const projectDir = tmpDirFixture.createNewTmpDir(); - const datasetName = "df_integration_test_jit_dry_run"; - const tableName = "jit_table_dry_run"; - setUpProject(projectDir, datasetName, tableName); - - const compiledGraph = await dfapi.compile({ projectDir }); - - // Drop dataset to start fresh - await dbadapter.execute( - `drop schema if exists \`${INTEGRATION_TEST_PROJECT}.${datasetName}\` cascade`, - ); - - const executionGraph = await dfapi.build(compiledGraph, {}, dbadapter); - - const runResult = await dfapi - .run(dbadapter, executionGraph, { - projectDir, - bigquery: { dryRun: true }, - }) - .result(); - - expect(dataform.RunResult.ExecutionStatus[runResult.status]).eql( - dataform.RunResult.ExecutionStatus[dataform.RunResult.ExecutionStatus.SUCCESSFUL], - ); - - // Verify that the table was NOT created - const tables = await dbadapter.schemas(INTEGRATION_TEST_PROJECT).then((schemas) => { - if (!schemas.includes(datasetName)) { - return []; - } - return dbadapter.tables(INTEGRATION_TEST_PROJECT, datasetName); + await withJitProject(JIT_DRY_RUN_DATASET, async (projectDir, datasetName) => { + // Create a simple project with a JiT table + const tableName = "jit_table_dry_run"; + setUpProject(projectDir, datasetName, tableName); + + const compiledGraph = await dfapi.compile({ projectDir }); + const executionGraph = await dfapi.build(compiledGraph, {}, dbadapter); + + const runResult = await dfapi + .run(dbadapter, executionGraph, { + projectDir, + bigquery: { dryRun: true }, + }) + .result(); + + expectSuccessfulRun(runResult); + + // Verify that the table was NOT created + const tables = await dbadapter.schemas(INTEGRATION_TEST_PROJECT).then((schemas) => { + if (!schemas.includes(datasetName)) { + return []; + } + return dbadapter.tables(INTEGRATION_TEST_PROJECT, datasetName); + }); + expect(tables.length).to.equal(0); }); - expect(tables.length).to.equal(0); }); }); diff --git a/tests/integration/property_graph.spec.ts b/tests/integration/property_graph.spec.ts index f4bd0fe89..f4acfb4d1 100644 --- a/tests/integration/property_graph.spec.ts +++ b/tests/integration/property_graph.spec.ts @@ -1,88 +1,106 @@ import { expect } from "chai"; -import { randomBytes } from "crypto"; import * as dfapi from "df/cli/api"; import { BigQueryDbAdapter } from "df/cli/api/dbadapters/bigquery"; -import { CREDENTIALS_PATH, INTEGRATION_TEST_PROJECT } from "df/cli/index_test_base"; +import { + CREDENTIALS_PATH, + INTEGRATION_TEST_LOCATION, + INTEGRATION_TEST_PROJECT, +} from "df/cli/index_test_base"; import { targetAsReadableString } from "df/core/targets"; import { dataform } from "df/protos/ts"; import { test } from "df/testing"; +import { dropDatasets, dropStaleDatasets, makeUniqueSuffix } from "df/testing/bigquery"; import { integrationSuite } from "df/testing/credentials"; -import { compile, keyBy } from "df/tests/integration/utils"; +import { compile, datasetsOf, keyBy } from "df/tests/integration/utils"; const GRAPH_NAME = "LibraryGraph"; -function makeSuffix() { - return process.env.GITHUB_RUN_ID ?? process.env.BUILD_ID ?? randomBytes(4).toString("hex"); -} +// The test compiles property_graph_project with this schema suffix, which dataform appends to the +// project's dataset. The unique suffix is appended to the names below. +const SCHEMA_SUFFIX = "e2e"; +const DATASET = `df_integration_test_pg_${SCHEMA_SUFFIX}`; -async function dropDataset(dbadapter: BigQueryDbAdapter, dataset: string) { - await dbadapter.execute( - `drop schema if exists \`${INTEGRATION_TEST_PROJECT}.${dataset}\` cascade`, - ); -} +integrationSuite( + "@dataform/integration/property_graph", + { parallel: true }, + ({ before, after }) => { + const credentials = dfapi.credentials.read(CREDENTIALS_PATH); + const uniqueSuffix = makeUniqueSuffix(); + const uniqueDataset = `${DATASET}_${uniqueSuffix}`; + const graphTarget = `${INTEGRATION_TEST_PROJECT}.${uniqueDataset}.${GRAPH_NAME}`; + let dbadapter: BigQueryDbAdapter; -integrationSuite("@dataform/integration/property_graph", { parallel: true }, (ctx) => { - const credentials = dfapi.credentials.read(CREDENTIALS_PATH); - const schemaSuffix = `e2e_${makeSuffix()}`; - const dataset = `df_integration_test_pg_${schemaSuffix}`; - const graphTarget = `${INTEGRATION_TEST_PROJECT}.${dataset}.${GRAPH_NAME}`; - let dbadapter: BigQueryDbAdapter; + before("create adapter", async () => { + dbadapter = new BigQueryDbAdapter(credentials); + }); - ctx.before("create adapter", async () => { - dbadapter = new BigQueryDbAdapter(credentials); - }); + // The datasets created by the test, without the unique suffix. + let datasets: string[]; - ctx.after("drop dataset", async () => { - // Tear-down hooks run even when a set-up hook (e.g. the ADC check registered by - // integrationSuite) fails, in which case the adapter was never created. - if (dbadapter) { - await dropDataset(dbadapter, dataset); - } - }); + before("drop stale datasets", async () => { + datasets = datasetsOf( + await compile("tests/integration/property_graph_project", SCHEMA_SUFFIX), + ); + await dropStaleDatasets(dbadapter, { + project: INTEGRATION_TEST_PROJECT, + location: INTEGRATION_TEST_LOCATION, + prefixes: datasets, + }); + }); - test("creates property graph end-to-end", { timeout: 120000 }, async () => { - const compiledGraph = await compile("tests/integration/property_graph_project", schemaSuffix); + after("drop datasets", async () => { + await dropDatasets( + dbadapter, + INTEGRATION_TEST_PROJECT, + datasets.map((dataset) => `${dataset}_${uniqueSuffix}`), + ); + }); - await dropDataset(dbadapter, dataset); + test("creates property graph end-to-end", { timeout: 120000 }, async () => { + const compiledGraph = await compile( + "tests/integration/property_graph_project", + `${SCHEMA_SUFFIX}_${uniqueSuffix}`, + ); - const executionGraph = await dfapi.build(compiledGraph, {}, dbadapter); - const executedGraph = await dfapi.run(dbadapter, executionGraph).result(); + const executionGraph = await dfapi.build(compiledGraph, {}, dbadapter); + const executedGraph = await dfapi.run(dbadapter, executionGraph).result(); - const actionMap = keyBy(executedGraph.actions, (v) => targetAsReadableString(v.target)); - expect(Object.keys(actionMap)).to.have.lengthOf(4); - for (const [name, action] of Object.entries(actionMap)) { - expect(action.status).equals( - dataform.ActionResult.ExecutionStatus.SUCCESSFUL, - `${name}: ${JSON.stringify(action, null, 2)}`, - ); - } + const actionMap = keyBy(executedGraph.actions, (v) => targetAsReadableString(v.target)); + expect(Object.keys(actionMap)).to.have.lengthOf(4); + for (const [name, action] of Object.entries(actionMap)) { + expect(action.status).equals( + dataform.ActionResult.ExecutionStatus.SUCCESSFUL, + `${name}: ${JSON.stringify(action, null, 2)}`, + ); + } - expect(actionMap).to.have.property(graphTarget); + expect(actionMap).to.have.property(graphTarget); - const rows = ( - await dbadapter.execute( - `select property_graph_catalog, property_graph_schema, + const rows = ( + await dbadapter.execute( + `select property_graph_catalog, property_graph_schema, property_graph_name, ddl - from \`${INTEGRATION_TEST_PROJECT}.${dataset}\`.INFORMATION_SCHEMA.PROPERTY_GRAPHS`, - ) - ).rows; - expect(rows).to.have.lengthOf(1); - const [row] = rows; - expect(row.property_graph_catalog).equals(INTEGRATION_TEST_PROJECT); - expect(row.property_graph_schema).equals(dataset); - expect(row.property_graph_name).equals(GRAPH_NAME); - for (const needle of [ - "NODE TABLES", - "EDGE TABLES", - "Author", - "Book", - "Wrote", - "authors", - "books", - "wrote", - ]) { - expect(row.ddl).to.contain(needle); - } - }); -}); + from \`${INTEGRATION_TEST_PROJECT}.${uniqueDataset}\`.INFORMATION_SCHEMA.PROPERTY_GRAPHS`, + ) + ).rows; + expect(rows).to.have.lengthOf(1); + const [row] = rows; + expect(row.property_graph_catalog).equals(INTEGRATION_TEST_PROJECT); + expect(row.property_graph_schema).equals(uniqueDataset); + expect(row.property_graph_name).equals(GRAPH_NAME); + for (const needle of [ + "NODE TABLES", + "EDGE TABLES", + "Author", + "Book", + "Wrote", + "authors", + "books", + "wrote", + ]) { + expect(row.ddl).to.contain(needle); + } + }); + }, +); diff --git a/tests/integration/run.spec.ts b/tests/integration/run.spec.ts index 777a9cc1b..e389586a4 100644 --- a/tests/integration/run.spec.ts +++ b/tests/integration/run.spec.ts @@ -3,27 +3,67 @@ import { expect } from "chai"; import * as dfapi from "df/cli/api"; import { BigQueryDbAdapter } from "df/cli/api/dbadapters/bigquery"; import { ExecutionSql } from "df/cli/api/dbadapters/execution_sql"; -import { CREDENTIALS_PATH, INTEGRATION_TEST_PROJECT } from "df/cli/index_test_base"; +import { + CREDENTIALS_PATH, + INTEGRATION_TEST_LOCATION, + INTEGRATION_TEST_PROJECT, +} from "df/cli/index_test_base"; import { targetAsReadableString } from "df/core/targets"; import { dataform } from "df/protos/ts"; import { test } from "df/testing"; +import { dropDatasets, dropStaleDatasets, makeUniqueSuffix } from "df/testing/bigquery"; import { integrationSuite } from "df/testing/credentials"; -import { cleanWarehouse, compile, getTableRows, keyBy } from "df/tests/integration/utils"; +import { compile, datasetsWithSuffixes, getTableRows, keyBy } from "df/tests/integration/utils"; -integrationSuite("@dataform/integration/run", { parallel: true }, () => { +// Each test compiles bigquery_project with its own schema suffix, which dataform appends to every +// dataset of the project. The unique suffix is appended to the names below. +const PROJECT_E2E_SCHEMA_SUFFIX = "project_e2e"; +const PROJECT_E2E_DATASET = `df_integration_test_${PROJECT_E2E_SCHEMA_SUFFIX}`; +const PROJECT_E2E_ASSERTIONS_DATASET = `df_integration_test_assertions_${PROJECT_E2E_SCHEMA_SUFFIX}`; + +const INCREMENTAL_TABLES_SCHEMA_SUFFIX = "incremental_tables"; +const INCREMENTAL_TABLES_DATASET = `df_integration_test_${INCREMENTAL_TABLES_SCHEMA_SUFFIX}`; + +const DATASET_METADATA_SCHEMA_SUFFIX = "dataset_metadata"; +const DATASET_METADATA_DATASET = `df_integration_test_${DATASET_METADATA_SCHEMA_SUFFIX}`; + +integrationSuite("@dataform/integration/run", { parallel: true }, ({ before, after }) => { const credentials = dfapi.credentials.read(CREDENTIALS_PATH); const dbadapter = new BigQueryDbAdapter(credentials); + const uniqueSuffix = makeUniqueSuffix(); + + // The datasets created by the tests, without the unique suffix. + let datasets: string[]; + + before("drop stale datasets", async () => { + datasets = datasetsWithSuffixes(await compile("tests/integration/bigquery_project", ""), [ + PROJECT_E2E_SCHEMA_SUFFIX, + INCREMENTAL_TABLES_SCHEMA_SUFFIX, + DATASET_METADATA_SCHEMA_SUFFIX, + ]); + await dropStaleDatasets(dbadapter, { + project: INTEGRATION_TEST_PROJECT, + location: INTEGRATION_TEST_LOCATION, + prefixes: datasets, + }); + }); - test("project e2e", { timeout: 60000 }, async () => { - const compiledGraph = await compile("tests/integration/bigquery_project", "project_e2e"); - - // Drop all the tables before we do anything. - await cleanWarehouse(compiledGraph, dbadapter); + after("drop datasets", async () => { + await dropDatasets( + dbadapter, + INTEGRATION_TEST_PROJECT, + datasets.map((dataset) => `${dataset}_${uniqueSuffix}`), + ); + }); - // Drop schemas to make sure schema creation works. - await dbadapter.execute( - `drop schema if exists \`${INTEGRATION_TEST_PROJECT}.df_integration_test_project_e2e\` cascade`, + test("project e2e", { timeout: 60000 }, async () => { + // The datasets are unique to this run, so they don't exist yet and schema creation is tested. + const compiledGraph = await compile( + "tests/integration/bigquery_project", + `${PROJECT_E2E_SCHEMA_SUFFIX}_${uniqueSuffix}`, ); + const dataset = `${PROJECT_E2E_DATASET}_${uniqueSuffix}`; + const assertionsDataset = `${PROJECT_E2E_ASSERTIONS_DATASET}_${uniqueSuffix}`; // Run the project. const executionGraph = await dfapi.build(compiledGraph, {}, dbadapter); @@ -34,8 +74,8 @@ integrationSuite("@dataform/integration/run", { parallel: true }, () => { // Check the status of action execution. const expectedFailedActions = [ - `${INTEGRATION_TEST_PROJECT}.df_integration_test_assertions_project_e2e.example_assertion_fail`, - `${INTEGRATION_TEST_PROJECT}.df_integration_test_project_e2e.example_operation_partial_fail`, + `${INTEGRATION_TEST_PROJECT}.${assertionsDataset}.example_assertion_fail`, + `${INTEGRATION_TEST_PROJECT}.${dataset}.example_operation_partial_fail`, ]; for (const actionName of Object.keys(actionMap)) { const expectedResult = expectedFailedActions.includes(actionName) @@ -48,23 +88,22 @@ integrationSuite("@dataform/integration/run", { parallel: true }, () => { } expect( - actionMap[ - `${INTEGRATION_TEST_PROJECT}.df_integration_test_assertions_project_e2e.example_assertion_fail` - ].tasks[1].errorMessage, + actionMap[`${INTEGRATION_TEST_PROJECT}.${assertionsDataset}.example_assertion_fail`].tasks[1] + .errorMessage, ).to.eql("bigquery error: Assertion failed: query returned 1 row(s)."); expect( - actionMap[ - `${INTEGRATION_TEST_PROJECT}.df_integration_test_project_e2e.example_operation_partial_fail` - ].tasks[0].errorMessage, + actionMap[`${INTEGRATION_TEST_PROJECT}.${dataset}.example_operation_partial_fail`].tasks[0] + .errorMessage, ).to.eql("bigquery error: Query error: Unrecognized name: invalid_column at [3:8]"); }); test("incremental tables", { timeout: 60000 }, async () => { - const compiledGraph = await compile("tests/integration/bigquery_project", "incremental_tables"); - - // Drop all the tables before we do anything. - await cleanWarehouse(compiledGraph, dbadapter); + const compiledGraph = await compile( + "tests/integration/bigquery_project", + `${INCREMENTAL_TABLES_SCHEMA_SUFFIX}_${uniqueSuffix}`, + ); + const dataset = `${INCREMENTAL_TABLES_DATASET}_${uniqueSuffix}`; // Run two iterations of the project. const adapter = new ExecutionSql( @@ -97,7 +136,7 @@ integrationSuite("@dataform/integration/run", { parallel: true }, () => { getTableRows( { database: INTEGRATION_TEST_PROJECT, - schema: "df_integration_test_incremental_tables", + schema: dataset, name: "example_incremental", }, adapter, @@ -106,7 +145,7 @@ integrationSuite("@dataform/integration/run", { parallel: true }, () => { getTableRows( { database: INTEGRATION_TEST_PROJECT, - schema: "df_integration_test_incremental_tables", + schema: dataset, name: "example_incremental_merge", }, adapter, @@ -119,10 +158,11 @@ integrationSuite("@dataform/integration/run", { parallel: true }, () => { }); test("dataset metadata set correctly", { timeout: 60000 }, async () => { - const compiledGraph = await compile("tests/integration/bigquery_project", "dataset_metadata"); - - // Drop all the tables before we do anything. - await cleanWarehouse(compiledGraph, dbadapter); + const compiledGraph = await compile( + "tests/integration/bigquery_project", + `${DATASET_METADATA_SCHEMA_SUFFIX}_${uniqueSuffix}`, + ); + const dataset = `${DATASET_METADATA_DATASET}_${uniqueSuffix}`; // Run the project. const executionGraph = await dfapi.build( @@ -143,7 +183,7 @@ integrationSuite("@dataform/integration/run", { parallel: true }, () => { { target: { database: INTEGRATION_TEST_PROJECT, - schema: "df_integration_test_dataset_metadata", + schema: dataset, name: "example_incremental", }, expectedDescription: "An incremental table", @@ -182,7 +222,7 @@ integrationSuite("@dataform/integration/run", { parallel: true }, () => { { target: { database: INTEGRATION_TEST_PROJECT, - schema: "df_integration_test_dataset_metadata", + schema: dataset, name: "example_view", }, expectedDescription: "An example view", diff --git a/tests/integration/utils.ts b/tests/integration/utils.ts index e2c8a6fee..114621126 100644 --- a/tests/integration/utils.ts +++ b/tests/integration/utils.ts @@ -16,16 +16,6 @@ export function keyBy(values: V[], keyFn: (value: V) => string): { [key: stri ); } -export async function dropAllTables( - tables: dataform.ITableMetadata[], - executionSql: ExecutionSql, - dbadapter: dbadapters.IDbAdapter, -) { - await Promise.all( - tables.map((table) => dbadapter.execute(executionSql.dropIfExists(table.target, table.type))), - ); -} - export async function getTableRows( target: dataform.ITarget, executionSql: ExecutionSql, @@ -34,6 +24,23 @@ export async function getTableRows( return (await dbadapter.execute(`SELECT * FROM ${executionSql.resolveTarget(target)}`)).rows; } +export function datasetsOf(compiledGraph: dataform.ICompiledGraph): string[] { + const actions: Array<{ target?: dataform.ITarget }> = [ + ...(compiledGraph.tables || []), + ...(compiledGraph.assertions || []), + ...(compiledGraph.operations || []), + ]; + return [...new Set(actions.map((action) => action.target?.schema).filter(Boolean))]; +} + +export function datasetsWithSuffixes( + compiledGraph: dataform.ICompiledGraph, + schemaSuffixes: string[], +): string[] { + const datasets = datasetsOf(compiledGraph); + return schemaSuffixes.flatMap((suffix) => datasets.map((dataset) => `${dataset}_${suffix}`)); +} + export async function compile( projectDir: string, schemaSuffixOverride: string, @@ -55,14 +62,3 @@ export async function compile( }; return compiledGraph; } - -export async function cleanWarehouse( - compiledGraph: dataform.CompiledGraph, - dbadapter: dbadapters.IDbAdapter, -) { - await dropAllTables( - (await dfapi.build(compiledGraph, {}, dbadapter)).warehouseState.tables, - new ExecutionSql(compiledGraph.projectConfig, compiledGraph.dataformCoreVersion), - dbadapter, - ); -}