Python twisted.python.log 模块，textFromEventDict() 实例源码

def log_dictionary(self, event_dict, level):
        """
        Format the given log event as text and write it to the output file.

        @param event_dict: a log event
        @type event_dict: L{dict} mapping L{str} (native string) to L{object}

        @param level: The event level to log.
        """
        text = textFromEventDict(event_dict)
        if text is None:
            return

        time_str = self.formatTime(event_dict["time"])
        fmt_dict = {
            "system": event_dict["system"],
            "text": text.replace("\n", "\n\t")
        }
        msg_str = _safeFormat("[%(system)s] %(text)s\n", fmt_dict)
        untilConcludes(logging.getLogger('parlay').log, level, time_str + " " + msg_str)

def test_malformedHeaderCGI(self):
        """
        Check for the error message in the duplicated header
        """
        cgiFilename = self.writeCGI(BROKEN_HEADER_CGI)

        portnum = self.startServer(cgiFilename)
        url = "http://localhost:%d/cgi" % (portnum,)
        agent = client.Agent(reactor)
        d = agent.request(b"GET", url)
        d.addCallback(discardBody)
        loggedMessages = []

        def addMessage(eventDict):
            loggedMessages.append(log.textFromEventDict(eventDict))

        log.addObserver(addMessage)
        self.addCleanup(log.removeObserver, addMessage)

        def checkResponse(ignored):
            self.assertIn("ignoring malformed CGI header: 'XYZ'",
                          loggedMessages)

        d.addCallback(checkResponse)
        return d

def test_logStderr(self):
        """
        When the _errFlag is set to L{StandardErrorBehavior.LOG},
        L{endpoints._WrapIProtocol} logs stderr (in childDataReceived).
        """
        d = self.ep.connect(self.factory)
        self.successResultOf(d)
        wpp = self.reactor.processProtocol
        log.addObserver(self._stdLog)
        self.addCleanup(log.removeObserver, self._stdLog)

        wpp.childDataReceived(2, b'stderr1')
        self.assertEqual(self.eventLog['executable'], wpp.executable)
        self.assertEqual(self.eventLog['data'], b'stderr1')
        self.assertEqual(self.eventLog['protocol'], wpp.protocol)
        self.assertIn(
            'wrote stderr unhandled by',
            log.textFromEventDict(self.eventLog))

def test_logErrorLogsErrorNoRepr(self):
        """
        The text logged by L{defer.logError} has no repr of the failure.
        """
        output = []

        def emit(eventDict):
            output.append(log.textFromEventDict(eventDict))

        log.addObserver(emit)

        error = failure.Failure(RuntimeError())
        defer.logError(error)
        self.flushLoggedErrors(RuntimeError)

        self.assertTrue(output[0].startswith("Unhandled Error\nTraceback "))

