Destinations CDK: better integration tests

[edgao]

Implement the test framework, write a minimal set of base tests, and implement those tests for destination-e2e-test. I stuck with our existing abstract class + concrete per-destination implementation strategy, because:

1. tests are opt-out
   1. like I mentioned in last (?) week's meeting - micronaut makes running the base class kind of wonky, so I think it's still best practice to have the override testFoo() { super.testFoo() } declarations. But even without that, we still get the test case.
2. it gives us an ok way to run individual test cases from inside intellij

general review guide:

1. DestinationMessage.kt has some simple convenience constructors
2. Start with IntegrationTest.kt, which is where the meat of the PR lives
3. Take a look at DestinationDataDumper, DestinationCleaner, ExpectedRecordMapper, and NameMapper - these are interfaces that destinations can/should implement. DataDumper doesn't have a noop implementation because every destination must implement it; everything else is optional.
4. Check out DestinationProcess, which is how we launch the connector via micronaut (and eventually Docker). It's basically just a wrapper around CliRunner.
5. CheckIntegrationTest + BasicFunctionalityIntegrationTest show how to write some basic tests.
6. Then skim through all the destination-e2e-test code - it's hopefully pretty self-explanatory after everything else in this PR

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Skipped Deployment

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	⬜️ Ignored (Inspect)	Visit Preview		Sep 19, 2024 9:59pm

[edgao]

Warning

This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
Learn more

• Bulk Load CDK: Add integration test using in-memory mock destination #45634
• Destinations cdk: implement snapshotting spec test #45468
• airbyte-ci: Ignore autogen files #45632
• Destinations CDK: better integration tests #45113 👈
• Bulk CDK: Add conveniencen method to BufferingOutputConsumer to track message consumption #45611
• Bulk CDK (Destinations): Use Descriptor instead of full Stream as map key #45610
• Destinations cdk: recorddiffer that handles strongly-typed records #44874
• Load CDK E2ETest Destination w/o tests #45147
• master

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @edgao and the rest of your teammates on Graphite

[edgao]

I'll revert this before merging, this was just driving me nuts locally

[edgao]

I didn't check what these two classes are doing, but will revive these tests before merging this PR

[edgao]

@johnny-schmidt before I go too far down this path - this (+the related plumbing code in CliRunner + AirbyteConnectorRunner) seems like a nicer way to inject custom beans to the destination? I.e. this way, test authors don't need to define a new InputStreamFactory + do property dancing

but lmk if you have a different idea for doing this

(this does have the slightly unfortunate result that in E2EDestination, we have to do .run(args = args) instead of .run(*args), b/c of varargs being a whiny child)

[johnny-schmidt]

If the idea is that we pass in config & catalog explicitly, would it make more sense to provide the input data explicitly also? Just wrap it in an serializing InputStream?

[edgao]

maaaybe, though I'm leaning towards no. There are tests that want to interleave messages and test logic, a la

sendSomeRecords()
verifySomething()
sendMoreRecords()
endSync()
verifySomethingElse()

At which point I'd rather wrap it in the destination process via the sendMessage function, rather than exposing the pipes to callers? I'm imagining something along these lines, which doesn't feel great?

val pipe = PipedOutputStream()
val pipeInputStream = PipedInputStream(pipe)
val dest = DestinationProcess(pipeInputStream)
messages1.each { pipe.write(serialize(it)) }
verifySomething()
messages2.each { pipe.write(serialize(it)) }
// flush all messages
pipe.close()
// this will magically cause the destination to shutdown
pipeInputStream.close()

plus I think we get most of the benefit by providing a utility method wrapping the destination process?

fun runSync(factory, config, catalog, messages) {
  val dest = factory.start(config, catalog)
  messages.each { dest.sendMessage(it) }
  return dest.readMessages()
}

[edgao]

this test case is mostly here to (a) be a quick smoke test on our state handling, and (b) show the typical interaction with DestinationProcess. Most tests we'll actually write don't need any of the state stuff, and would look more like:

runSync(config, stream, listOf(records...))
dumpAndDiffRecords(...)
// (potentially more syncs, if we're e.g. testing refreshes)

[edgao]

this is a hack. We get the integrationTestJava task from this plugin. https://github.com/airbytehq/airbyte-internal-issues/issues/9864 will fix.

[johnny-schmidt]

Does this mean that we never test the null namespace case?

[johnny-schmidt]

Dq, why is the cleaner its own thing and not an entry point into the process?

[johnny-schmidt]

This looks solid to me. If you want to merge it down into e2e I'll start working on the final productionalization.

I think overall we'll want to iterate on these interfaces (mine and yours) and work out how we want to handle configs, injection in general, mapping to and from schemas and records, all that. But we can learn as we go.