test_releaseupgrader.py 文件源码

def test_upgrade_with_failure(self):
        """
        The L{ReleaseUpgrader.upgrade} sends a message with failed status
        field if the upgrade-tool exits with non-zero code.
        """
        self.upgrader.logs_directory = self.makeDir()
        upgrade_tool_directory = self.config.upgrade_tool_directory
        upgrade_tool_filename = os.path.join(upgrade_tool_directory, "karmic")
        fd = open(upgrade_tool_filename, "w")
        fd.write("#!/bin/sh\n"
                 "echo out\n"
                 "echo err >&2\n"
                 "exit 3")
        fd.close()
        os.chmod(upgrade_tool_filename, 0o755)

        deferred = Deferred()

        def do_test():

            result = self.upgrader.upgrade("karmic", 100)

            def check_result(ignored):
                result_text = (u"=== Standard output ===\n\nout\n\n\n"
                               "=== Standard error ===\n\nerr\n\n\n")
                self.assertMessages(self.get_pending_messages(),
                                    [{"type": "operation-result",
                                      "operation-id": 100,
                                      "status": FAILED,
                                      "result-text": result_text,
                                      "result-code": 3}])

            result.addCallback(check_result)
            result.chainDeferred(deferred)

        reactor.callWhenRunning(do_test)
        return deferred