| # 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. |
| ''' tracker_unittest.py ''' |
| |
| # pylint: disable=missing-docstring, attribute-defined-outside-init |
| import unittest2 as unittest |
| from mock import call, patch, Mock |
| |
| import heron.proto.execution_state_pb2 as protoEState |
| from heron.statemgrs.src.python import statemanagerfactory |
| from heron.tools.tracker.src.python.topology import Topology |
| from heron.tools.tracker.src.python.tracker import Tracker |
| from mock_proto import MockProto |
| |
| class TrackerTest(unittest.TestCase): |
| def setUp(self): |
| mock_config = Mock() |
| mock_config.validate.return_value = True |
| self.tracker = Tracker(mock_config) |
| |
| # pylint: disable=unused-argument |
| @patch.object(Tracker, 'getTopologiesForStateLocation') |
| @patch.object(Tracker, 'removeTopology') |
| @patch.object(Tracker, 'addNewTopology') |
| @patch.object(statemanagerfactory, 'get_all_state_managers') |
| def test_first_synch_topologies( |
| self, mock_get_all_state_managers, |
| mock_add_new_topology, mock_remove_topology, |
| mock_get_topologies_for_state_location): |
| mock_state_manager_1 = Mock() |
| mock_state_manager_1.name = 'mock_name1' |
| |
| mock_state_manager_2 = Mock() |
| mock_state_manager_2.name = 'mock_name2' |
| |
| watches = {} |
| mock_get_all_state_managers.return_value = [mock_state_manager_1, mock_state_manager_2] |
| |
| mock_get_topologies_for_state_location.return_value = [] |
| def side_effect1(on_topologies_watch): |
| watches["1"] = on_topologies_watch |
| on_topologies_watch(['top_name1', 'top_name2']) |
| mock_state_manager_1.get_topologies = side_effect1 |
| |
| def side_effect2(on_topologies_watch): |
| watches["2"] = on_topologies_watch |
| on_topologies_watch(['top_name3', 'top_name4']) |
| mock_state_manager_2.get_topologies = side_effect2 |
| |
| self.tracker.synch_topologies() |
| mock_get_topologies_for_state_location.assert_has_calls( |
| [call("mock_name2"), |
| call("mock_name1")], |
| any_order=True) |
| mock_add_new_topology.assert_has_calls([call(mock_state_manager_1, 'top_name1'), |
| call(mock_state_manager_1, 'top_name2'), |
| call(mock_state_manager_2, 'top_name3'), |
| call(mock_state_manager_2, 'top_name4')], |
| any_order=True) |
| |
| @patch.object(Tracker, 'getTopologiesForStateLocation') |
| @patch.object(Tracker, 'removeTopology') |
| @patch.object(Tracker, 'addNewTopology') |
| @patch.object(statemanagerfactory, 'get_all_state_managers') |
| def test_synch_topologies_leading_with_add_and_remove_topologies( |
| self, mock_get_all_state_managers, |
| mock_add_new_topology, mock_remove_topology, |
| mock_get_topologies_for_state_location): |
| mock_state_manager_1 = Mock() |
| mock_state_manager_1.name = 'mock_name1' |
| |
| mock_state_manager_2 = Mock() |
| mock_state_manager_2.name = 'mock_name2' |
| |
| watches = {} |
| mock_get_all_state_managers.return_value = [mock_state_manager_1, mock_state_manager_2] |
| mock_get_topologies_for_state_location.return_value = [] |
| |
| def side_effect1(on_topologies_watch): |
| watches["1"] = on_topologies_watch |
| on_topologies_watch(['top_name1', 'top_name2']) |
| mock_state_manager_1.get_topologies = side_effect1 |
| |
| def side_effect2(on_topologies_watch): |
| watches["2"] = on_topologies_watch |
| on_topologies_watch(['top_name3', 'top_name4']) |
| mock_state_manager_2.get_topologies = side_effect2 |
| |
| self.tracker.synch_topologies() |
| mock_get_topologies_for_state_location.assert_has_calls( |
| [call("mock_name2"), |
| call("mock_name1")], |
| any_order=True) |
| mock_add_new_topology.assert_has_calls([call(mock_state_manager_1, 'top_name1'), |
| call(mock_state_manager_1, 'top_name2'), |
| call(mock_state_manager_2, 'top_name3'), |
| call(mock_state_manager_2, 'top_name4')], |
| any_order=True) |
| self.assertEqual(4, mock_add_new_topology.call_count) |
| self.assertEqual(0, mock_remove_topology.call_count) |
| mock_get_topologies_for_state_location.reset_mock() |
| mock_add_new_topology.reset_mock() |
| mock_remove_topology.reset_mock() |
| |
| def get_topologies_for_state_location_side_effect(name): |
| if name == 'mock_name1': |
| return [Topology('top_name1', 'mock_name1'), |
| Topology('top_name2', 'mock_name1')] |
| if name == 'mock_name2': |
| return [Topology('top_name3', 'mock_name2'), |
| Topology('top_name4', 'mock_name2')] |
| return [] |
| |
| # pylint: disable=line-too-long |
| mock_get_topologies_for_state_location.side_effect = get_topologies_for_state_location_side_effect |
| |
| watches["1"](['top_name1', 'top_name3']) |
| watches["2"](['top_name5', 'top_name6']) |
| mock_add_new_topology.assert_has_calls([call(mock_state_manager_1, 'top_name3'), |
| call(mock_state_manager_2, 'top_name5'), |
| call(mock_state_manager_2, 'top_name6')], |
| any_order=True) |
| mock_remove_topology.assert_has_calls([call('top_name2', 'mock_name1'), |
| call('top_name3', 'mock_name2'), |
| call('top_name4', 'mock_name2')], |
| any_order=False) |
| self.assertEqual(3, mock_add_new_topology.call_count) |
| self.assertEqual(3, mock_remove_topology.call_count) |
| |
| def fill_tracker_topologies(self): |
| |
| def create_mock_execution_state(role): |
| estate = protoEState.ExecutionState() |
| estate.role = role |
| return estate |
| |
| self.topology1 = Topology('top_name1', 'mock_name1') |
| self.topology1.cluster = 'cluster1' |
| self.topology1.environ = 'env1' |
| self.topology1.execution_state = create_mock_execution_state('mark') |
| |
| self.topology2 = Topology('top_name2', 'mock_name1') |
| self.topology2.cluster = 'cluster1' |
| self.topology2.environ = 'env1' |
| self.topology2.execution_state = create_mock_execution_state('bob') |
| |
| self.topology3 = Topology('top_name3', 'mock_name1') |
| self.topology3.cluster = 'cluster1' |
| self.topology3.environ = 'env2' |
| self.topology3.execution_state = create_mock_execution_state('tom') |
| |
| self.topology4 = Topology('top_name4', 'mock_name2') |
| self.topology4.cluster = 'cluster2' |
| self.topology4.environ = 'env1' |
| |
| self.topology5 = Topology('top_name5', 'mock_name2') |
| self.topology5.cluster = 'cluster2' |
| self.topology5.environ = 'env2' |
| self.tracker.topologies = [ |
| self.topology1, |
| self.topology2, |
| self.topology3, |
| self.topology4, |
| self.topology5] |
| |
| # pylint: disable=line-too-long |
| def test_get_topology_by_cluster_environ_and_name(self): |
| self.fill_tracker_topologies() |
| self.assertEqual(self.topology1, self.tracker.getTopologyByClusterRoleEnvironAndName('cluster1', 'mark', 'env1', 'top_name1')) |
| self.assertEqual(self.topology1, self.tracker.getTopologyByClusterRoleEnvironAndName('cluster1', None, 'env1', 'top_name1')) |
| self.assertEqual(self.topology2, self.tracker.getTopologyByClusterRoleEnvironAndName('cluster1', 'bob', 'env1', 'top_name2')) |
| self.assertEqual(self.topology2, self.tracker.getTopologyByClusterRoleEnvironAndName('cluster1', None, 'env1', 'top_name2')) |
| self.assertEqual(self.topology3, self.tracker.getTopologyByClusterRoleEnvironAndName('cluster1', 'tom', 'env2', 'top_name3')) |
| self.assertEqual(self.topology3, self.tracker.getTopologyByClusterRoleEnvironAndName('cluster1', None, 'env2', 'top_name3')) |
| self.assertEqual(self.topology4, self.tracker.getTopologyByClusterRoleEnvironAndName('cluster2', None, 'env1', 'top_name4')) |
| self.assertEqual(self.topology5, self.tracker.getTopologyByClusterRoleEnvironAndName('cluster2', None, 'env2', 'top_name5')) |
| |
| def test_get_topolies_for_state_location(self): |
| self.fill_tracker_topologies() |
| self.assertItemsEqual( |
| [self.topology1, self.topology2, self.topology3], |
| self.tracker.getTopologiesForStateLocation('mock_name1')) |
| self.assertItemsEqual( |
| [self.topology4, self.topology5], |
| self.tracker.getTopologiesForStateLocation('mock_name2')) |
| |
| def test_add_new_topology(self): |
| self.assertItemsEqual([], self.tracker.topologies) |
| mock_state_manager_1 = Mock() |
| mock_state_manager_1.name = 'mock_name1' |
| |
| self.tracker.addNewTopology(mock_state_manager_1, 'top_name1') |
| self.assertItemsEqual( |
| ['top_name1'], |
| map(lambda t: t.name, self.tracker.topologies)) |
| |
| self.tracker.addNewTopology(mock_state_manager_1, 'top_name2') |
| self.assertItemsEqual( |
| ['top_name1', 'top_name2'], |
| map(lambda t: t.name, self.tracker.topologies)) |
| |
| self.assertEqual(2, mock_state_manager_1.get_pplan.call_count) |
| self.assertEqual(2, mock_state_manager_1.get_execution_state.call_count) |
| self.assertEqual(2, mock_state_manager_1.get_tmaster.call_count) |
| |
| def test_remove_topology(self): |
| self.fill_tracker_topologies() |
| self.tracker.removeTopology('top_name1', 'mock_name1') |
| self.assertItemsEqual([self.topology2, self.topology3, self.topology4, self.topology5], |
| self.tracker.topologies) |
| self.tracker.removeTopology('top_name2', 'mock_name1') |
| self.assertItemsEqual([self.topology3, self.topology4, self.topology5], |
| self.tracker.topologies) |
| # Removing one that is not there should not have any affect |
| self.tracker.removeTopology('top_name8', 'mock_name1') |
| self.assertItemsEqual([self.topology3, self.topology4, self.topology5], |
| self.tracker.topologies) |
| self.tracker.removeTopology('top_name4', 'mock_name2') |
| self.assertItemsEqual([self.topology3, self.topology5], |
| self.tracker.topologies) |
| |
| def test_extract_physical_plan(self): |
| # Create topology |
| pb_pplan = MockProto().create_mock_simple_physical_plan() |
| topology = Topology('topology_name', 'state_manager') |
| topology.set_physical_plan(pb_pplan) |
| # Extract physical plan |
| pplan = self.tracker.extract_physical_plan(topology) |
| # Mock topology doesn't have topology config and instances |
| self.assertEqual(pplan['config'], {}) |
| self.assertEqual(pplan['bolts'], {'mock_bolt': []}) |
| self.assertEqual(pplan['spouts'], {'mock_spout': []}) |
| self.assertEqual(pplan['components']['mock_bolt']['config'], |
| {'topology.component.parallelism': '1'}) |
| self.assertEqual(pplan['components']['mock_spout']['config'], |
| {'topology.component.parallelism': '1'}) |
| self.assertEqual(pplan['instances'], {}) |
| self.assertEqual(pplan['stmgrs'], {}) |