def test_errorLogNoRepr(self):
        """
        Verify that when a L{Deferred} with no references to it is fired,
        the logged message does not contain a repr of the failure object.
        """
        defer.Deferred().addCallback(lambda x: 1 // 0).callback(1)

        gc.collect()
        self._check()

        self.assertEqual(2, len(self.c))
        msg = log.textFromEventDict(self.c[-1])
        expected = "Unhandled Error\nTraceback "
        self.assertTrue(msg.startswith(expected),
                        "Expected message starting with: {0!r}".
                            format(expected))

def test_errorLogDebugInfo(self):
        """
        Verify that when a L{Deferred} with no references to it is fired,
        the logged message includes debug info if debugging on the deferred
        is enabled.
        """
        def doit():
            d = defer.Deferred()
            d.debug = True
            d.addCallback(lambda x: 1 // 0)
            d.callback(1)

        doit()
        gc.collect()
        self._check()

        self.assertEqual(2, len(self.c))
        msg = log.textFromEventDict(self.c[-1])
        expected = "(debug:  I"
        self.assertTrue(msg.startswith(expected),
                        "Expected message starting with: {0!r}".
                            format(expected))

def emit(self, eventDict):
        """
        Send a message event to the I{syslog}.

        @param eventDict: The event to send.  If it has no C{'message'} key, it
            will be ignored.  Otherwise, if it has C{'syslogPriority'} and/or
            C{'syslogFacility'} keys, these will be used as the syslog priority
            and facility.  If it has no C{'syslogPriority'} key but a true
            value for the C{'isError'} key, the B{LOG_ALERT} priority will be
            used; if it has a false value for C{'isError'}, B{LOG_INFO} will be
            used.  If the C{'message'} key is multiline, each line will be sent
            to the syslog separately.
        """
        # Figure out what the message-text is.
        text = log.textFromEventDict(eventDict)
        if text is None:
            return

        # Figure out what syslog parameters we might need to use.
        priority = syslog.LOG_INFO
        facility = 0
        if eventDict['isError']:
            priority = syslog.LOG_ALERT
        if 'syslogPriority' in eventDict:
            priority = int(eventDict['syslogPriority'])
        if 'syslogFacility' in eventDict:
            facility = int(eventDict['syslogFacility'])

        # Break the message up into lines and send them.
        lines = text.split('\n')
        while lines[-1:] == ['']:
            lines.pop()

        firstLine = True
        for line in lines:
            if firstLine:
                firstLine = False
            else:
                line = '\t' + line
            self.syslog(priority | facility,
                        '[%s] %s' % (eventDict['system'], line))

def test_connectionLostLogMsg(self):
        """
        When a connection is lost, an informative message should be logged
        (see L{getExpectedConnectionLostLogMsg}): an address identifying
        the port and the fact that it was closed.
        """

        loggedMessages = []
        def logConnectionLostMsg(eventDict):
            loggedMessages.append(log.textFromEventDict(eventDict))

        reactor = self.buildReactor()
        p = self.getListeningPort(reactor, ServerFactory())
        expectedMessage = self.getExpectedConnectionLostLogMsg(p)
        log.addObserver(logConnectionLostMsg)

        def stopReactor(ignored):
            log.removeObserver(logConnectionLostMsg)
            reactor.stop()

        def doStopListening():
            log.addObserver(logConnectionLostMsg)
            maybeDeferred(p.stopListening).addCallback(stopReactor)

        reactor.callWhenRunning(doStopListening)
        reactor.run()

        self.assertIn(expectedMessage, loggedMessages)

def test_message(self):
        """
        The C{"message"} value, when specified, is concatenated to generate the
        message.
        """
        eventDict = dict(message=("a", "b", "c"))
        text = log.textFromEventDict(eventDict)
        self.assertEqual(text, "a b c")

def test_noMessageNoFormat(self):
        """
        If C{"format"} is unspecified and C{"message"} is empty, return
        L{None}.
        """
        eventDict = dict(message=(), isError=0)
        text = log.textFromEventDict(eventDict)
        self.assertIsNone(text)

def test_whySpecified(self):
        """
        The C{"why"} value, when specified, is first part of message.
        """
        try:
            raise RuntimeError()
        except:
            eventDict = dict(
                message=(), isError=1, failure=failure.Failure(), why="foo"
            )
            text = log.textFromEventDict(eventDict)
            self.assertTrue(text.startswith("foo\n"))

def test_whyDefault(self):
        """
        The C{"why"} value, when unspecified, defaults to C{"Unhandled Error"}.
        """
        try:
            raise RuntimeError()
        except:
            eventDict = dict(message=(), isError=1, failure=failure.Failure())
            text = log.textFromEventDict(eventDict)
            self.assertTrue(text.startswith("Unhandled Error\n"))

def test_noTracebackForYou(self):
        """
        If unable to obtain a traceback due to an exception, catch it and note
        the error.
        """
        # Invalid failure object doesn't implement .getTraceback()
        eventDict = dict(message=(), isError=1, failure=object())
        text = log.textFromEventDict(eventDict)
        self.assertIn("\n(unable to obtain traceback)", text)

def emit(self, eventDict):
        """
        Produce a log output.
        """
        from twisted.trial._dist import managercommands
        text = textFromEventDict(eventDict)
        if text is None:
            return
        self.protocol.callRemote(managercommands.TestWrite, out=text)

def test_format(self):
        """
        Formatting is translated such that text is rendered correctly, even
        though old-style logging doesn't use PEP 3101 formatting.
        """
        event = self.forwardAndVerify(
            dict(log_format="Hello, {who}!", who="world")
        )
        self.assertEqual(
            legacyLog.textFromEventDict(event),
            "Hello, world!"
        )

def test_formatAlreadySet(self):
        """
        Formatting is not altered if the old-style C{"format"} key already
        exists.
        """
        event = self.forwardAndVerify(
            dict(log_format="Hello!", format="Howdy!")
        )
        self.assertEqual(legacyLog.textFromEventDict(event), "Howdy!")

def test_observed(self):
        """
        The observer is called exactly once.
        """
        publishToNewObserver(
            self.observer, self.legacyEvent(), legacyLog.textFromEventDict
        )
        self.assertEqual(len(self.events), 1)

def test_time(self):
        """
        The old-style C{"time"} key is copied to the new-style C{"log_time"}
        key.
        """
        publishToNewObserver(
            self.observer, self.legacyEvent(), legacyLog.textFromEventDict
        )
        self.assertEqual(
            self.events[0]["log_time"], self.events[0]["time"]
        )

def test_message(self):
        """
        A published old-style event should format as text in the same way as
        the given C{textFromEventDict} callable would format it.
        """
        def textFromEventDict(event):
            return "".join(reversed(" ".join(event["message"])))

        event = self.legacyEvent("Hello,", "world!")
        text = textFromEventDict(event)

        publishToNewObserver(self.observer, event, textFromEventDict)
        self.assertEqual(formatEvent(self.events[0]), text)

def test_defaultLogLevel(self):
        """
        Published event should have log level of L{LogLevel.info}.
        """
        publishToNewObserver(
            self.observer, self.legacyEvent(), legacyLog.textFromEventDict
        )
        self.assertEqual(self.events[0]["log_level"], LogLevel.info)

def test_stdlibLogLevel(self):
        """
        If the old-style C{"logLevel"} key is set to a standard library logging
        level, using a predefined (L{int}) constant, the new-style
        C{"log_level"} key should get set to the corresponding log level.
        """
        publishToNewObserver(
            self.observer,
            self.legacyEvent(logLevel=py_logging.WARNING),
            legacyLog.textFromEventDict
        )
        self.assertEqual(self.events[0]["log_level"], LogLevel.warn)

def test_stdlibLogLevelWithString(self):
        """
        If the old-style C{"logLevel"} key is set to a standard library logging
        level, using a string value, the new-style C{"log_level"} key should
        get set to the corresponding log level.
        """
        publishToNewObserver(
            self.observer,
            self.legacyEvent(logLevel="WARNING"),
            legacyLog.textFromEventDict
        )
        self.assertEqual(self.events[0]["log_level"], LogLevel.warn)

def test_stdlibLogLevelWithGarbage(self):
        """
        If the old-style C{"logLevel"} key is set to a standard library logging
        level, using an unknown value, the new-style C{"log_level"} key should
        not get set.
        """
        publishToNewObserver(
            self.observer,
            self.legacyEvent(logLevel="Foo!!!!!"),
            legacyLog.textFromEventDict
        )
        self.assertNotIn("log_level", self.events[0])

def test_defaultNamespace(self):
        """
        Published event should have a namespace of C{"log_legacy"} to indicate
        that it was forwarded from legacy logging.
        """
        publishToNewObserver(
            self.observer, self.legacyEvent(), legacyLog.textFromEventDict
        )
        self.assertEqual(self.events[0]["log_namespace"], "log_legacy")

def test_system(self):
        """
        The old-style C{"system"} key is copied to the new-style
        C{"log_system"} key.
        """
        publishToNewObserver(
            self.observer, self.legacyEvent(), legacyLog.textFromEventDict
        )
        self.assertEqual(
            self.events[0]["log_system"], self.events[0]["system"]
        )

def emit(self, eventDict):
        if not self.should_emit(eventDict):
            return

        text = tw_log.textFromEventDict(eventDict)
        if eventDict['isError']:
            self.stderr.write(text + "\n")
            self.stderr.flush()
        else:
            self.write(text + "\n")
            self.flush()

def dump(self):
        """Return a string of events formatted with `textFromEventDict`.

        This returns a single string, where each log message is separated from
        the next by a line containing "---". Formatting is done by Twisted's
        *legacy* log machinery, which may or may not differ from the modern
        machinery.
        """
        return "\n---\n".join(
            log.textFromEventDict(event) for event in self.events)

    # For compatibility with fixtures.FakeLogger.

def formatEvent(event, includeType=False):
    event['isError'] = 'failure' in event
    msg = textFromEventDict(event)

    if includeType:
        type_tag = '[%s] ' % event.get('type', 'console')
    else:
        type_tag = ''

    timestamp = time.strftime("%d/%m/%Y %H:%M:%S")
    return "%s\t%s\t%s" % (timestamp, type_tag, msg)

def test_deployedMode(self):
        """
        The C{dropin.cache} file may not be writable: the cache should still be
        attainable, but an error should be logged to show that the cache
        couldn't be updated.
        """
        # Generate the cache
        plugin.getCache(self.module)

        cachepath = self.package.child('dropin.cache')

        # Add a new plugin
        FilePath(__file__).sibling('plugin_extra1.py'
            ).copyTo(self.package.child('pluginextra.py'))
        invalidateImportCaches()

        os.chmod(self.package.path, 0o500)
        # Change the right of dropin.cache too for windows
        os.chmod(cachepath.path, 0o400)
        self.addCleanup(os.chmod, self.package.path, 0o700)
        self.addCleanup(os.chmod, cachepath.path, 0o700)

        # Start observing log events to see the warning
        events = []
        addObserver(events.append)
        self.addCleanup(removeObserver, events.append)

        cache = plugin.getCache(self.module)
        # The new plugin should be reported
        self.assertIn('pluginextra', cache)
        self.assertIn(self.originalPlugin, cache)

        # Make sure something was logged about the cache.
        expected = "Unable to write to plugin cache %s: error number %d" % (
            cachepath.path, errno.EPERM)
        for event in events:
            if expected in textFromEventDict(event):
                break
        else:
            self.fail(
                "Did not observe unwriteable cache warning in log "
                "events: %r" % (events,))



# This is something like the Twisted plugins file.