diff options
Diffstat (limited to 'ick2/buildgraph_tests.py')
| -rw-r--r-- | ick2/buildgraph_tests.py | 206 |
1 files changed, 206 insertions, 0 deletions
diff --git a/ick2/buildgraph_tests.py b/ick2/buildgraph_tests.py new file mode 100644 index 0000000..a81d2c6 --- /dev/null +++ b/ick2/buildgraph_tests.py @@ -0,0 +1,206 @@ +# Copyright (C) 2018 Lars Wirzenius +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + + +import unittest + + +import ick2 + + +class BuildGraphTests(unittest.TestCase): + + def test_has_no_actions_initially(self): + graph = ick2.BuildGraph() + self.assertEqual(graph.get_actions(), {}) + + def test_initialises_from_existing_graph(self): + as_dict = { + '1': { + 'action': {'action': 'foo'}, + 'status': 'ready', + 'depends': [], + }, + '2': { + 'action': {'action': 'bar'}, + 'status': 'blocked', + 'depends': ['1'], + }, + } + graph = ick2.BuildGraph(graph=as_dict) + self.assertEqual(graph.get_actions(), as_dict) + + def test_appends_first_action(self): + action = { + 'action': 'foo', + } + + graph = ick2.BuildGraph() + action_id = graph.append_action(action) + self.assertEqual(action_id, '1') + self.assertEqual( + graph.get_actions(), + { + action_id: { + 'action': {'action': 'foo'}, + 'status': 'ready', + 'depends': [] + } + } + ) + self.assertEqual(graph.get_action(action_id), action) + + def test_appends_second_action(self): + action1 = { + 'action': 'foo', + } + action2 = { + 'action': 'bar', + } + + graph = ick2.BuildGraph() + action_id1 = graph.append_action(action1) + action_id2 = graph.append_action(action2) + self.assertEqual(action_id1, '1') + self.assertEqual(action_id2, '2') + self.assertEqual( + graph.get_actions(), + { + action_id1: { + 'action': {'action': 'foo'}, + 'status': 'ready', + 'depends': [], + }, + action_id2: { + 'action': {'action': 'bar'}, + 'status': 'blocked', + 'depends': [action_id1], + }, + } + ) + + def test_changes_action_status(self): + action = { + 'action': 'foo', + } + + graph = ick2.BuildGraph() + action_id = graph.append_action(action) + self.assertEqual(graph.get_action_status(action_id), 'ready') + + graph.set_action_status(action_id, 'building') + self.assertEqual(graph.get_action_status(action_id), 'building') + + def test_appends_pipeline_actions(self): + pipeline_as_dict = { + 'actions': [ + { + 'action': 'foo', + }, + { + 'action': 'bar', + }, + ], + } + pipeline = ick2.resource_from_dict(pipeline_as_dict) + graph = ick2.BuildGraph() + graph.append_pipeline(pipeline) + self.assertEqual( + graph.get_actions(), + { + '1': { + 'action': {'action': 'foo'}, + 'depends': [], + 'status': 'ready', + }, + '2': { + 'action': {'action': 'bar'}, + 'depends': ['1'], + 'status': 'blocked', + }, + } + ) + + def test_appending_action_triggers_observer(self): + + def observer(): + setattr(self, 'observed', True) + + action = {} + + setattr(self, 'observed', False) + graph = ick2.BuildGraph() + graph.set_observer(observer) + self.assertFalse(getattr(self, 'observed')) + action_id = graph.append_action(action) + self.assertTrue(getattr(self, 'observed')) + + def test_finds_actions(self): + pipeline_as_dict = { + 'actions': [ + { + 'action': 'foo', + }, + { + 'action': 'bar', + }, + ], + } + pipeline = ick2.resource_from_dict(pipeline_as_dict) + graph = ick2.BuildGraph() + graph.append_pipeline(pipeline) + + self.assertEqual(graph.find_actions('no-such-state'), []) + self.assertEqual(graph.find_actions('ready'), ['1']) + self.assertEqual(graph.find_actions('blocked'), ['2']) + + def test_doesnt_unblock_when_deps_are_not_done(self): + pipeline_as_dict = { + 'actions': [ + { + 'action': 'foo', + }, + { + 'action': 'bar', + }, + ], + } + pipeline = ick2.resource_from_dict(pipeline_as_dict) + graph = ick2.BuildGraph() + graph.append_pipeline(pipeline) + + graph.unblock() + self.assertEqual(graph.get_action_status('1'), 'ready') + self.assertEqual(graph.get_action_status('2'), 'blocked') + + def test_unblocks_when_deps_are_done(self): + pipeline_as_dict = { + 'actions': [ + { + 'action': 'foo', + }, + { + 'action': 'bar', + }, + ], + } + pipeline = ick2.resource_from_dict(pipeline_as_dict) + graph = ick2.BuildGraph() + graph.append_pipeline(pipeline) + + graph.set_action_status('1', 'done') + graph.unblock() + self.assertEqual(graph.get_action_status('1'), 'done') + self.assertEqual(graph.get_action_status('2'), 'ready') |