| 840 | |
| 841 | |
| 842 | class Occurrence(Expectation): |
| 843 | has_lower_bound = True |
| 844 | |
| 845 | def __init__(self, *circumstances): |
| 846 | assert circumstances |
| 847 | self.circumstances = circumstances |
| 848 | |
| 849 | self.timeline = None |
| 850 | self.timestamp = None |
| 851 | self.index = None |
| 852 | self.previous = None |
| 853 | self._next = None |
| 854 | self.observed = False |
| 855 | |
| 856 | @property |
| 857 | def next(self): |
| 858 | if self.timeline is None: |
| 859 | return None |
| 860 | |
| 861 | with self.timeline.frozen(): |
| 862 | was_last = self is self.timeline.last |
| 863 | occ = self._next |
| 864 | |
| 865 | if was_last: |
| 866 | # The .next property of the last occurrence in a timeline can change |
| 867 | # at any moment when timeline isn't frozen. So if it wasn't frozen by |
| 868 | # the caller, this was an unsafe operation, and we should complain. |
| 869 | self.timeline.expect_frozen() |
| 870 | |
| 871 | return occ |
| 872 | |
| 873 | def preceding(self): |
| 874 | it = self.and_preceding() |
| 875 | next(it) |
| 876 | return it |
| 877 | |
| 878 | def and_preceding(self, up_to=None, inclusive=False): |
| 879 | assert self.timeline is not None |
| 880 | assert up_to is None or isinstance(up_to, Expectation) |
| 881 | |
| 882 | if isinstance(up_to, Occurrence) and self < up_to: |
| 883 | return |
| 884 | |
| 885 | occ = self |
| 886 | while occ != up_to: |
| 887 | yield occ |
| 888 | occ = occ.previous |
| 889 | |
| 890 | if inclusive: |
| 891 | yield occ |
| 892 | |
| 893 | def following(self): |
| 894 | it = self.and_following() |
| 895 | next(it) |
| 896 | return it |
| 897 | |
| 898 | def and_following(self, up_to=None, inclusive=False): |
| 899 | assert self.timeline is not None |
no outgoing calls
no test coverage detected
searching dependent graphs…