Module ImportFrom alias ImportFrom alias Import alias ClassDef bases NameLoad body Assign NameStore NameLoad FunctionDef arguments args NameParam defaults body Assign NameStore ListLoad Str NameLoad Call AttributeLoad NameLoad attr FunctionDef arguments args NameParam NameParam defaults body AugAssignAdd SubscriptStore NameLoad Index Num NameLoad decorator_list FunctionDef arguments args NameParam defaults body Assign SubscriptStore NameLoad Index Num NameLoad Expr Call AttributeLoad SubscriptLoad NameLoad Index Num attr decorator_list Expr Call AttributeLoad AttributeLoad NameLoad attr attr ListLoad Str Str keyword NameLoad keyword NameLoad Expr Call AttributeLoad SubscriptLoad NameLoad Index Num attr Expr Call AttributeLoad NameLoad attr SubscriptLoad NameLoad Index Num Str decorator_list Expr Str decorator_list
N/A sparts.tasks.twisted_command CommandTask sparts.tests.base SingleTaskTestCase N/A threading PollerTests N/A SingleTaskTestCase N/A N/A TASK CommandTask test_line_buffered N/A N/A self N/A N/A N/A result N/A None N/A N/A threading Event on_stdout N/A N/A trans msg N/A N/A N/A N/A result N/A 0 msg N/A on_exit N/A N/A reason N/A N/A N/A N/A result N/A 1 reason N/A N/A N/A N/A result N/A 2 set N/A N/A N/A N/A N/A self task run N/A echo Hello World! on_stdout on_stdout on_exit on_exit N/A N/A N/A N/A result N/A 2 wait N/A N/A N/A self assertEqual N/A result N/A 0 Hello World! N/A N/A 

    def test_unbuffered(self):
        result = ["", None, threading.Event()]

        def on_stdout(self, msg):
            result[0] += msg

        def on_exit(self, reason):
            result[1] = reason
            result[2].set()

        self.task.run(['echo', 'Hello World!'], on_stdout=on_stdout,
                      on_exit=on_exit, line_buffered=False)

        self.assertTrue(result[2].wait(30.0))
        self.assertEqual(result[0], 'Hello World!')
     N/A