-
-
Notifications
You must be signed in to change notification settings - Fork 77
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add more local pipeline operations #24
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -46,9 +46,8 @@ def setUpClass(cls): | |
cls.ops = LocalPipelineOperations() | ||
|
||
def test_local_map(self): | ||
some_map = self.ops.map([1, 2, 3], lambda x: x) | ||
# some_map is its own consumable iterator | ||
self.assertIs(some_map, iter(some_map)) | ||
self.assertEqual(list(self.ops.map([], lambda x: x / 0)), | ||
[]) | ||
|
||
self.assertEqual(list(self.ops.map([1, 2, 3], str)), | ||
["1", "2", "3"]) | ||
|
@@ -64,6 +63,17 @@ def test_local_map_tuple(self): | |
self.assertEqual(list(self.ops.map_tuple(tuple_list, lambda k, v: ( | ||
str(k), str(v)))), [("1", "2"), ("2", "3"), ("3", "4")]) | ||
|
||
def test_local_map_values(self): | ||
self.assertEqual(list(self.ops.map_values([], lambda x: x / 0)), | ||
[]) | ||
|
||
tuple_list = [(1, 2), (2, 3), (3, 4)] | ||
|
||
self.assertEqual(list(self.ops.map_values(tuple_list, str)), | ||
[(1, "2"), (2, "3"), (3, "4")]) | ||
self.assertEqual(list(self.ops.map_values(tuple_list, lambda x: x**2)), | ||
[(1, 4), (2, 9), (3, 16)]) | ||
|
||
def test_local_group_by_key(self): | ||
some_dict = [("cheese", "brie"), ("bread", "sourdough"), | ||
("cheese", "swiss")] | ||
|
@@ -72,6 +82,57 @@ def test_local_group_by_key(self): | |
("cheese", ["brie", "swiss"]), | ||
("bread", ["sourdough"])]) | ||
|
||
def test_local_filter(self): | ||
self.assertEqual(list(self.ops.filter([], lambda x: True)), | ||
[]) | ||
self.assertEqual(list(self.ops.filter([], lambda x: False)), | ||
[]) | ||
|
||
example_list = [1, 2, 2, 3, 3, 4, 2] | ||
|
||
self.assertEqual(list(self.ops.filter(example_list, lambda x: x % 2)), | ||
[1, 3, 3]) | ||
self.assertEqual(list(self.ops.filter(example_list, lambda x: x < 3)), | ||
[1, 2, 2, 2]) | ||
|
||
def test_local_values(self): | ||
self.assertEqual(list(self.ops.values([])), | ||
[]) | ||
|
||
example_list = [(1, 2), (2, 3), (3, 4), (4, 8)] | ||
|
||
self.assertEqual(list(self.ops.values(example_list)), | ||
[2, 3, 4, 8]) | ||
|
||
def test_local_count_per_element(self): | ||
example_list = [1, 2, 3, 4, 5, 6, 1, 4, 0, 1] | ||
result = self.ops.count_per_element(example_list) | ||
|
||
self.assertEqual(dict(result), | ||
{1: 3, 2: 1, 3: 1, 4: 2, 5: 1, 6: 1, 0: 1}) | ||
|
||
def test_laziness(self): | ||
def exceptions_generator_function(): | ||
yield 1 / 0 | ||
|
||
def assert_laziness(operator, *args): | ||
try: | ||
operator(exceptions_generator_function(), *args) | ||
except ZeroDivisionError: | ||
self.fail(f"local {operator.__name__} is not lazy") | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This function exists for tests to fail properly with e.g. when filter implementation is broken: return list(filter(fn, col)) the fail looks like this:
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Thanks for explanation. I like that approach! |
||
|
||
# reading from exceptions_generator_function() results in error: | ||
self.assertRaises(ZeroDivisionError, | ||
next, exceptions_generator_function()) | ||
|
||
# lazy operators accept exceptions_generator_function() | ||
# as argument without raising errors: | ||
assert_laziness(self.ops.map, str) | ||
assert_laziness(self.ops.map_values, str) | ||
assert_laziness(self.ops.filter, bool) | ||
assert_laziness(self.ops.values) | ||
assert_laziness(self.ops.count_per_element) | ||
|
||
|
||
if __name__ == '__main__': | ||
unittest.main() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I like idea with Counter!
One question, in this line, is 'col' accessed the first time only when iterations over count_per_element() started?
What do you think about adding the 'lazyness' test:
Let 'exception_generator' is a generator which raise an exception on the first access to it
and then
def test_lazyness():
self.ops.map(exception_generator) # should be no exception
self.ops.map_tuple(exception_generator)
....
on implementing a new method it would be easy to add to the test and to ensure that the method is lazy
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Cool idea for testing!
I added a test like this and yes,
count_per_element
works as expected.