stephen_bailey
10/18/2021, 4:46 PMtap-slack
, I was thinking of having some automated tests that hit a sandbox workspace. My approach was basically:
1. Write a utility function to parse stdout
singer records into a summary object.
2. Write a utility function that takes a stream name, runs stream.sync()
, parses the records, and applies a catalog-schema-vs-record-schema test, and a record-count-greater-than-0 test.
3. Call the utliity function for each stream
Full code is here -- https://github.com/MeltanoLabs/tap-slack/blob/9bac8957413c16a146632ef47f921d4d8051684a/tests/test_core.py#L25 -- but the instantiated test is as simple as:
def test_channels_sync(capsys, tap_slack):
standard_stream_test(tap_slack, "channels", capsys)
Works well for top-level streams, but it looks like directly running ChildStream.sync()
will yield no records, because the context isn't passed in.
Curious if anyone has any thoughts or has implemented something similar?