test_service.py 文件源码

def test_usr1_rotates_logs(self):
        """
        SIGUSR1 should cause logs to be reopened.
        """
        logging.getLogger().addHandler(logging.FileHandler(self.makeFile()))
        # Store the initial set of handlers
        original_streams = [handler.stream for handler in
                            logging.getLogger().handlers if
                            isinstance(handler, logging.FileHandler)]

        # Instantiating LandscapeService should register the handler
        TestService(self.config)
        # We'll call it directly
        handler = signal.getsignal(signal.SIGUSR1)
        self.assertTrue(handler)
        handler(None, None)

        def check(ign):
            new_streams = [handler.stream for handler in
                           logging.getLogger().handlers if
                           isinstance(handler, logging.FileHandler)]

            for stream in new_streams:
                self.assertTrue(stream not in original_streams)

        # We need to give some room for the callFromThread to run
        d = deferLater(reactor, 0, lambda: None)
        return d.addCallback(check)