# Copyright Materialize, Inc. and contributors. All rights reserved. # # Use of this software is governed by the Business Source License # included in the LICENSE file at the root of this repository. # # As of the Change Date specified in that file, in accordance with # the Business Source License, use of this software will be governed # by the Apache License, Version 2.0. """Tests for the Materialize Fivetran destination. This composition is a lightweight test harness for the Materialize Fivetran destination. Each test is structured as a directory within whose name begins with `test-`. Each test directory may contain: * any number of Testdrive scripts, which are files whose name ends in `.td` * any number of Fivetran Destination Tester scripts, which are files whose name ends in `.json` * a README file, which must be named `00-README` The test harness boots Materialize and the Materialize Fivetran Destination server. At the start of each test, the harness creates a database named `test` that is owned by the `materialize` user. Then, the test harness runs each Testdrive and Fivetran Destination Tester script within the test directory in lexicographic order. If a script fails, the test is marked as failed and no further scripts from the test are executed. A script is normally considered to fail if Testdrive or the Fivetran Destination Tester exit with a non-zero code. However, if the last line of a Fivetran Destination Tester script matches the pattern `// FAIL: `, the test harness will expect the Fivetran Destination Tester to exit with a non-zero code and with `` printed to stdout; the test script will be marked as failed if it exits with code zero or if `` is not printed to stdout. For details on Testdrive, consult doc/developer/testdrive.md. For details on the Fivetran Destination Tester, which is a tool provided by Fivetran, consult misc/fivetran-sdk/tools/README.md. To invoke the test harness locally: $ cd test/fivetran-destination $ ./mzcompose [--dev] run default -- [FILTER] The optional FILTER argument indicates a pattern which limits which test cases are run. A pattern matches a test case if the pattern is contained within the name of the test directory. """ import shutil from pathlib import Path from materialize.mzcompose.composition import Composition, WorkflowArgumentParser from materialize.mzcompose.services.fivetran_destination import FivetranDestination from materialize.mzcompose.services.fivetran_destination_tester import ( FivetranDestinationTester, ) from materialize.mzcompose.services.materialized import Materialized from materialize.mzcompose.services.mz import Mz from materialize.mzcompose.services.testdrive import Testdrive ROOT = Path(__file__).parent SERVICES = [ Mz(app_password=""), Materialized(), Testdrive( no_reset=True, default_timeout="5s", ), FivetranDestination( volumes_extra=["./data:/data"], ), FivetranDestinationTester( destination_host="fivetran-destination", destination_port=6874, volumes_extra=["./data:/data"], ), ] # Tests that are currently broken because the Fivetran Tester seems to do the wrong thing. BROKEN_TESTS = [] def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None: parser.add_argument("filter", nargs="?") args = parser.parse_args() c.up("materialized", "fivetran-destination") for path in ROOT.iterdir(): if path.name.startswith("test-"): if args.filter and args.filter not in path.name: print(f"Test case {path.name!r} does not match filter; skipping...") continue if path.name in BROKEN_TESTS: print(f"Test case {path.name!r} is currently broken; skipping...") continue with c.test_case(path.name): _run_test_case(c, path) def _run_test_case(c: Composition, path: Path): c.sql("DROP DATABASE IF EXISTS test") c.sql("CREATE DATABASE test") c.sql('DROP CLUSTER IF EXISTS "name with space" CASCADE') c.sql("CREATE CLUSTER \"name with space\" SIZE '1'") for test_file in sorted(p for p in path.iterdir()): test_file = test_file.relative_to(ROOT) if test_file.suffix == ".td": c.run_testdrive_files(str(test_file)) elif test_file.suffix == ".json": _run_destination_tester(c, test_file) elif test_file.name != "00-README": raise RuntimeError(f"unexpected test file: {test_file}") # Run the Fivetran Destination Tester with a single file. def _run_destination_tester(c: Composition, test_file: Path): # The Fivetran Destination tester operates on an entire directory at a time. We run # individual test cases by copying everything into a single "data" directory which # automatically gets cleaned up at the start and end of every run. with DataDirGuard(ROOT / "data") as data_dir: test_file = ROOT / test_file shutil.copy(test_file, data_dir.path()) last_line = test_file.read_text().splitlines()[-1] if last_line.startswith("// FAIL: "): expected_failure = last_line.removeprefix("// FAIL: ") else: expected_failure = None if expected_failure: ret = c.run("fivetran-destination-tester", check=False, capture=True) print("stdout:") print(ret.stdout) assert ( ret.returncode != 0 ), "destination tester returned success code when expected failure" # TODO(parkmycar): Re-enable this assertion when the Fivetran Destination Tester starts # outputting errors again. # # assert ( # expected_failure in ret.stderr # ), f"destination tester did not fail with expected message {expected_failure!r}\n\tfound: {ret.stdout!r}" else: c.run("fivetran-destination-tester") # Type that implements the Context Protocol that makes it easy to automatically clean up our data # directory before and after every test run. class DataDirGuard: def __init__(self, dir: Path): self._dir = dir def clean(self): for file in self._dir.iterdir(): if file.name in ("configuration.json", ".gitignore"): continue file.unlink() def path(self) -> Path: return self._dir def __enter__(self): self.clean() return self def __exit__(self, exc_type, exc_value, traceback): self.clean()