def test_on_dependency_only_fires_on_changes(self):
        """
             A
            / \
           B   C
            \ / \
             D   E
        """
        a_ready_calls = list()
        a_lost_calls = list()

        def reset_calls():
            a_ready_calls[:] = list()
            a_lost_calls[:] = list()

        def assertCalls(ready=list(), lost=list(), msg=''):
            self.assertEquals(sorted(a_ready_calls), sorted(ready), msg=msg)
            self.assertEquals(sorted(a_lost_calls), sorted(lost), msg=msg)

        def add_ready_call(dependency):
            a_ready_calls.append(dependency)

        def add_lost_call(dependency, reason):
            a_lost_calls.append(dependency)

        a_dependency = self.dependency_manager.as_dependency('A')
        a_dependency.on_dependency_ready += add_ready_call
        a_dependency.on_dependency_lost += add_lost_call

        b_dependency = self.dependency_manager.as_dependency('B')
        c_dependency = self.dependency_manager.as_dependency('C')
        d_dependency = self.dependency_manager.as_dependency('D')
        e_dependency = self.dependency_manager.as_dependency('E')

        self.dependency_manager.add_dependency(a_dependency, b_dependency)
        self.dependency_manager.add_dependency(a_dependency, c_dependency)
        self.dependency_manager.add_dependency(b_dependency, d_dependency)
        self.dependency_manager.add_dependency(c_dependency, d_dependency)
        self.dependency_manager.add_dependency(c_dependency, e_dependency)

        # assert that no calls have been made
        assertCalls()

        # resolving the initial states should result in one lost call, then one ready call for each dependency
        self.dependency_manager.resolve_initial_states()
        # The dependencies will be made available to A:
        assertCalls([b_dependency, c_dependency])
        reset_calls()

        # a change in a direct dependency should result in a single lost call
        for direct_dependency in b_dependency, c_dependency:
            direct_dependency.fire_on_lost('lost dependency')
            assertCalls(lost=[direct_dependency])
            reset_calls()

            direct_dependency.fire_on_ready()
            assertCalls(ready=[direct_dependency])
            reset_calls()

        # bringing down D should result in both B and C going down
        d_dependency.fire_on_lost('lost dependency')
        assertCalls(lost=[b_dependency, c_dependency])
        reset_calls()

        # if E goes up or down, nothing should change for A
        e_dependency.fire_on_lost('lost dependency')
        e_dependency.fire_on_ready()
        assertCalls()

        # when d comes up again, both b and c should become ready
        d_dependency.fire_on_ready()
        assertCalls(ready=[b_dependency, c_dependency])
        reset_calls()

        # bring down the dependencies in any order, and bring them up again - that should result in only
        # one on_lost and only one on_ready per dependency
        for dependencies in itertools.permutations((b_dependency, c_dependency, d_dependency, e_dependency)):
            for dependency in dependencies:
                # since lost-events cascade, and we're triggering the on_lost manually, we have to check
                # if the dependency already is lost, as otherwise it could cause one dependency to be on_lost twice
                # (once by the cascade, then later again by this call)
                if dependency.is_ready:
                    dependency.fire_on_lost('lost dependency')
                # as soon as one dependency is lost, one of As dependencies should be lost
                # wrap this in an if to avoid asserting too many times
                if dependency == dependencies[0]:
                    self.assertTrue(1 >= len(a_lost_calls))

            for dependency in dependencies:
                # see the above reasoning about checking .ready
                if not dependency.is_ready:
                    dependency.fire_on_ready()

            assertCalls(lost=[b_dependency, c_dependency], ready=[b_dependency, c_dependency], msg=','.join([repr(dep) for dep in dependencies]))
            reset_calls()
dataset/ETHPy150Open foundit/Piped/piped/test/test_dependencies.py/DependencyManagerTest.test_on_dependency_only_fires_on_changes/swapped operands of `>=`