

def test_copy_listeners(self):
    handler1 = (lambda event_type, details: None)
    handler2 = (lambda event_type, details: None)
    a_task = MyTask()
    a_task.notifier.register(task.EVENT_UPDATE_PROGRESS, handler1)
    b_task = a_task.copy()
    self.assertEqual(1, len(b_task.notifier))
    self.assertTrue(a_task.notifier.deregister_event(task.EVENT_UPDATE_PROGRESS))
    self.assertEqual(0, len(a_task.notifier))
    self.assertEqual(1, len(b_task.notifier))
    self.notifier.register(task.EVENT_UPDATE_PROGRESS, handler2)
    listeners = dict(list(b_task.notifier.listeners_iter()))
    self.assertEqual(2, len(listeners[task.EVENT_UPDATE_PROGRESS]))
    self.assertEqual(0, len(a_task.notifier))

dataset/ETHPy150Open openstack/taskflow/taskflow/tests/unit/test_task.py TaskTest.test_copy_listeners/VarMisuse@117/157 `b_task`->`self`