def setUp(self):
        """ Set up test fixtures. """
        self.mock_tracker = scaffold.MockTracker()

        pidlockfile_scenarios = make_pidlockfile_scenarios()
        self.pidlockfile_scenario = pidlockfile_scenarios['simple']
        pidfile_path = self.pidlockfile_scenario['path']

        scaffold.mock(
            "pidlockfile.PIDLockFile.__init__",
            tracker=self.mock_tracker)
        scaffold.mock(
            "pidlockfile.PIDLockFile.acquire",
            tracker=self.mock_tracker)

        self.scenario = {
            'pidfile_path': self.pidlockfile_scenario['path'],
            'acquire_timeout': object(),
            }

        self.test_kwargs = dict(
            path=self.scenario['pidfile_path'],
            acquire_timeout=self.scenario['acquire_timeout'],
            )
        self.test_instance = pidlockfile.TimeoutPIDLockFile(**self.test_kwargs)

def setup_lockfile_method_mocks(testcase, scenario, class_name):
    """ Set up common mock methods for lockfile class. """

    def mock_read_pid():
        return scenario["pidfile_pid"]

    def mock_is_locked():
        return scenario["locking_pid"] is not None

    def mock_i_am_locking():
        return scenario["locking_pid"] == scenario["pid"]

    def mock_acquire(timeout=None):
        if scenario["locking_pid"] is not None:
            raise lockfile.AlreadyLocked()
        scenario["locking_pid"] = scenario["pid"]

    def mock_release():
        if scenario["locking_pid"] is None:
            raise lockfile.NotLocked()
        if scenario["locking_pid"] != scenario["pid"]:
            raise lockfile.NotMyLock()
        scenario["locking_pid"] = None

    def mock_break_lock():
        scenario["locking_pid"] = None

    for func_name in ["read_pid", "is_locked", "i_am_locking", "acquire", "release", "break_lock"]:
        mock_func = vars()["mock_%(func_name)s" % vars()]
        lockfile_func_name = "%(class_name)s.%(func_name)s" % vars()
        mock_lockfile_func = scaffold.Mock(lockfile_func_name, returns_func=mock_func, tracker=testcase.mock_tracker)
        try:
            scaffold.mock(lockfile_func_name, mock_obj=mock_lockfile_func, tracker=testcase.mock_tracker)
        except NameError:
            pass

    def setUp(self):
        """ Set up test fixtures. """
        setup_pidfile_fixtures(self)
        self.scenario = self.scenarios['exist-currentpid']

        scaffold.mock(
            "os.remove",
            tracker=self.mock_tracker)

    def setUp(self):
        """ Set up test fixtures. """
        setup_runner_fixtures(self)
        set_runner_scenario(self, 'simple')

        scaffold.mock(
            "daemon.runner.DaemonRunner._usage_exit",
            raises=NotImplementedError,
            tracker=self.mock_tracker)

