chamikaramj commented on code in PR #40183:
URL: https://github.com/apache/beam/pull/40183#discussion_r4051211850
##########
sdks/python/apache_beam/examples/snippets/snippets_test.py:
##########
@@ -1467,6 +1467,43 @@ def test_side_input_slow_update(self):
for i in range(-1, 10, 1):
os.unlink(src_file_pattern + str(first_ts + interval * i))
+ def test_side_input_slow_update_global_window(self):
+ side_input_interval = 5
+ main_input_interval = 5
+ duration = 30
+
+ first_ts = math.floor(time.time()) - duration
+ # Aligning the timestamp to get persistent results.
+ first_ts = first_ts - (
+ first_ts % (side_input_interval * main_input_interval))
+ last_ts = first_ts + duration
+
+ expected_main_input_elements = sorted(
+ first_ts + main_input_interval * i
+ for i in range(duration // main_input_interval))
+ expected_side_input_values = set(
+ str(first_ts + side_input_interval * i)
+ for i in range(duration // side_input_interval))
+
+ def check_enriched_elements(actual):
+ main_input_elements = sorted(element for element, _ in actual)
+ assert main_input_elements == expected_main_input_elements, (
+ 'Expected main input elements %s, got %s' %
+ (expected_main_input_elements, main_input_elements))
+
+ side_input_values = set(value for _, value in actual)
+ assert side_input_values, 'No side input value was observed.'
Review Comment:
This assertion might be passing trivially for `(main input, side input)`
combinations `(<valid value>, None)`. Try changing to:
```
side_input_values = set(value for _, value in actual)
assert None not in side_input_values, (
'Some elements were processed before the side input was ready.')
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
