(self, condition, freeze=None)
| 163 | self.observe(*occs) |
| 164 | |
| 165 | def wait_until(self, condition, freeze=None): |
| 166 | freeze = freeze or self.is_frozen |
| 167 | try: |
| 168 | with self._recorded_new: |
| 169 | # First, test the condition against the timeline as it currently is. |
| 170 | with self.frozen(): |
| 171 | result = condition() |
| 172 | if result: |
| 173 | return result |
| 174 | |
| 175 | # Now keep spinning waiting for new occurrences to come, and test the |
| 176 | # condition against every new batch in turn. |
| 177 | self.unfreeze() |
| 178 | while True: |
| 179 | self._recorded_new.wait() |
| 180 | with self.frozen(): |
| 181 | result = condition() |
| 182 | if result: |
| 183 | return result |
| 184 | assert not self.is_final |
| 185 | |
| 186 | finally: |
| 187 | if freeze: |
| 188 | self.freeze() |
| 189 | |
| 190 | def _wait_until_realized( |
| 191 | self, expectation, freeze=None, explain=True, observe=True |
no test coverage detected