MCPcopy Create free account
hub / github.com/apache/cassandra-python-driver / TimerCallback

Class TimerCallback

tests/unit/io/utils.py:49–75  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

47
48
49class TimerCallback(object):
50
51 invoked = False
52 created_time = 0
53 invoked_time = 0
54 expected_wait = 0
55
56 def __init__(self, expected_wait):
57 self.invoked = False
58 self.created_time = time.time()
59 self.expected_wait = expected_wait
60
61 def invoke(self):
62 self.invoked_time = time.time()
63 self.invoked = True
64
65 def was_invoked(self):
66 return self.invoked
67
68 def get_wait_time(self):
69 elapsed_time = self.invoked_time - self.created_time
70 return elapsed_time
71
72 def wait_match_excepted(self):
73 if self.expected_wait - .01 <= self.get_wait_time() <= self.expected_wait + .01:
74 return True
75 return False
76
77
78def get_timeout(gross_time, start, end, precision, split_range):

Callers 3

Calls

no outgoing calls

Tested by 2