Unit Testing in Beam Blog Post

[svetakvsundhar]

This blog posts details opinionated examples and practices for unit testing in Beam. Examples use the Python SDK.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[svetakvsundhar]

R: @liferoad
R: @rszper
R: @damccorm
cc: https://github.com/surjits254

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

[liferoad]

shall we mention a lot of testing utils under https://github.com/apache/beam/tree/master/sdks/python/apache_beam/testing?

[svetakvsundhar]

Are these useful for customers writing pipeline? IIUC these utils are designed more for tests written in the Beam repo

[robertwb]

Users should generally use assert_that, equal_to, and (for streaming) test stream.

[damccorm]

Note - this isn't rendering in https://apache-beam-website-pull-requests.storage.googleapis.com/31701/blog/index.html

I suspect it is because the date is in the future, but I'm not sure. If so, that's a neat way to gate blog posts for a given time I guess

[damccorm]

Hm, I also don't see it rendered at https://apache-beam-website-pull-requests.storage.googleapis.com/31701/blog/unit-testing-blog/index.html

Maybe something is malformatted? If you stage it locally with an older date, does it render?

[damccorm]

If needed, lets temporarily set this to an older date so we can review the staged version

[damccorm]

Looks like this is poorly formatted (indentation needed, too much white space). I'd also recommend defining examples=[1,2,3] and expected=[1,4,9] and using those in the pipeline rather than explicit lists

[damccorm]

Same as above - formatting seems off, + lets make an expected=['Strawberry','Carrot','Eggplant'] instead of an explicit list here

[damccorm]

Also, we're missing an equal_to. Lets make sure we quickly run these test snippets locally before submitting

[robertwb]

I commented on this above, but let's ensure they're continuously run.

[damccorm]

I find this example is confusing - is the custom_function just the identity function? We have examples of assert_that above, can we use that instead?

[robertwb]

I don't think this works, but I'm also trying to understand what it's even trying to do. You can't apply a ParDo to a raw pipeline object. ParDo accepts a DoFn, is custom_function(some_tuple) returning a DoFn that yields the tuple?

Again, if we actually ran this code we'd uncover these issues.

[damccorm]

Could we also point to some example tests here as well? The beam repo is probably the best place for example PTransform tests. For example, we have quite extensive RunInference tests -

beam/sdks/python/apache_beam/ml/inference/base_test.py

Line 262 in 736cf50

class RunInferenceBaseTest(unittest.TestCase):

- and we could add similarly well tested transforms in Java and Go

[svetakvsundhar]

Thanks for the feedback. To address the code running as intended, I plan to create a colab notebook with the runnable snippets, and link it to the blog post (as well as check it in to the Beam repo). Will create after the holiday.

[damccorm]

SGTM - at that point we may be able to ignore the date related comments anyways and just validate that it does indeed render the blog

[svetakvsundhar]

I've added a colab notebook and tested these examples locally. PTAL, once we have a consensus on the notebook examples, I will update the blog post and address the other comments.

[damccorm]

I wonder if a better way to do this would be to split beam.Map(str.strip) out into a separate function which can be called from the test. As it is, the test isn't actually invoking any of the code we've written.

A more interesting example might be:

def manipulate_strings(incoming_pcoll):
   return incoming_pcoll | beam.Map(str.strip) | beam.Map(str.upper)

The functions themselves don't need tested, but the beam transforms do. That would let you test the actual code you've written below with:

  with TestPipeline() as p:
    inputs = p | beam.Create(strings)
    output = manipulate_strings(inputs)
    assert_that(output, equal_to(expected))

[damccorm]

Basically, I don't like that you could totally change the user code (which you're supposed to be testing) without a test failing.

[robertwb]

Seeing output = manipulate_strings(inputs) makes me wonder how we could make composite PTransforms even easier/more natural.

[svetakvsundhar]

Basically, I don't like that you could totally change the user code (which you're supposed to be testing) without a test failing.

I see your point and think it's valid (I can make the change). Out of curiosity though, what if a users entire transform was an inbuilt function (like str.strip)? Would the guidance be that they wouldn't need to test the Beam transform?

[robertwb]

Is three any advantage to users of using TestPipeline?

[svetakvsundhar]

AFAICT, no distinct advantages outside of the reasons mentioned here, as well as it being the de-facto choice for tests from previous documentation.

[robertwb]

Thanks for writing this up, guidance on good unit testing practices for Beam is much needed.

[robertwb]

It would seems one of the main points of unit testing is to not have heavyweight dependencies. I wouldn't say skipping like this is best practices unless absolutely necessary.

[robertwb]

If compute_square is an ordinary Python function, I would recommend writing "ordinary" unit tests for it rather than testing it as part of a pipeline.

[robertwb]

Mocking often interacts poorly with serialization; I would avoid this when possible. (Also, are these examples automatically tested?

[robertwb]

CustomClass.custom_function() returns a DoFn? I'm a bit confused at what you're trying to test here.

[robertwb]

You cand ParDo(lambda). I would look into how snippets are used in the programming guide to ensure the code is (and remains) correct.

[robertwb]

I don't think this works, but I'm also trying to understand what it's even trying to do. You can't apply a ParDo to a raw pipeline object. ParDo accepts a DoFn, is custom_function(some_tuple) returning a DoFn that yields the tuple?

Again, if we actually ran this code we'd uncover these issues.

[robertwb]

Nit: If we want ints, lets use ints, if we want strings, let's use strings (like "a", "b", "c" or fruit names or whatever, not numeric strings).

[robertwb]

Java-style imports?

[robertwb]

So, the problem is one can't really test this pipeline without reproducing it in the test. I think there are a couple of ways of re-structuring the code to make it more testable (and realistic).

(1) For an end-to-end test, put your pipeline in a function that takes the input and output paths as parameters. Your production code will call this with, e.g. gcs paths, but your test could create temporary directories and files and validate that. This tests your whole pipeline, including IOs, is structured correct.

(2) Factor out the "processing" part of your pipeline into its own PTransform, or at least function. E.g. your pipeline would be

with beam.Pipeline(argv=self.args) as p:
  _ = (p
       | beam.io.ReadFromText("gs://my-storage-bucket/csv_location.csv")
       | ProcessData(...)
       | beam.io.WriteToText()

and then your unit test would look like

with beam.Pipeline(argv=self.args) as p:
  _ = (p
       | beam.Create(["some", "sample", "elements"])
       | ProcessData(...)
       | AssertEqualTo(["expected", "outputs"]))

or (equivalently, but less parallel)

with beam.Pipeline(argv=self.args) as p:
  output_pcoll = (p
       | beam.Create(["some", "sample", "elements"])
       | ProcessData(...))
    assert_that(output_pcoll, equal_to(...))

If writing a custom ProcessData PTransform is too much work, one could at least have

output_pcoll = process_data(input_pcoll)

[robertwb]

I am not following this at all. Where is get_record being called? Why is it being mocked?

[svetakvsundhar]

Hi @robertwb thanks for taking a look! The blog and the colab notebook are actually currently out of sync so that we could test runnable examples. Once aligned on those, I was going to update the blog. Anyhow, I will take a look at the comments and see where they apply.

[github-actions]

This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@beam.apache.org list. Thank you for your contributions.