common_bits.py

@dataclass
class Status:
    # important bits pulls from juju FullStatus
    ...


def check(full_status: juju.FullStatus, **kwargs) -> Status:
    # simple, sync code here
    ...

fancy.py

# async filter style, fancy and ergonomic
async def loop(statuses: AsyncIterable[Status], **kwargs) -> AsyncIterable[bool]:
    local_state = ...
    async for status in statuses:
        now = time.time()

        # time-dependent code here
        if status.something in local_state.some_other:
            yield False

        for blip in local_state.foo:
            blip.set_foo(min(now, local_state.old_timestamp))
            if status.check_foo_against(blip):
                yield False

        yield True


# prod usage
async def new_wait_for_idle(self: Model, **kwargs):
    # check kwrags, like apps is not a str

    async def statuses():
        while True:
            yield check(await self.get_full_status(), **some_kw)

    async for done in loop(statuses(), **other_kw):
        if done:
            return

        if deadline_reached: raise TimeoutError()

        await asyncio.sleep(a_bit)


# test usage
async def test_unstable():
    good = Status(...)
    bad = Status(...)

    async def checks():
        with freeze_time() as clock:
            for status in [good, bad, good, bad]:
                yield status
                clock.tick(10)

    assert await alist(loop(checks(), **somekw)) == [False, False, False, False]


async def alist(agen): return [v async for v in agen]

simple.py

# plain old class style, simple and understandable
class Loop:
    def __init__():
        self.local_state = ...

    def next(self, status: Status) -> bool:
        now = time.time()

        # time-dependent code here
        if status.something in self.local_state.some_other:
            return False

        for blip in self.local_state.foo:
            blip.set_foo(min(now, self.local_state.old_timestamp))
            if status.check_foo_against(blip):
                return False

        return True

      
# prod usage
async def new_wait_for_idle(self: Model, **kwargs):
    # check kwrags, like apps is not a str

    loop = Loop()
    while True:
        status = check(await self.get_full_status(), **some_kw)
        if loop.next(status):
            return

        if deadline_reached: raise TimeoutError()

        await asyncio.sleep(a_bit)


# test usage
def test_unstable():
    good = Status(...)
    bad = Status(...)

    def checks():
        with freeze_time() as clock:
            for status in [good, bad, good, bad]:
                yield status
                clock.tick(10)

    assert listify(checks(), **somekw) == [False, False, False, False]


def listify(statuses, **kw):
    loop = Loop(**kw)
    return [loop.next(s) for s in statuses]