[PATCH 1/2] Remove dispatch.py and its associated test case.

Chris Lumens clumens at redhat.com
Fri Oct 26 19:37:44 UTC 2012


Good riddance.
---
 pyanaconda/dispatch.py                 | 375 ---------------------------------
 tests/pyanaconda_test/dispatch_test.py | 265 -----------------------
 2 files changed, 640 deletions(-)
 delete mode 100644 pyanaconda/dispatch.py
 delete mode 100644 tests/pyanaconda_test/dispatch_test.py

diff --git a/pyanaconda/dispatch.py b/pyanaconda/dispatch.py
deleted file mode 100644
index d8bb6d0..0000000
--- a/pyanaconda/dispatch.py
+++ /dev/null
@@ -1,375 +0,0 @@
-#
-# dispatch.py: install/upgrade master flow control
-#
-# Copyright (C) 2001, 2002, 2003, 2004, 2005, 2006  Red Hat, Inc.
-# All rights reserved.
-#
-# This program is free software; you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation; either version 2 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 General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program.  If not, see <http://www.gnu.org/licenses/>.
-#
-# Author(s): Erik Troan <ewt at redhat.com>
-#
-
-import string
-from types import *
-
-import indexed_dict
-from constants import *
-from packages import doPostAction
-from packages import setupTimezone
-from storage import storageInitialize
-from storage import storageComplete
-from storage.partitioning import doAutoPartition
-from bootloader import writeBootLoader
-from flags import flags
-from installmethod import doMethodComplete
-from rescue import doRescue
-
-from backend import doPostSelection, doBackendSetup, doBasePackageSelect
-from backend import doPreInstall, doPostInstall, doInstall
-from backend import writeConfiguration
-
-from packages import doReIPL
-
-import logging
-log = logging.getLogger("anaconda")
-
-class DispatchError(Exception):
-    pass
-
-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
-        [True,  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.
-
-            Target is a callable that performs the step (direct step). It
-            accepts a sole argument: the anaconda object. If target is None,
-            user interface object is told to handle the step (i.e. indirect
-            step).
-        """
-        self._changes = {} # tracks changes (FROM, TO) in scheduling of other steps
-        self.name = name
-        self.target = target # None for dynamic target (e.g. gui view)
-        self._sched = self.SCHED_UNSCHEDULED
-        self.client_data = {} # used by the client code, not reset when
-                              # scheduling is reverted
-
-    def _reschedule(self, to_sched, current_step):
-        s_from = self.sched
-        new_sched = self.sched_state_machine[self.sched][to_sched]
-        if new_sched is False:
-            raise DispatchError(
-                "Can not reschedule step '%s' from '%s' to '%s'" %
-                (self.name,
-                 self.namesched(self._sched),
-                 self.namesched(to_sched)))
-        self._sched = to_sched
-        # only track scheduling if we are in a step and if something changes:
-        if current_step and s_from != self._sched:
-            current_step.record_history(self, s_from, self.sched)
-
-    @property
-    def changes(self):
-        return self._changes
-
-    def clear_changes(self):
-        self._changes = {}
-
-    @property
-    def direct(self):
-        return self.target is not None
-
-    def done(self, current_step):
-        return self._reschedule(self.SCHED_DONE, current_step)
-
-    def request(self, current_step):
-        return self._reschedule(self.SCHED_REQUESTED, current_step)
-
-    def unschedule(self, current_step):
-        return self._reschedule(self.SCHED_UNSCHEDULED, current_step)
-
-    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]
-
-    def record_history(self, step, s_from, s_to):
-        """ Stores information about scheduling changes into self.
-
-            step is a step where scheduling changed from s_from to s_to. the
-            self object in this method should be the currently executing object.
-        """
-        if step.name in self._changes:
-            s_from = self._changes[step.name][0]
-        self._changes[step.name] = (s_from, s_to)
-
-    def revert_sched(self, s_from, s_to):
-        assert(self.sched == s_to)
-        self._sched = s_from
-
-    @property
-    def sched(self):
-        return self._sched
-
-    def schedule(self, current_step):
-        return self._reschedule(self.SCHED_SCHEDULED, current_step)
-
-    def skip(self, current_step):
-        return self._reschedule(self.SCHED_SKIPPED, current_step)
-
-class Dispatcher(object):
-
-    def __init__(self, anaconda):
-        self.anaconda = anaconda
-        self.anaconda.dir = DISPATCH_FORWARD
-        self.step = None # name of the current step
-        self.stop = False
-        # step dictionary mapping step names to step objects
-        self.steps = indexed_dict.IndexedDict()
-        self.init_steps()
-
-    def _advance_step(self):
-        if self.step is None:
-            # initialization
-            log.info("dispatch: resetting to the first step.")
-            self.step = self.steps[0].name
-        elif self._step_index() < len(self.steps) - 1:
-            i = self._step_index()
-            if self.dir == DISPATCH_BACK:
-                # revert whatever changed in the current step
-                self._revert_scheduling(self.step)
-            self.step = self.steps[i + self.dir].name
-            if self.dir == DISPATCH_BACK:
-                # revert whatever changed in the step we moved back to
-                self._revert_scheduling(self.step)
-        else:
-            # advancing from the last step
-            self.step = "_invalid_"
-            self.stop = True
-
-    def _current_step(self):
-        if self.step:
-            return self.steps[self.step]
-        return None
-
-    def _revert_scheduling(self, reverted_step):
-        """ Revert scheduling changes that happened during reverted_step. """
-        for (step, (s_from, s_to)) in self.steps[reverted_step].changes.items():
-            self.steps[step].revert_sched(s_from, s_to)
-        self.steps[reverted_step].clear_changes()
-
-    def _step_index(self):
-        return self.steps.index(self.step)
-
-    def add_step(self, name, target = None):
-        self.steps[name] = Step(name, target)
-
-    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.
-        if self.step == None:
-            return False
-        i = self._step_index() - 1
-        while i >= 0:
-            sname = self.steps[i].name
-            if not self.step_is_direct(sname) and self.step_enabled(sname):
-                return True
-            i -= 1
-        return False
-
-    #pylint: disable-msg=E0202
-    @property
-    def dir(self):
-        return self.anaconda.dir
-
-    # pylint: disable-msg=E0102,E0202,E1101
-    @dir.setter
-    def dir(self, dir):
-        if dir not in [DISPATCH_BACK, DISPATCH_FORWARD, DISPATCH_DEFAULT]:
-            raise RuntimeError("dispatch: wrong direction code")
-        if dir in [DISPATCH_BACK, DISPATCH_FORWARD]:
-            self.anaconda.dir = dir
-
-    def done_steps(self, *steps):
-        changes = map(lambda s: self.steps[s].done(self._current_step()), steps)
-
-    def go_back(self):
-        """
-        The caller should make sure can_go_back() is True before calling this
-        method.
-        """
-        self.dir = DISPATCH_BACK
-        self.dispatch()
-
-    def go_forward(self):
-        self.dir = DISPATCH_FORWARD
-        self.dispatch()
-
-    def init_steps(self):
-        # Note that not only a subset of the steps is executed for a particular
-        # run, depending on the kind of installation, user selection, kickstart
-        # commands, used installclass and used user interface.
-        self.add_step("rescue", doRescue)
-        self.add_step("language")
-        self.add_step("keyboard")
-        self.add_step("filtertype")
-        self.add_step("filter")
-        self.add_step("storageinit", storageInitialize)
-        self.add_step("findinstall")
-        self.add_step("network")
-        self.add_step("timezone")
-        self.add_step("accounts")
-        self.add_step("setuptime", setupTimezone)
-        self.add_step("parttype")
-        self.add_step("cleardiskssel")
-        self.add_step("autopartitionexecute", doAutoPartition)
-        self.add_step("partition")
-        self.add_step("upgrademigratefs")
-        self.add_step("storagedone", storageComplete)
-        self.add_step("upgbootloader")
-        self.add_step("bootloader")
-        self.add_step("reposetup", doBackendSetup)
-        self.add_step("tasksel")
-        self.add_step("basepkgsel", doBasePackageSelect)
-        self.add_step("group-selection")
-        self.add_step("postselection", doPostSelection)
-        self.add_step("install")
-        self.add_step("preinstallconfig", doPreInstall)
-        self.add_step("installpackages", doInstall)
-        self.add_step("postinstallconfig", doPostInstall)
-        self.add_step("writeconfig", writeConfiguration)
-        self.add_step("instbootloader", writeBootLoader)
-        self.add_step("reipl", doReIPL)
-        self.add_step("methodcomplete", doMethodComplete)
-        self.add_step("dopostaction", doPostAction)
-        self.add_step("complete")
-
-    def request_steps(self, *steps):
-        changes = map(lambda s: self.steps[s].request(self._current_step()), steps)
-
-    def request_steps_gently(self, *steps):
-        """ Requests steps and won't raise an error if it is not possible for
-            some of them.
-        """
-        for step in steps:
-            try:
-                self.request_steps(step)
-            except DispatchError as e:
-                log.debug("dispatch: %s" % e)
-
-    def reset_scheduling(self):
-        log.info("dispatch: resetting scheduling")
-        for step in self.steps:
-            try:
-                self.steps[step].unschedule(self._current_step())
-            except DispatchError as e:
-                log.debug("dispatch: %s" % e)
-        log.info("dispatch: resetting finished.")
-
-    def run(self):
-        self.anaconda.intf.run(self.anaconda)
-        log.info("dispatch: finished.")
-
-    def schedule_steps(self, *steps):
-        changes = map(lambda s: self.steps[s].schedule(self._current_step()), steps)
-
-    def schedule_steps_gently(self, *steps):
-        """ Schedules steps and won't raise an error if it is not possible for
-            some of them.
-        """
-        for step in steps:
-            try:
-                self.schedule_steps(step)
-            except DispatchError as e:
-                log.debug("dispatch: %s" % e)
-
-    def step_data(self, step):
-        return self.steps[step].client_data
-
-    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 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 skip_steps(self, *steps):
-        changes = map(lambda s: self.steps[s].skip(self._current_step()), steps)
-
-    def step_is_direct(self, step):
-        return self.steps[step].direct
-
-    def dispatch(self):
-        if self.step:
-            log.info("dispatch: leaving (%d) step %s" %
-                     (self.dir, self.step))
-            self.done_steps(self.step)
-        self._advance_step()
-
-        while True:
-            if self.stop:
-                # installation has proceeded beyond the last step: finished
-                self.anaconda.intf.shutdown()
-                return
-            if self.step_disabled(self.step):
-                self._advance_step()
-                continue
-            log.info("dispatch: moving (%d) to step %s" %
-                     (self.dir, self.step))
-            if self.step_is_direct(self.step):
-                # handle a direct step by just calling the function
-                log.debug("dispatch: %s is a direct step" % self.step)
-                self.dir = self.steps[self.step].target(self.anaconda)
-            else:
-                # handle an indirect step (IOW the user interface has a screen
-                # to display to the user):
-                rc = self.anaconda.intf.display_step(self.step)
-                if rc == DISPATCH_WAITING:
-                    # a new screen has been set up and we are waiting for the
-                    # user input now (this only ever happens with the GTK UI and
-                    # is because we need to get back to gtk.main())
-                    return
-                elif rc == DISPATCH_DEFAULT:
-                    log.debug("dispatch: the interface chose "
-                              "not to display step %s." % self.step)
-                else:
-                    self.dir = rc
-            log.info("dispatch: leaving (%d) step %s" %
-                     (self.dir, self.step))
-            self.done_steps(self.step)
-            self._advance_step()
diff --git a/tests/pyanaconda_test/dispatch_test.py b/tests/pyanaconda_test/dispatch_test.py
deleted file mode 100644
index 9ee80eb..0000000
--- a/tests/pyanaconda_test/dispatch_test.py
+++ /dev/null
@@ -1,265 +0,0 @@
-import mock
-
-class StepTest(mock.TestCase):
-    def setUp(self):
-        self.setupModules(
-            ['_isys', 'logging', 'logging.config',
-             '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.dispatch import DispatchError
-        s = Step("a_step")
-        s.schedule(None)
-        s.request(None)
-        s.done(None)
-        self.assertEquals(s.sched, Step.SCHED_DONE)
-        self.assertRaises(DispatchError, s.skip, None)
-
-    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.dispatch import DispatchError
-        s = Step("a_step")
-        s._reschedule(Step.SCHED_UNSCHEDULED, None)
-        self.assertEqual(s.sched, Step.SCHED_UNSCHEDULED)
-        s._reschedule(Step.SCHED_SCHEDULED, None)
-        self.assertEqual(s.sched, Step.SCHED_SCHEDULED)
-        s._reschedule(Step.SCHED_REQUESTED, None)
-        self.assertEqual(s.sched, Step.SCHED_REQUESTED)
-        self.assertRaises(DispatchError, s._reschedule, Step.SCHED_SKIPPED, None)
-        self.assertRaises(DispatchError, s._reschedule, Step.SCHED_UNSCHEDULED, None)
-        s._reschedule(Step.SCHED_DONE, None)
-        self.assertEqual(s.sched, Step.SCHED_DONE, None)
-
-        s = Step("another_step")
-        s._reschedule(Step.SCHED_SKIPPED, None)
-        self.assertEqual(s.sched, Step.SCHED_SKIPPED)
-        self.assertRaises(DispatchError, s._reschedule, Step.SCHED_SCHEDULED, None)
-        self.assertRaises(DispatchError, s._reschedule, Step.SCHED_REQUESTED, None)
-        self.assertRaises(DispatchError, s._reschedule, Step.SCHED_DONE, None)
-
-    def request_test(self):
-        from pyanaconda.dispatch import Step
-        from pyanaconda.dispatch import DispatchError
-        s = Step("a_step")
-        s.request(None)
-        self.assertRaises(DispatchError, s.skip, None)
-
-    def schedule_test(self):
-        from pyanaconda.dispatch import Step
-        s = Step("a_step")
-        s.schedule(None)
-        self.assertEquals(s.sched, Step.SCHED_SCHEDULED)
-
-    def unschedule_test(self):
-        from pyanaconda.dispatch import Step
-        from pyanaconda.dispatch import DispatchError
-        s = Step("a_step")
-        s.schedule(None)
-        self.assertEquals(s.sched, Step.SCHED_SCHEDULED)
-        s.unschedule(None)
-        self.assertEquals(s.sched, Step.SCHED_UNSCHEDULED)
-        s.request(None)
-        self.assertEquals(s.sched, Step.SCHED_REQUESTED)
-        self.assertRaises(DispatchError, s.unschedule, None)
-
-    def skip_test(self):
-        from pyanaconda.dispatch import Step
-        from pyanaconda.dispatch import DispatchError
-        s = Step("a_step")
-        s.skip(None)
-        self.assertEquals(s.sched, Step.SCHED_SKIPPED)
-        self.assertRaises(DispatchError, s.done, None)
-        self.assertRaises(DispatchError, s.request, None)
-
-    def record_history_test(self):
-        from pyanaconda.dispatch import Step
-        s = Step("first_step")
-        self.assertDictEqual(s.changes, {})
-
-        s2 = Step("second_step")
-        s.record_history(s2, Step.SCHED_UNSCHEDULED, Step.SCHED_SCHEDULED)
-        self.assertDictEqual(
-            s.changes,
-            {'second_step' : (Step.SCHED_UNSCHEDULED, Step.SCHED_SCHEDULED)})
-        s.record_history(s2, Step.SCHED_SCHEDULED, Step.SCHED_DONE)
-        self.assertDictEqual(
-            s.changes,
-            {'second_step' : (Step.SCHED_UNSCHEDULED, Step.SCHED_DONE)})
-
-    def revert_sched_test(self):
-        from pyanaconda.dispatch import Step
-        s = Step("first_step")
-        s.request(None)
-        s.revert_sched(Step.SCHED_UNSCHEDULED, Step.SCHED_REQUESTED)
-        self.assertEqual(s.sched, Step.SCHED_UNSCHEDULED)
-        s.request(None)
-        self.assertRaises(AssertionError, s.revert_sched, Step.SCHED_SKIPPED,
-                          Step.SCHED_UNSCHEDULED)
-
-    def tracking_test(self):
-        """ Tests that reschedule correctly registers steps to revert """
-        from pyanaconda.dispatch import Step
-        s1 = Step("first_step")
-        s2 = Step("second_step")
-        s = Step("current_step")
-        s1.schedule(s)
-        s2.schedule(s)
-        s2.request(s)
-        self.assertDictEqual(
-            s.changes,
-            {"first_step" : (Step.SCHED_UNSCHEDULED, Step.SCHED_SCHEDULED),
-             "second_step" : (Step.SCHED_UNSCHEDULED, Step.SCHED_REQUESTED)})
-
-class DispatchTest(mock.TestCase):
-    def setUp(self):
-        self.setupModules(
-            ['_isys', 'logging', 'logging.config',
-             '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 can_go_back_test(self):
-        d = self._getDispatcher()
-        d.schedule_steps("rescue", "kickstart", "language", "keyboard")
-        # can not go back if dispatcher hasn't moved out of the initial state
-        self.assertFalse(d.can_go_back())
-        # can not go back if all preceding steps are direct
-        d.step = "language"
-        self.assertFalse(d.can_go_back())
-        d.done_steps("language")
-        # but in "keyboard" we can go back to "language"
-        d.step = "keyboard"
-        self.assertTrue(d.can_go_back())
-
-    def done_test(self):
-        from pyanaconda.dispatch import Dispatcher
-        from pyanaconda.dispatch import Step
-        from pyanaconda.dispatch 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.skip_steps, "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.skip_steps("betanag", "complete")
-        self.assertFalse(d.step_enabled("betanag"))
-        self.assertFalse(d.step_enabled("complete"))
-        self.assertTrue(d.step_enabled("filtertype"))
-
-    def request_steps_gently_test(self):
-        from pyanaconda.dispatch import DispatchError
-        from pyanaconda.dispatch import Step
-        d = self._getDispatcher()
-        d.schedule_steps("betanag", "complete")
-        d.skip_steps("betanag")
-        self.assertRaises(DispatchError, d.request_steps, "betanag")
-        d.request_steps_gently("betanag")
-        d.request_steps_gently("complete")
-        self.assertEqual(d.steps["betanag"].sched, Step.SCHED_SKIPPED)
-        self.assertEqual(d.steps["complete"].sched, Step.SCHED_REQUESTED)
-
-    def track_scheduling_test(self):
-        from pyanaconda.dispatch import Step
-        d = self._getDispatcher()
-        d.schedule_steps("betanag", "filtertype", "filter")
-        d.step = "betanag"
-        # tested through the request_steps
-        d.request_steps("filtertype", "filter")
-        self.assertDictEqual(
-            d.steps[d.step].changes,
-            {"filtertype" : (Step.SCHED_SCHEDULED, Step.SCHED_REQUESTED),
-             "filter" : (Step.SCHED_SCHEDULED, Step.SCHED_REQUESTED)})
-
-    def revert_scheduling_test(self):
-        from pyanaconda.dispatch import Step
-        d = self._getDispatcher()
-        d.schedule_steps("betanag", "filtertype", "filter")
-        d.step = "betanag"
-        d.request_steps("filtertype", "filter")
-        d._revert_scheduling("betanag")
-        self.assertEqual(d.steps["filtertype"].sched, Step.SCHED_SCHEDULED)
-        self.assertEqual(d.steps["filter"].sched, Step.SCHED_SCHEDULED)
-        self.assertDictEqual(d.steps[d.step].changes, {})
-
-    def reset_scheduling_test(self):
-        from pyanaconda.dispatch import Step
-        d = self._getDispatcher()
-        # initial setup
-        d.schedule_steps("betanag", "filtertype")
-        d.request_steps("filter")
-        # in step betanag scheduling gets reset:
-        d.step = "betanag"
-        d.reset_scheduling()
-        # what is requested can not be unrequested:
-        self.assertEqual(d.steps["betanag"].sched, Step.SCHED_UNSCHEDULED)
-        self.assertEqual(d.steps["filtertype"].sched, Step.SCHED_UNSCHEDULED)
-        self.assertEqual(d.steps["filter"].sched, Step.SCHED_REQUESTED)
-        # make sure the tracking works fine
-        self.assertEqual(
-            d.steps["betanag"].changes,
-            {"betanag" : (Step.SCHED_SCHEDULED, Step.SCHED_UNSCHEDULED),
-             "filtertype" : (Step.SCHED_SCHEDULED, Step.SCHED_UNSCHEDULED)})
-
-    def step_data_test(self):
-        from pyanaconda.dispatch import Step
-        d = self._getDispatcher()
-        self.assertEqual(d.step_data("filter"), {})
-        d.step_data("filter")["key"] = 42
-        # we can retrieve it
-        self.assertEqual(d.step_data("filter")["key"], 42)
-        # it persists over scheduling changes
-        d.schedule_steps("filter")
-        d._revert_scheduling("filter")
-        self.assertEqual(d.step_data("filter")["key"], 42)
-- 
1.7.11.2



More information about the anaconda-patches mailing list