def setup_pidlockfile_fixtures(testcase, scenario_name=None):
    """ Set up common fixtures for PIDLockFile test cases. """

    setup_pidfile_fixtures(testcase)

    scaffold.mock("pidlockfile.write_pid_to_pidfile", tracker=testcase.mock_tracker)
    scaffold.mock("pidlockfile.remove_existing_pidfile", tracker=testcase.mock_tracker)

    if scenario_name is not None:
        set_pidlockfile_scenario(testcase, scenario_name, clear_tracker=False)

 def test_should_parse_system_argv_by_default(self):
     """ Should parse sys.argv by default. """
     instance = self.test_instance
     expect_action = 'start'
     argv = self.valid_argv_params['start']
     scaffold.mock(
         "sys.argv",
         mock_obj=argv,
         tracker=self.mock_tracker)
     instance.parse_args()
     self.failUnlessEqual(expect_action, instance.action)

 def test_openid_store_created_with_datadir(self):
     """ OpenIDStore should be created with specified datadir """
     params = self.valid_servers['datadir']
     datadir = params['datadir']
     scaffold.mock("server.OpenIDStore",
         outfile=self.mock_outfile)
     expect_mock_output = """\
         Called server.OpenIDStore(%(datadir)r)
         """ % vars()
     instance = self.server_class(**params['args'])
     self.failUnlessOutputCheckerMatch(
         expect_mock_output, self.mock_outfile.getvalue()
         )

 def test_parent_exits(self):
     """ become_daemon parent process should exit """
     parent_pid = 23
     scaffold.mock("os.fork", returns_iter=[parent_pid],
         outfile=self.mock_outfile)
     expect_mock_output = """\
         Called os.fork()
         Called os._exit(0)
         ...
         """
     server.become_daemon()
     self.failUnlessOutputCheckerMatch(
         expect_mock_output, self.mock_outfile.getvalue()
         )

 def test_requests_stop_then_start(self):
     """ Should request stop, then start. """
     instance = self.test_instance
     scaffold.mock(
         "daemon.runner.DaemonRunner._start",
         tracker=self.mock_tracker)
     scaffold.mock(
         "daemon.runner.DaemonRunner._stop",
         tracker=self.mock_tracker)
     expect_mock_output = """\
         Called daemon.runner.DaemonRunner._stop()
         Called daemon.runner.DaemonRunner._start()
         """
     instance.do_action()
     self.failUnlessMockCheckerMatch(expect_mock_output)

    def setUp(self):
        """ Set up test fixtures """
        self.mock_outfile = StringIO()

        test_pids = [0, 0]
        scaffold.mock("os.fork", returns_iter=test_pids,
            outfile=self.mock_outfile)
        scaffold.mock("os.setsid",
            outfile=self.mock_outfile)
        scaffold.mock("os._exit",
            outfile=self.mock_outfile)
        scaffold.mock("sys.exit",
            outfile=self.mock_outfile)
        scaffold.mock("server.remove_standard_files",
            outfile=self.mock_outfile)

    def setUp(self):
        """ Set up test fixtures. """
        self.mock_tracker = scaffold.MockTracker()

        pidlockfile_scenarios = make_pidlockfile_scenarios()
        self.pidlockfile_scenario = pidlockfile_scenarios["simple"]
        pidfile_path = self.pidlockfile_scenario["path"]

        scaffold.mock("pidlockfile.PIDLockFile.__init__", tracker=self.mock_tracker)
        scaffold.mock("pidlockfile.PIDLockFile.acquire", tracker=self.mock_tracker)

        self.scenario = {"pidfile_path": self.pidlockfile_scenario["path"], "acquire_timeout": object()}

        self.test_kwargs = dict(path=self.scenario["pidfile_path"], acquire_timeout=self.scenario["acquire_timeout"])
        self.test_instance = pidlockfile.TimeoutPIDLockFile(**self.test_kwargs)

 def test_child_forks_next_parent_exits(self):
     """ become_daemon should fork, then exit if parent """
     test_pids = [0, 42]
     scaffold.mock("os.fork", returns_iter=test_pids,
         outfile=self.mock_outfile)
     expect_mock_output = """\
         Called os.fork()
         Called os.setsid()
         Called os.fork()
         Called os._exit(0)
         ...
         """
     server.become_daemon()
     self.failUnlessOutputCheckerMatch(
         expect_mock_output, self.mock_outfile.getvalue()
         )

 def test_server_creates_http_server(self):
     """ GracieServer should create an HTTP server """
     params = self.valid_servers['simple']
     args = params['args']
     opts = params['opts']
     server_address = (opts.host, opts.port)
     scaffold.mock("server.HTTPServer",
         outfile=self.mock_outfile)
     expect_mock_output = """\
         Called server.HTTPServer(
             %(server_address)r,
             <class '...HTTPRequestHandler'>,
             <gracie.server.GracieServer object ...>)
         """ % vars()
     instance = self.server_class(**params['args'])
     self.failUnlessOutputCheckerMatch(
         expect_mock_output, self.mock_outfile.getvalue()
         )

def setup_pidlockfile_fixtures(testcase):
    """ Set up common fixtures for PIDLockFile test cases. """

    setup_pidfile_fixtures(testcase)

    testcase.pidlockfile_args = dict(
        path=testcase.scenario['path'],
        )

    testcase.test_instance = pidlockfile.PIDLockFile(
        **testcase.pidlockfile_args)

    scaffold.mock(
        "pidlockfile.write_pid_to_pidfile",
        tracker=testcase.mock_tracker)
    scaffold.mock(
        "pidlockfile.remove_existing_pidfile",
        tracker=testcase.mock_tracker)

