diff options
| -rw-r--r-- | pyanaconda/dispatch.py | 108 | ||||
| -rw-r--r-- | pyanaconda/errors.py | 3 | ||||
| -rw-r--r-- | pyanaconda/installclass.py | 2 | ||||
| -rw-r--r-- | pyanaconda/upgrade.py | 2 | ||||
| -rw-r--r-- | tests/pyanaconda_test/dispatch_test.py | 133 |
5 files changed, 215 insertions, 33 deletions
diff --git a/pyanaconda/dispatch.py b/pyanaconda/dispatch.py index ee15ce2a1..ef42f145b 100644 --- a/pyanaconda/dispatch.py +++ b/pyanaconda/dispatch.py @@ -24,6 +24,7 @@ import string from types import * import indexed_dict +import errors from constants import * from packages import writeKSConfiguration, turnOnFilesystems from packages import doPostAction @@ -55,11 +56,24 @@ from packages import doReIPL import logging log = logging.getLogger("anaconda") -SCHED_SKIPPED = -2 -SCHED_SKIPPED_PERMANENTLY = -1 -SCHED_SCHEDULED = 1 - class Step(object): + SCHED_UNSCHEDULED = 0 + SCHED_SCHEDULED = 1 # will execute if not explicitly skipped + SCHED_SKIPPED = 2 # is never going to execute + SCHED_REQUESTED = 3 # will execute and can not be skipped + SCHED_DONE = 4 # done is a final state + + sched_state_machine = [ + # Table of allowed state changes, rows are the current state, columns + # are what we want to transition into. False when such transition is not + # allowed. + # unsch sched skip req done + [True , True , True , True , True ], # unscheduled + [False, True , True , True , True ], # scheduled + [False, False, True , False, False], # skipped + [False, False, False, True , True ], # requested + [False, False, False, False, True ]] # done + def __init__(self, name, target = None): """ Dispatcher step object. @@ -70,22 +84,48 @@ class Step(object): """ self.name = name self.target = target # None for dynamic target (e.g. gui view) - self._sched = SCHED_SKIPPED + self._sched = self.SCHED_UNSCHEDULED - def reschedule(self, new_sched): - if self._sched == SCHED_SKIPPED_PERMANENTLY: - # do not allow "upgrading" of permanently skipped steps - return - self._sched = new_sched + def _reschedule(self, to_sched): + new_sched = self.sched_state_machine[self._sched][to_sched] + if new_sched is False: + raise errors.DispatchError( + "Can not reschedule step '%s' from '%s' to '%s'" % + (self.name, + self.namesched(self._sched), + self.namesched(to_sched))) + self._sched = to_sched @property def direct(self): return self.target is not None + def done(self): + self._reschedule(self.SCHED_DONE) + + def request(self): + self._reschedule(self.SCHED_REQUESTED) + + def namesched(self, sched): + return { + self.SCHED_UNSCHEDULED : "unscheduled", + self.SCHED_SCHEDULED : "scheduled", + self.SCHED_SKIPPED : "skipped", + self.SCHED_REQUESTED : "requested", + self.SCHED_DONE : "done" + }[sched] + @property def sched(self): return self._sched + def schedule(self): + self._reschedule(self.SCHED_SCHEDULED) + + def skip(self): + self._reschedule(self.SCHED_SKIPPED) + + class Dispatcher(object): def __init__(self, anaconda): @@ -158,6 +198,9 @@ class Dispatcher(object): def _step_index(self): return self.steps.index(self.step) + def done_steps(self, *steps): + map(lambda s: self.steps[s].done(), steps) + def go_back(self): """ The caller should make sure canGoBack() is True before calling this @@ -170,42 +213,43 @@ class Dispatcher(object): self._setDir(DISPATCH_FORWARD) self.dispatch() - def canGoBack(self): + def can_go_back(self): # Begin with the step before this one. If all steps are skipped, # we can not go backwards from this one. i = self._step_index() - 1 while i >= 0: sname = self.steps[i].name - if not (self.stepIsDirect(sname) or self.stepInSkipList(sname)): + if not self.step_is_direct(sname) and self.step_enabled(sname): return True i -= 1 return False + def request_step(self, *steps): + map(lambda s: self.steps[s].request(), steps) + def run(self): self.anaconda.intf.run(self.anaconda) log.info("dispatch: finished.") - def setStepList(self, *new_steps): - for (name, step) in self.steps.items(): - if name in new_steps: - step.reschedule(SCHED_SCHEDULED) - else: - step.reschedule(SCHED_SKIPPED) + def schedule_steps(self, *steps): + map(lambda s: self.steps[s].schedule(), steps) - def stepInSkipList(self, step): - return self.steps[step].sched in [SCHED_SKIPPED_PERMANENTLY, - SCHED_SKIPPED] + def step_disabled(self, step): + """ True if step is not yet scheduled to be run or will never be run + (i.e. is skipped). + """ + return not self.step_enabled(step) - def skipStep(self, step, skip = 1, permanent = 0): - new_sched = SCHED_SCHEDULED - if skip: - if permanent: - new_sched = SCHED_SKIPPED_PERMANENTLY - else: - new_sched = SCHED_SKIPPED - self.steps[step].reschedule(new_sched) + def step_enabled(self, step): + """ True if step is scheduled to be run or have been run already. """ + return self.steps[step].sched in [Step.SCHED_SCHEDULED, + Step.SCHED_REQUESTED, + Step.SCHED_DONE] + + def skipStep(self, *steps): + map(lambda s: self.steps[s].skip(), steps) - def stepIsDirect(self, step): + def step_is_direct(self, step): return self.steps[step].direct def dispatch(self): @@ -216,6 +260,7 @@ class Dispatcher(object): else: log.info("dispatch: leaving (%d) step %s" % (self.dir, self.step)) + self.done_steps(self.step) self._advance_step() while True: @@ -223,7 +268,7 @@ class Dispatcher(object): # installation has proceeded beyond the last step: finished self.anaconda.intf.shutdown() return - if self.stepInSkipList(self.step): + if self.step_disabled(self.step): self._advance_step() continue log.info("dispatch: moving (%d) to step %s" % @@ -248,6 +293,7 @@ class Dispatcher(object): self.dir = rc log.info("dispatch: leaving (%d) step %s" % (self.dir, self.step)) + self.done_steps(self.step) self._advance_step() def _getDir(self): diff --git a/pyanaconda/errors.py b/pyanaconda/errors.py index 97337335f..0bb62376d 100644 --- a/pyanaconda/errors.py +++ b/pyanaconda/errors.py @@ -157,3 +157,6 @@ class NoSuchGroup(Exception): def __str__ (self): return self.value + +class DispatchError(RuntimeError): + pass diff --git a/pyanaconda/installclass.py b/pyanaconda/installclass.py index 8092e94cf..781505176 100644 --- a/pyanaconda/installclass.py +++ b/pyanaconda/installclass.py @@ -83,7 +83,7 @@ class BaseInstallClass(object): def setSteps(self, anaconda): dispatch = anaconda.dispatch - dispatch.setStepList( + dispatch.schedule_steps( "sshd", "language", "keyboard", diff --git a/pyanaconda/upgrade.py b/pyanaconda/upgrade.py index 6a1616985..2b96eb505 100644 --- a/pyanaconda/upgrade.py +++ b/pyanaconda/upgrade.py @@ -288,7 +288,7 @@ def upgradeMountFilesystems(anaconda): def setSteps(anaconda): dispatch = anaconda.dispatch - dispatch.setStepList( + dispatch.schedule_steps( "language", "keyboard", "filtertype", diff --git a/tests/pyanaconda_test/dispatch_test.py b/tests/pyanaconda_test/dispatch_test.py new file mode 100644 index 000000000..485dddeb1 --- /dev/null +++ b/tests/pyanaconda_test/dispatch_test.py @@ -0,0 +1,133 @@ +import mock + +class StepTest(mock.TestCase): + def setUp(self): + self.setupModules( + ['logging', 'pyanaconda.anaconda_log', 'block']) + + import pyanaconda + pyanaconda.anaconda_log = mock.Mock() + + def tearDown(self): + self.tearDownModules() + + def done_test(self): + from pyanaconda.dispatch import Step + from pyanaconda.errors import DispatchError + s = Step("a_step") + s.schedule() + s.request() + s.done() + self.assertEquals(s.sched, Step.SCHED_DONE) + self.assertRaises(DispatchError, s.skip) + + def instantiation_test(self): + from pyanaconda.dispatch import Step + s = Step("yeah") + self.assertIsInstance(s, Step) + # name + self.assertEqual(s.name, "yeah") + # default scheduling + self.assertEqual(s.sched, Step.SCHED_UNSCHEDULED) + + def namesched_test(self): + from pyanaconda.dispatch import Step + s = Step("a_step") + self.assertEqual(s.namesched(Step.SCHED_UNSCHEDULED), "unscheduled") + + def reschedule_test(self): + from pyanaconda.dispatch import Step + from pyanaconda.errors import DispatchError + s = Step("a_step") + s._reschedule(Step.SCHED_UNSCHEDULED) + self.assertEqual(s.sched, Step.SCHED_UNSCHEDULED) + s._reschedule(Step.SCHED_SCHEDULED) + self.assertEqual(s.sched, Step.SCHED_SCHEDULED) + s._reschedule(Step.SCHED_REQUESTED) + self.assertEqual(s.sched, Step.SCHED_REQUESTED) + self.assertRaises(DispatchError, s._reschedule, Step.SCHED_SKIPPED) + self.assertRaises(DispatchError, s._reschedule, Step.SCHED_UNSCHEDULED) + s._reschedule(Step.SCHED_DONE) + self.assertEqual(s.sched, Step.SCHED_DONE) + + s = Step("another_step") + s._reschedule(Step.SCHED_SKIPPED) + self.assertEqual(s.sched, Step.SCHED_SKIPPED) + self.assertRaises(DispatchError, s._reschedule, Step.SCHED_SCHEDULED) + self.assertRaises(DispatchError, s._reschedule, Step.SCHED_REQUESTED) + self.assertRaises(DispatchError, s._reschedule, Step.SCHED_DONE) + + def request_test(self): + from pyanaconda.dispatch import Step + from pyanaconda.errors import DispatchError + s = Step("a_step") + s.request() + self.assertRaises(DispatchError, s.skip) + + def schedule_test(self): + from pyanaconda.dispatch import Step + s = Step("a_step") + s.schedule() + self.assertEquals(s.sched, Step.SCHED_SCHEDULED) + + def skip_test(self): + from pyanaconda.dispatch import Step + from pyanaconda.errors import DispatchError + s = Step("a_step") + s.skip() + self.assertEquals(s.sched, Step.SCHED_SKIPPED) + self.assertRaises(DispatchError, s.done) + self.assertRaises(DispatchError, s.request) + +class DispatchTest(mock.TestCase): + def setUp(self): + self.setupModules( + ['logging', 'pyanaconda.anaconda_log', 'block']) + + import pyanaconda + pyanaconda.anaconda_log = mock.Mock() + + def tearDown(self): + self.tearDownModules() + + def _getDispatcher(self): + from pyanaconda.dispatch import Dispatcher + self.anaconda_obj = mock.Mock() + return Dispatcher(self.anaconda_obj) + + def done_test(self): + from pyanaconda.dispatch import Dispatcher + from pyanaconda.dispatch import Step + from pyanaconda.errors import DispatchError + + d = self._getDispatcher() + self.assertFalse(d.step_enabled("betanag")) + d.schedule_steps("betanag") + d.done_steps("betanag") + self.assertTrue(d.step_enabled("betanag")) + self.assertTrue(d.steps["betanag"], Step.SCHED_DONE) + self.assertRaises(DispatchError, d.skipStep, "betanag") + + def instantiation_test(self): + from pyanaconda.dispatch import Dispatcher + d = self._getDispatcher() + self.assertIsInstance(d, Dispatcher) + + def schedule_test(self): + from pyanaconda.dispatch import Step + d = self._getDispatcher() + d.schedule_steps("betanag", "complete") + self.assertEqual(d.steps["betanag"].sched, Step.SCHED_SCHEDULED) + self.assertEqual(d.steps["complete"].sched, Step.SCHED_SCHEDULED) + self.assertEqual(d.steps["bootloader"].sched, Step.SCHED_UNSCHEDULED) + # impossible to reach nonexistent steps: + self.assertRaises(KeyError, d.steps.__getitem__, "nonexistent") + + def skip_test(self): + d = self._getDispatcher() + d.schedule_steps("betanag", "filtertype", "complete") + self.assertTrue(d.step_enabled("betanag")) + d.skipStep("betanag", "complete") + self.assertFalse(d.step_enabled("betanag")) + self.assertFalse(d.step_enabled("complete")) + self.assertTrue(d.step_enabled("filtertype")) |