| #!/usr/bin/env python3 |
| # -*- encoding: utf-8 -*- |
| |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| |
| # pylint: disable=missing-docstring |
| # pylint: disable=protected-access |
| import unittest |
| |
| from heronpy.api.component.component_spec import HeronComponentSpec, GlobalStreamId |
| from heronpy.api.stream import Grouping, Stream |
| |
| class ComponentSpecTest(unittest.TestCase): |
| def test_sanitize_args(self): |
| # good args |
| spec = HeronComponentSpec( |
| name="string", python_class_path="string.path", |
| is_spout=True, |
| par=1 |
| ) |
| self.assertIsNotNone(spec) |
| name_none_spec = HeronComponentSpec(name=None, python_class_path="string.path", |
| is_spout=True, par=1) |
| self.assertIsNotNone(name_none_spec) |
| |
| # bad name |
| with self.assertRaises(AssertionError): |
| HeronComponentSpec(123, "classpath", True, 1) |
| with self.assertRaises(AssertionError): |
| HeronComponentSpec(False, "classpath", True, 1) |
| |
| # bad classpath |
| with self.assertRaises(AssertionError): |
| HeronComponentSpec("name", {}, True, 1) |
| with self.assertRaises(AssertionError): |
| HeronComponentSpec("name", None, True, 1) |
| |
| # bad is_spout |
| with self.assertRaises(AssertionError): |
| HeronComponentSpec("name", "classpath", 1, 1) |
| with self.assertRaises(AssertionError): |
| HeronComponentSpec("name", "classpath", None, 1) |
| |
| # bad par |
| with self.assertRaises(AssertionError): |
| HeronComponentSpec("name", "classpath", True, "1") |
| with self.assertRaises(AssertionError): |
| HeronComponentSpec("name", "classpath", True, 1.35) |
| with self.assertRaises(AssertionError): |
| HeronComponentSpec("name", "classpath", True, -21) |
| with self.assertRaises(AssertionError): |
| HeronComponentSpec("name", "classpath", True, None) |
| |
| def test_sanitize_config(self): |
| # empty dict |
| ret = HeronComponentSpec._sanitize_config({}) |
| self.assertEqual(ret, {}) |
| |
| # non-dict given |
| with self.assertRaises(TypeError): |
| HeronComponentSpec._sanitize_config("{key: value}") |
| with self.assertRaises(TypeError): |
| HeronComponentSpec._sanitize_config(True) |
| with self.assertRaises(TypeError): |
| HeronComponentSpec._sanitize_config(None) |
| |
| # non-string key |
| with self.assertRaises(TypeError): |
| HeronComponentSpec._sanitize_config({['k', 'e', 'y']: "value"}) |
| with self.assertRaises(TypeError): |
| HeronComponentSpec._sanitize_config({None: "value"}) |
| |
| # convert boolean value |
| ret = HeronComponentSpec._sanitize_config({"key": True}) |
| self.assertEqual(ret["key"], "true") |
| ret = HeronComponentSpec._sanitize_config({"key": False}) |
| self.assertEqual(ret["key"], "false") |
| |
| # convert int and float |
| ret = HeronComponentSpec._sanitize_config({"key": 10}) |
| self.assertEqual(ret["key"], "10") |
| ret = HeronComponentSpec._sanitize_config({"key": -2400000}) |
| self.assertEqual(ret["key"], "-2400000") |
| ret = HeronComponentSpec._sanitize_config({"key": 0.0000001}) |
| self.assertEqual(ret["key"], "1e-07") |
| ret = HeronComponentSpec._sanitize_config({"key": -15.33333}) |
| self.assertEqual(ret["key"], "-15.33333") |
| |
| # non-string value -> should expect the same object |
| ret = HeronComponentSpec._sanitize_config({"key": ['v', 'a', 'l', 'u', 'e']}) |
| self.assertEqual(ret["key"], ['v', 'a', 'l', 'u', 'e']) |
| ret = HeronComponentSpec._sanitize_config({"key": None}) |
| self.assertEqual(ret["key"], None) |
| |
| def test_sanitize_inputs(self): |
| # Note that _sanitize_inputs() should only be called after HeronComponentSpec's |
| # name attribute is set |
| |
| # invalid inputs given as argument (valid ones are either dict, list, tuple or None) |
| invalid_spec = HeronComponentSpec("name", "classpath", True, 1, inputs="string") |
| with self.assertRaises(TypeError): |
| invalid_spec._sanitize_inputs() |
| |
| invalid_spec = HeronComponentSpec("name", "classpath", True, 1, inputs=100) |
| with self.assertRaises(TypeError): |
| invalid_spec._sanitize_inputs() |
| |
| # dict <HeronComponentSpec -> Grouping> |
| from_spec = HeronComponentSpec("spout", "sp_clspath", True, 1) |
| to_spec = HeronComponentSpec("bolt", "bl_clspath", False, 1, |
| inputs={from_spec: Grouping.SHUFFLE}) |
| ret = to_spec._sanitize_inputs() |
| self.assertEqual(ret, {GlobalStreamId("spout", "default"): Grouping.SHUFFLE}) |
| |
| from_spec = HeronComponentSpec("spout", "sp_clspath", True, 1) |
| from_spec.outputs = [Stream(name='another_stream')] |
| to_spec = HeronComponentSpec("bolt", "bl_clspath", False, 1, |
| inputs={from_spec['another_stream']: Grouping.ALL}) |
| ret = to_spec._sanitize_inputs() |
| self.assertEqual(ret, {GlobalStreamId("spout", "another_stream"): Grouping.ALL}) |
| |
| # HeronComponentSpec's name attribute not set |
| from_spec = HeronComponentSpec(None, "sp_clspath", True, 1) |
| to_spec = HeronComponentSpec("bolt", "bl_clspath", False, 1, |
| inputs={from_spec: Grouping.ALL}) |
| with self.assertRaises(RuntimeError): |
| to_spec._sanitize_inputs() |
| |
| # dict <GlobalStreamId -> Grouping> |
| inputs_dict = {GlobalStreamId("some_spout", "some_stream"): Grouping.NONE, |
| GlobalStreamId("another_spout", "default"): Grouping.fields(['word', 'count'])} |
| spec = HeronComponentSpec("bolt", "classpath", False, 1, inputs=inputs_dict) |
| ret = spec._sanitize_inputs() |
| self.assertEqual(ret, inputs_dict) |
| |
| # list of HeronComponentSpec |
| from_spec1 = HeronComponentSpec("spout1", "sp1_cls", True, 1) |
| from_spec2 = HeronComponentSpec("spout2", "sp2_cls", True, 1) |
| to_spec = HeronComponentSpec("bolt", "bl_cls", False, 1, inputs=[from_spec1, from_spec2]) |
| ret = to_spec._sanitize_inputs() |
| self.assertEqual(ret, {GlobalStreamId("spout1", "default"): Grouping.SHUFFLE, |
| GlobalStreamId("spout2", "default"): Grouping.SHUFFLE}) |
| |
| # HeronComponentSpec's name attribute not set |
| from_spec = HeronComponentSpec(None, "sp_clspath", True, 1) |
| to_spec = HeronComponentSpec("bolt", "bl_clspath", False, 1, inputs=[from_spec]) |
| with self.assertRaises(RuntimeError): |
| to_spec._sanitize_inputs() |
| |
| # list of GlobalStreamId |
| inputs_list = [GlobalStreamId("spout1", "default"), GlobalStreamId("spout2", "some_stream")] |
| spec = HeronComponentSpec("bolt", "bl_cls", False, 1, inputs=inputs_list) |
| ret = spec._sanitize_inputs() |
| self.assertEqual(ret, dict(list(zip(inputs_list, [Grouping.SHUFFLE] * 2)))) |
| |
| # list of neither GlobalStreamId nor HeronComponentSpec |
| inputs_list = [None, 123, "string", [GlobalStreamId("sp", "default")]] |
| spec = HeronComponentSpec("bolt", "bl_cls", False, 1, inputs=inputs_list) |
| with self.assertRaises(ValueError): |
| spec._sanitize_inputs() |
| |
| # pylint: disable=redefined-variable-type |
| # pylint: disable=pointless-statement |
| def test_sanitize_outputs(self): |
| # outputs is None (no argument to outputs) |
| spec = HeronComponentSpec("spout", "class", True, 1) |
| ret = spec._sanitize_outputs() |
| self.assertIsNone(ret) |
| |
| # outputs neither list nor tuple |
| spec = HeronComponentSpec("spout", "class", True, 1) |
| spec.outputs = "string" |
| with self.assertRaises(TypeError): |
| spec._sanitize_outputs() |
| |
| # output list contains a non-string and non-Stream object |
| spec = HeronComponentSpec("spout", "class", True, 1) |
| spec.outputs = ["string", False, 123] |
| with self.assertRaises(TypeError): |
| spec._sanitize_outputs() |
| |
| # output list is all string |
| spec = HeronComponentSpec("spout", "class", True, 1) |
| spec.outputs = ["string", "hello", "heron"] |
| ret = spec._sanitize_outputs() |
| self.assertEqual(ret, {"default": ["string", "hello", "heron"]}) |
| |
| # output list has mixed stream |
| spec = HeronComponentSpec("spout", "class", True, 1) |
| spec.outputs = ["string", "hello", Stream(fields=["abc", "def"], name="another_stream"), |
| Stream(fields=["another", "default"], name="default")] |
| ret = spec._sanitize_outputs() |
| self.assertEqual(ret, {"default": ["string", "hello", "another", "default"], |
| "another_stream": ["abc", "def"]}) |
| |
| def test_get_out_streamids(self): |
| # outputs is none |
| spec = HeronComponentSpec("spout", "class", True, 1) |
| ret = spec.get_out_streamids() |
| self.assertEqual(ret, set()) |
| |
| # outputs neither list nor tuple |
| spec = HeronComponentSpec("spout", "class", True, 1) |
| spec.outputs = "string" |
| with self.assertRaises(TypeError): |
| spec.get_out_streamids() |
| |
| # outputs sane |
| spec = HeronComponentSpec("spout", "class", True, 1) |
| spec.outputs = ["string", "hello", Stream(fields=["abc", "def"], name="another_stream"), |
| Stream(fields=["another", "default"], name="default")] |
| ret = spec.get_out_streamids() |
| self.assertEqual(ret, {"default", "another_stream"}) |
| |
| def test_get_item(self): |
| # HeronComponentSpec name set |
| spec = HeronComponentSpec("spout", "class", True, 1) |
| spec.outputs = ["string", "hello", Stream(fields=["abc", "def"], name="another_stream"), |
| Stream(fields=["another", "default"], name="default")] |
| ret = spec['another_stream'] |
| self.assertEqual(ret, GlobalStreamId("spout", "another_stream")) |
| |
| # HeronComponentSpec name not set |
| spec = HeronComponentSpec(None, "class", True, 1) |
| spec.outputs = ["string", "hello", Stream(fields=["abc", "def"], name="another_stream"), |
| Stream(fields=["another", "default"], name="default")] |
| ret = spec['default'] |
| self.assertEqual(ret, GlobalStreamId(spec, "default")) |
| |
| # stream id not registered |
| spec = HeronComponentSpec(None, "class", True, 1) |
| spec.outputs = ["string", "hello", Stream(fields=["abc", "def"], name="another_stream"), |
| Stream(fields=["another", "default"], name="default")] |
| with self.assertRaises(ValueError): |
| spec['non_existent_stream'] |
| |
| class GlobalStreamIdTest(unittest.TestCase): |
| def test_constructor(self): |
| # component id not string nor HeronComponentSpec |
| with self.assertRaises(TypeError): |
| GlobalStreamId(componentId=123, streamId="default") |
| # stream id not string |
| with self.assertRaises(TypeError): |
| GlobalStreamId(componentId="component", streamId=12345) |
| |
| def test_component_id_property(self): |
| # component id is string |
| gsi = GlobalStreamId(componentId="component", streamId="stream") |
| self.assertEqual(gsi.component_id, "component") |
| |
| # component id is HeronComponentSpec with name |
| spec = HeronComponentSpec("spout", "class", True, 1) |
| gsi = GlobalStreamId(spec, "stream") |
| self.assertEqual(gsi.component_id, "spout") |
| |
| # component id is HeronComponentSpec without name |
| spec = HeronComponentSpec(None, "class", True, 1) |
| gsi = GlobalStreamId(spec, "stream") |
| # expecting "<No name available for HeronComponentSpec yet, uuid: %s>" |
| self.assertIn(spec.uuid, gsi.component_id) |