def setup_pidfile_fixtures(testcase):
    """ Set up common fixtures for PID file test cases. """
    testcase.mock_tracker = scaffold.MockTracker()

    mock_current_pid = 235
    mock_other_pid = 8642
    mock_pidfile_current = FakeFileDescriptorStringIO(
        "%(mock_current_pid)d\n" % vars())
    mock_pidfile_other = FakeFileDescriptorStringIO(
        "%(mock_other_pid)d\n" % vars())
    mock_pidfile_path = tempfile.mktemp()

    scaffold.mock(
        "os.getpid",
        returns=mock_current_pid,
        tracker=testcase.mock_tracker)

    def mock_pidfile_open_nonexist(filename, mode, buffering):
        if 'r' in mode:
            raise IOError("No such file %(filename)r" % vars())
        else:
            result = testcase.scenario['pidfile']
        return result

    def mock_pidfile_open_exist(filename, mode, buffering):
        result = testcase.scenario['pidfile']
        return result

    def mock_open(filename, mode='r', buffering=None):
        if filename == testcase.scenario['path']:
            result = testcase.scenario['open_func'](
                filename, mode, buffering)
        else:
            result = FakeFileDescriptorStringIO()
        return result

    scaffold.mock(
        "__builtin__.open",
        returns_func=mock_open,
        tracker=testcase.mock_tracker)

    def mock_pidfile_os_open_nonexist(filename, flags, mode):
        if (flags & os.O_CREAT):
            result = testcase.scenario['pidfile'].fileno()
        else:
            raise OSError(errno.ENOENT, "No such file", filename)
        return result

    def mock_pidfile_os_open_exist(filename, flags, mode):
        result = testcase.scenario['pidfile'].fileno()
        return result

    def mock_os_open(filename, flags, mode=None):
        if filename == testcase.scenario['path']:
            result = testcase.scenario['os_open_func'](
                filename, flags, mode)
        else:
            result = FakeFileDescriptorStringIO().fileno()
        return result

    scaffold.mock(
        "os.open",
        returns_func=mock_os_open,
        tracker=testcase.mock_tracker)

    def mock_os_fdopen(fd, mode='r', buffering=None):
        if fd == testcase.scenario['pidfile'].fileno():
            result = testcase.scenario['pidfile']
        else:
            raise OSError(errno.EBADF, "Bad file descriptor")
        return result

    scaffold.mock(
        "os.fdopen",
        returns_func=mock_os_fdopen,
        tracker=testcase.mock_tracker)

    def mock_os_path_exists(path):
        if path == testcase.scenario['path']:
            result = testcase.scenario['path_exists_func']()
        else:
            result = False
        return result

    scaffold.mock(
        "os.path.exists",
        mock_obj=mock_os_path_exists)

    testcase.scenarios = {
        'nonexist': {
            'path': mock_pidfile_path,
            'pidfile': None,
            'pidfile_pid': None,
            'path_exists_func': (lambda: False),
            'os_open_func': mock_pidfile_os_open_nonexist,
            'open_func': mock_pidfile_open_nonexist,
            },
        'exist-currentpid': {
            'path': mock_pidfile_path,
            'pidfile': mock_pidfile_current,
#.........这里部分代码省略.........

def setup_pidfile_fixtures(testcase):
    """ Set up common fixtures for PID file test cases. """
    testcase.mock_tracker = scaffold.MockTracker()

    scenarios = make_pidlockfile_scenarios()
    testcase.pidlockfile_scenarios = scenarios

    def get_scenario_option(testcase, key, default=None):
        value = default
        try:
            value = testcase.scenario[key]
        except (NameError, TypeError, AttributeError, KeyError):
            pass
        return value

    scaffold.mock(
        "os.getpid",
        returns=scenarios['simple']['pid'],
        tracker=testcase.mock_tracker)

    def make_mock_open_funcs(testcase):

        def mock_open_nonexist(filename, mode, buffering):
            if 'r' in mode:
                raise IOError(
                    errno.ENOENT, "No such file %(filename)r" % vars())
            else:
                result = testcase.scenario['pidfile']
            return result

        def mock_open_read_denied(filename, mode, buffering):
            if 'r' in mode:
                raise IOError(
                    errno.EPERM, "Read denied on %(filename)r" % vars())
            else:
                result = testcase.scenario['pidfile']
            return result

        def mock_open_okay(filename, mode, buffering):
            result = testcase.scenario['pidfile']
            return result

        def mock_os_open_nonexist(filename, flags, mode):
            if (flags & os.O_CREAT):
                result = testcase.scenario['pidfile'].fileno()
            else:
                raise OSError(
                    errno.ENOENT, "No such file %(filename)r" % vars())
            return result

        def mock_os_open_read_denied(filename, flags, mode):
            if (flags & os.O_CREAT):
                result = testcase.scenario['pidfile'].fileno()
            else:
                raise OSError(
                    errno.EPERM, "Read denied on %(filename)r" % vars())
            return result

        def mock_os_open_okay(filename, flags, mode):
            result = testcase.scenario['pidfile'].fileno()
            return result

        funcs = dict(
            (name, obj) for (name, obj) in vars().items()
            if hasattr(obj, '__call__'))

        return funcs

    testcase.mock_pidfile_open_funcs = make_mock_open_funcs(testcase)

    def mock_open(filename, mode='r', buffering=None):
        scenario_path = get_scenario_option(testcase, 'path')
        if filename == scenario_path:
            func_name = testcase.scenario['open_func_name']
            mock_open_func = testcase.mock_pidfile_open_funcs[func_name]
            result = mock_open_func(filename, mode, buffering)
        else:
            result = FakeFileDescriptorStringIO()
        return result

    scaffold.mock(
        "__builtin__.open",
        returns_func=mock_open,
        tracker=testcase.mock_tracker)

    def mock_os_open(filename, flags, mode=None):
        scenario_path = get_scenario_option(testcase, 'path')
        if filename == scenario_path:
            func_name = testcase.scenario['os_open_func_name']
            mock_os_open_func = testcase.mock_pidfile_open_funcs[func_name]
            result = mock_os_open_func(filename, flags, mode)
        else:
            result = FakeFileDescriptorStringIO().fileno()
        return result

    scaffold.mock(
        "os.open",
        returns_func=mock_os_open,
        tracker=testcase.mock_tracker)

#.........这里部分代码省略.........

def setup_runner_fixtures(testcase):
    """ Set up common test fixtures for DaemonRunner test case. """
    testcase.mock_tracker = scaffold.MockTracker()

    setup_pidfile_fixtures(testcase)
    setup_streams_fixtures(testcase)

    testcase.runner_scenarios = make_runner_scenarios()

    testcase.mock_stderr = FakeFileDescriptorStringIO()
    scaffold.mock(
        "sys.stderr",
        mock_obj=testcase.mock_stderr,
        tracker=testcase.mock_tracker)

    simple_scenario = testcase.runner_scenarios['simple']

    testcase.lockfile_class_name = "pidlockfile.TimeoutPIDLockFile"

    testcase.mock_runner_lock = scaffold.Mock(
        testcase.lockfile_class_name,
        tracker=testcase.mock_tracker)
    testcase.mock_runner_lock.path = simple_scenario['pidfile_path']

    scaffold.mock(
        testcase.lockfile_class_name,
        returns=testcase.mock_runner_lock,
        tracker=testcase.mock_tracker)

    class TestApp(object):

        def __init__(self):
            self.stdin_path = testcase.stream_file_paths['stdin']
            self.stdout_path = testcase.stream_file_paths['stdout']
            self.stderr_path = testcase.stream_file_paths['stderr']
            self.pidfile_path = simple_scenario['pidfile_path']
            self.pidfile_timeout = simple_scenario['pidfile_timeout']

        run = scaffold.Mock(
            "TestApp.run",
            tracker=testcase.mock_tracker)

    testcase.TestApp = TestApp

    scaffold.mock(
        "daemon.runner.DaemonContext",
        returns=scaffold.Mock(
            "DaemonContext",
            tracker=testcase.mock_tracker),
        tracker=testcase.mock_tracker)

    testcase.test_app = testcase.TestApp()

    testcase.test_program_name = "bazprog"
    testcase.test_program_path = (
        "/foo/bar/%(test_program_name)s" % vars(testcase))
    testcase.valid_argv_params = {
        'start': [testcase.test_program_path, 'start'],
        'stop': [testcase.test_program_path, 'stop'],
        'restart': [testcase.test_program_path, 'restart'],
        }

    def mock_open(filename, mode=None, buffering=None):
        if filename in testcase.stream_files_by_path:
            result = testcase.stream_files_by_path[filename]
        else:
            result = FakeFileDescriptorStringIO()
        result.mode = mode
        result.buffering = buffering
        return result

    scaffold.mock(
        "builtins.open",
        returns_func=mock_open,
        tracker=testcase.mock_tracker)

    scaffold.mock(
        "os.kill",
        tracker=testcase.mock_tracker)

    scaffold.mock(
        "sys.argv",
        mock_obj=testcase.valid_argv_params['start'],
        tracker=testcase.mock_tracker)

    testcase.test_instance = runner.DaemonRunner(testcase.test_app)

    testcase.scenario = NotImplemented