diff --git a/kopf/_core/actions/execution.py b/kopf/_core/actions/execution.py
index 79cea223..c0312c63 100644
--- a/kopf/_core/actions/execution.py
+++ b/kopf/_core/actions/execution.py
@@ -86,7 +86,6 @@ class Outcome:
subrefs: Collection[ids.HandlerId] = ()
-@dataclasses.dataclass(frozen=True)
class HandlerState:
"""
A persisted state of a single handler, as stored on the resource's status.
@@ -105,9 +104,9 @@ class HandlerState:
# started: Optional[float] = None # None means this information was lost.
# stopped: Optional[float] = None # None means it is still running (e.g. delayed).
# delayed: Optional[float] = None # None means it is finished (succeeded/failed).
- retries: int = 0
- success: bool = False
- failure: bool = False
+ retries: int
+ success: bool
+ failure: bool
@property
def finished(self) -> bool:
@@ -309,19 +308,20 @@ async def execute_handler_once(
# Definitely a temporary error, regardless of the error strictness.
except TemporaryError as e:
# Maybe false-negative, never false-positive checks to prevent extra cycles & time wasted.
- lookahead_runtime = (state.runtime + datetime.timedelta(seconds=e.delay)).total_seconds()
+ delay = e.delay or 0
+ lookahead_runtime = (state.runtime + datetime.timedelta(seconds=delay)).total_seconds()
lookahead_timeout = handler.timeout is not None and lookahead_runtime >= handler.timeout
lookahead_retries = handler.retries is not None and state.retries + 1 >= handler.retries
if lookahead_timeout:
- exc = HandlerTimeoutError(f"{handler} failed temporarily but would time out after "
+ exc_t = HandlerTimeoutError(f"{handler} failed temporarily but would time out after "
f"{handler.timeout} seconds: {str(e) or repr(e)}")
- logger.error(f"{exc}") # already formatted
- return Outcome(final=True, exception=exc, subrefs=subrefs)
+ logger.error(f"{exc_t}") # already formatted
+ return Outcome(final=True, exception=exc_t, subrefs=subrefs)
elif lookahead_retries:
- exc = HandlerRetriesError(f"{handler} failed temporarily but would exceed "
+ exc_r = HandlerRetriesError(f"{handler} failed temporarily but would exceed "
f"{handler.retries} retries: {str(e) or repr(e)}")
- logger.error(f"{exc}") # already formatted
- return Outcome(final=True, exception=exc, subrefs=subrefs)
+ logger.error(f"{exc_r}") # already formatted
+ return Outcome(final=True, exception=exc_r, subrefs=subrefs)
else:
logger.error(f"{handler} failed temporarily: {str(e) or repr(e)}")
return Outcome(final=the_last_try, exception=e, delay=e.delay, subrefs=subrefs)
@@ -348,15 +348,15 @@ async def execute_handler_once(
logger.exception(f"{handler} failed with an exception. Will ignore.")
return Outcome(final=True, subrefs=subrefs)
elif errors_mode == ErrorsMode.TEMPORARY and lookahead_timeout:
- exc = HandlerTimeoutError(f"{handler} failed with an exception but would time out after "
+ exc_t = HandlerTimeoutError(f"{handler} failed with an exception but would time out after "
f"{handler.timeout} seconds: {str(e) or repr(e)}")
- logger.exception(f"{exc}") # already formatted
- return Outcome(final=True, exception=exc, subrefs=subrefs)
+ logger.exception(f"{exc_t}") # already formatted
+ return Outcome(final=True, exception=exc_t, subrefs=subrefs)
elif errors_mode == ErrorsMode.TEMPORARY and lookahead_retries:
- exc = HandlerRetriesError(f"{handler} failed with an exception but would exceed "
+ exc_r = HandlerRetriesError(f"{handler} failed with an exception but would exceed "
f"{handler.retries} retries: {str(e) or repr(e)}")
- logger.exception(f"{exc}") # already formatted
- return Outcome(final=True, exception=exc, subrefs=subrefs)
+ logger.exception(f"{exc_r}") # already formatted
+ return Outcome(final=True, exception=exc_r, subrefs=subrefs)
elif errors_mode == ErrorsMode.TEMPORARY:
logger.exception(f"{handler} failed with an exception. Will retry.")
return Outcome(final=False, exception=e, delay=backoff, subrefs=subrefs)
diff --git a/kopf/_core/actions/progression.py b/kopf/_core/actions/progression.py
index 3dc2b0ac..4a4d5b34 100644
--- a/kopf/_core/actions/progression.py
+++ b/kopf/_core/actions/progression.py
@@ -55,6 +55,7 @@ class HandlerState(execution.HandlerState):
carried over for logging of counts/extras, and for final state purging,
but not participating in the current handling cycle.
"""
+ _basetime: datetime.datetime
# Some fields may overlap the base class's fields, but this is fine (the types are the same).
active: Optional[bool] = None # is it used in done/delays [T]? or only in counters/purges [F]?
@@ -83,25 +84,37 @@ def runtime(self) -> datetime.timedelta:
@property
def started_as_datetime(self) -> datetime.datetime:
- looptime = asyncio.get_running_loop().time()
- basetime = datetime.datetime.utcnow() - datetime.timedelta(seconds=looptime)
- return basetime + datetime.timedelta(seconds=self.started or 0) # "or" is for type-checking
+ return self._basetime + datetime.timedelta(seconds=self.started or 0) # "or" is for type-checking
+
+ @property
+ def delayed_as_datetime(self) -> datetime.datetime:
+ return self._basetime + datetime.timedelta(seconds=self.delayed or 0) # "or" is for type-checking
@classmethod
def from_scratch(cls, *, purpose: Optional[str] = None) -> "HandlerState":
+ # Beware: utcnow() changes across calls, so we fix the base time by getting it only once.
+ # The "base time" is used for stable wall-clock ↔ ︎loop-clock conversion and has no meaning.
+ looptime = asyncio.get_running_loop().time()
+ basetime = datetime.datetime.utcnow() - datetime.timedelta(seconds=looptime)
return cls(
+ _basetime=basetime,
active=True,
- started=asyncio.get_running_loop().time(),
+ started=looptime,
purpose=purpose,
)
@classmethod
def from_storage(cls, __d: progress.ProgressRecord) -> "HandlerState":
+ # Beware: utcnow() changes across calls, so we fix the base time by getting it only once.
+ # The "base time" is used for stable wall-clock ↔ ︎loop-clock conversion and has no meaning.
+ looptime = asyncio.get_running_loop().time()
+ basetime = datetime.datetime.utcnow() - datetime.timedelta(seconds=looptime)
return cls(
+ _basetime=basetime,
active=False,
- started=_parse_iso8601(__d.get('started')) or asyncio.get_running_loop().time(),
- stopped=_parse_iso8601(__d.get('stopped')),
- delayed=_parse_iso8601(__d.get('delayed')),
+ started=parse_iso8601(__d.get('started'), basetime) or looptime,
+ stopped=parse_iso8601(__d.get('stopped'), basetime),
+ delayed=parse_iso8601(__d.get('delayed'), basetime),
purpose=__d.get('purpose') if __d.get('purpose') else None,
retries=__d.get('retries') or 0,
success=__d.get('success') or False,
@@ -113,9 +126,9 @@ def from_storage(cls, __d: progress.ProgressRecord) -> "HandlerState":
def for_storage(self) -> progress.ProgressRecord:
return progress.ProgressRecord(
- started=None if self.started is None else _format_iso8601(self.started),
- stopped=None if self.stopped is None else _format_iso8601(self.stopped),
- delayed=None if self.delayed is None else _format_iso8601(self.delayed),
+ started=None if self.started is None else format_iso8601(self.started, self._basetime),
+ stopped=None if self.stopped is None else format_iso8601(self.stopped, self._basetime),
+ delayed=None if self.delayed is None else format_iso8601(self.delayed, self._basetime),
purpose=None if self.purpose is None else str(self.purpose),
retries=None if self.retries is None else int(self.retries),
success=None if self.success is None else bool(self.success),
@@ -144,6 +157,7 @@ def with_outcome(
now = asyncio.get_running_loop().time()
cls = type(self)
return cls(
+ _basetime=self._basetime,
active=self.active,
purpose=self.purpose,
started=self.started if self.started is not None else now,
@@ -350,6 +364,10 @@ def delays(self) -> Collection[float]:
processing routine, based on all delays of different origin:
e.g. postponed daemons, stopping daemons, temporarily failed handlers.
"""
+ # TODO: a big problem: state is now loop-dependent, it cannot run outside of it.
+ # all tests must be base `async def`. Which is fine for us.
+ # though, keeping the state simple (non-asyncio-dependent) could be good too.
+ # ?=> is there a way to use some other time?
now = asyncio.get_running_loop().time()
return [
max(0.0, round(handler_state.delayed - now, 6)) if handler_state.delayed else 0
@@ -392,34 +410,30 @@ def deliver_results(
@overload
-def _format_iso8601(val: None) -> None: ...
+def format_iso8601(val: None, basetime: datetime.datetime) -> None: ...
@overload
-def _format_iso8601(val: float) -> str: ...
+def format_iso8601(val: float, basetime: datetime.datetime) -> str: ...
-def _format_iso8601(val: Optional[float]) -> Optional[str]:
+def format_iso8601(val: Optional[float], basetime: datetime.datetime) -> Optional[str]:
if val is None:
return None
else:
- looptime = asyncio.get_running_loop().time()
- basetime = datetime.datetime.utcnow() - datetime.timedelta(seconds=looptime)
return (basetime + datetime.timedelta(seconds=val)).isoformat(timespec='microseconds')
@overload
-def _parse_iso8601(val: None) -> None: ...
+def parse_iso8601(val: None, basetime: datetime.datetime) -> None: ...
@overload
-def _parse_iso8601(val: str) -> float: ...
+def parse_iso8601(val: str, basetime: datetime.datetime) -> float: ...
-def _parse_iso8601(val: Optional[str]) -> Optional[float]:
+def parse_iso8601(val: Optional[str], basetime: datetime.datetime) -> Optional[float]:
if val is None:
return None
else:
- looptime = asyncio.get_running_loop().time()
- basetime = datetime.datetime.utcnow() - datetime.timedelta(seconds=looptime)
return (datetime.datetime.fromisoformat(val) - basetime).total_seconds()
diff --git a/kopf/_core/engines/peering.py b/kopf/_core/engines/peering.py
index 834102df..0400cc2c 100644
--- a/kopf/_core/engines/peering.py
+++ b/kopf/_core/engines/peering.py
@@ -151,6 +151,7 @@ async def process_peering_event(
# from other peers that existed a moment earlier, this should not be a problem.
now = datetime.datetime.utcnow()
delays = [(peer.deadline - now).total_seconds() for peer in same_peers + prio_peers]
+ print(delays)
unslept = await aiotime.sleep(delays, wakeup=stream_pressure)
if unslept is None and delays:
await touch(
diff --git a/tests/handling/indexing/test_index_exclusion.py b/tests/handling/indexing/test_index_exclusion.py
index f015613e..715d8381 100644
--- a/tests/handling/indexing/test_index_exclusion.py
+++ b/tests/handling/indexing/test_index_exclusion.py
@@ -76,9 +76,9 @@ async def test_temporary_failures_with_expired_delays_are_reindexed(
resource, namespace, settings, registry, memories, indexers, index, caplog, event_type, handlers):
caplog.set_level(logging.DEBUG)
body = {'metadata': {'namespace': namespace, 'name': 'name1'}}
- delayed = datetime.datetime(2020, 12, 31, 23, 59, 59, 0)
+ delayed = '2020-12-31T23:59:59.000000'
memory = await memories.recall(raw_body=body)
- memory.indexing_memory.indexing_state = State({'index_fn': HandlerState(delayed=delayed)})
+ memory.indexing_memory.indexing_state = State({'index_fn': HandlerState.from_storage(dict(delayed=delayed))})
await process_resource_event(
lifecycle=all_at_once,
registry=registry,
@@ -91,6 +91,7 @@ async def test_temporary_failures_with_expired_delays_are_reindexed(
event_queue=asyncio.Queue(),
resource_indexed=Toggle(), # used! only to enable indexing.
)
+ print(repr(caplog.messages))
assert handlers.index_mock.call_count == 1
@@ -156,7 +157,7 @@ async def test_removed_and_remembered_on_permanent_errors(
(dict(), datetime.datetime(2020, 12, 31, 0, 1, 0)),
(dict(delay=0), datetime.datetime(2020, 12, 31, 0, 0, 0)),
(dict(delay=9), datetime.datetime(2020, 12, 31, 0, 0, 9)),
- (dict(delay=None), None),
+ (dict(delay=None), datetime.datetime(2020, 12, 31, 0, 0, 0)),
])
@pytest.mark.usefixtures('indexed_123')
@pytest.mark.parametrize('event_type', EVENT_TYPES_WHEN_EXISTS)
@@ -184,7 +185,7 @@ async def test_removed_and_remembered_on_temporary_errors(
assert memory.indexing_memory.indexing_state['index_fn'].failure == False
assert memory.indexing_memory.indexing_state['index_fn'].success == False
assert memory.indexing_memory.indexing_state['index_fn'].message == 'boo!'
- assert memory.indexing_memory.indexing_state['index_fn'].delayed == expected_delayed
+ assert memory.indexing_memory.indexing_state['index_fn'].delayed_as_datetime == expected_delayed
@pytest.mark.usefixtures('indexed_123')
diff --git a/tests/handling/subhandling/test_subhandling.py b/tests/handling/subhandling/test_subhandling.py
index b83604fb..414bbd45 100644
--- a/tests/handling/subhandling/test_subhandling.py
+++ b/tests/handling/subhandling/test_subhandling.py
@@ -16,7 +16,7 @@
@pytest.mark.parametrize('event_type', EVENT_TYPES_WHEN_EXISTS)
async def test_1st_level(registry, settings, resource, cause_mock, event_type,
- caplog, assert_logs, k8s_mocked):
+ caplog, assert_logs, k8s_mocked, looptime):
caplog.set_level(logging.DEBUG)
cause_mock.reason = Reason.CREATE
@@ -57,7 +57,7 @@ async def sub1b(**_):
assert sub1a_mock.call_count == 1
assert sub1b_mock.call_count == 1
- assert k8s_mocked.sleep.call_count == 0
+ assert looptime == 0
assert k8s_mocked.patch.call_count == 1
assert not event_queue.empty()
@@ -76,7 +76,7 @@ async def sub1b(**_):
@pytest.mark.parametrize('event_type', EVENT_TYPES_WHEN_EXISTS)
async def test_2nd_level(registry, settings, resource, cause_mock, event_type,
- caplog, assert_logs, k8s_mocked):
+ caplog, assert_logs, k8s_mocked, looptime):
caplog.set_level(logging.DEBUG)
cause_mock.reason = Reason.CREATE
@@ -137,7 +137,7 @@ def sub1b2b(**kwargs):
assert sub1b2a_mock.call_count == 1
assert sub1b2b_mock.call_count == 1
- assert k8s_mocked.sleep.call_count == 0
+ assert looptime == 0
assert k8s_mocked.patch.call_count == 1
assert not event_queue.empty()
diff --git a/tests/handling/test_activity_triggering.py b/tests/handling/test_activity_triggering.py
index 8feab694..ac799b38 100644
--- a/tests/handling/test_activity_triggering.py
+++ b/tests/handling/test_activity_triggering.py
@@ -153,36 +153,36 @@ def sample_fn(**_):
assert mock.call_count == 3
-@pytest.mark.parametrize('activity', list(Activity))
-async def test_delays_are_simulated(settings, activity, mocker):
-
- def sample_fn(**_):
- raise TemporaryError('to be retried', delay=123)
-
- registry = OperatorRegistry()
- registry._activities.append(ActivityHandler(
- fn=sample_fn, id='id', activity=activity,
- param=None, errors=None, timeout=None, retries=3, backoff=None,
- ))
-
- with freezegun.freeze_time() as frozen:
-
- async def sleep_substitute(*_, **__):
- frozen.tick(123)
-
- sleep = mocker.patch('kopf._cogs.aiokits.aiotime.sleep', wraps=sleep_substitute)
-
- with pytest.raises(ActivityError) as e:
- await run_activity(
- registry=registry,
- settings=settings,
- activity=activity,
- lifecycle=all_at_once,
- indices=OperatorIndexers().indices,
- memo=Memo(),
- )
-
- assert sleep.call_count >= 3 # 3 retries, 1 sleep each
- assert sleep.call_count <= 4 # 3 retries, 1 final success (delay=None), not more
- if sleep.call_count > 3:
- sleep.call_args_list[-1][0][0] is None
+# @pytest.mark.parametrize('activity', list(Activity))
+# async def test_delays_are_simulated(settings, activity, mocker):
+#
+# def sample_fn(**_):
+# raise TemporaryError('to be retried', delay=123)
+#
+# registry = OperatorRegistry()
+# registry._activities.append(ActivityHandler(
+# fn=sample_fn, id='id', activity=activity,
+# param=None, errors=None, timeout=None, retries=3, backoff=None,
+# ))
+#
+# with freezegun.freeze_time() as frozen:
+#
+# async def sleep_substitute(*_, **__):
+# frozen.tick(123)
+#
+# sleep = mocker.patch('kopf._cogs.aiokits.aiotime.sleep', wraps=sleep_substitute)
+#
+# with pytest.raises(ActivityError) as e:
+# await run_activity(
+# registry=registry,
+# settings=settings,
+# activity=activity,
+# lifecycle=all_at_once,
+# indices=OperatorIndexers().indices,
+# memo=Memo(),
+# )
+#
+# assert sleep.call_count >= 3 # 3 retries, 1 sleep each
+# assert sleep.call_count <= 4 # 3 retries, 1 final success (delay=None), not more
+# if sleep.call_count > 3:
+# sleep.call_args_list[-1][0][0] is None
diff --git a/tests/handling/test_cause_handling.py b/tests/handling/test_cause_handling.py
index 5ccf3115..4e275d2e 100644
--- a/tests/handling/test_cause_handling.py
+++ b/tests/handling/test_cause_handling.py
@@ -39,7 +39,6 @@ async def test_create(registry, settings, handlers, resource, cause_mock, event_
assert not handlers.update_mock.called
assert not handlers.delete_mock.called
- assert k8s_mocked.sleep.call_count == 0
assert k8s_mocked.patch.call_count == 1
assert not event_queue.empty()
@@ -80,7 +79,6 @@ async def test_update(registry, settings, handlers, resource, cause_mock, event_
assert handlers.update_mock.call_count == 1
assert not handlers.delete_mock.called
- assert k8s_mocked.sleep.call_count == 0
assert k8s_mocked.patch.call_count == 1
assert not event_queue.empty()
@@ -123,7 +121,6 @@ async def test_delete(registry, settings, handlers, resource, cause_mock, event_
assert not handlers.update_mock.called
assert handlers.delete_mock.call_count == 1
- assert k8s_mocked.sleep.call_count == 0
assert k8s_mocked.patch.call_count == 1
assert not event_queue.empty()
@@ -194,8 +191,6 @@ async def test_free(registry, settings, handlers, resource, cause_mock, event_ty
assert not handlers.create_mock.called
assert not handlers.update_mock.called
assert not handlers.delete_mock.called
-
- assert not k8s_mocked.sleep.called
assert not k8s_mocked.patch.called
assert event_queue.empty()
@@ -226,8 +221,6 @@ async def test_noop(registry, settings, handlers, resource, cause_mock, event_ty
assert not handlers.create_mock.called
assert not handlers.update_mock.called
assert not handlers.delete_mock.called
-
- assert not k8s_mocked.sleep.called
assert not k8s_mocked.patch.called
assert event_queue.empty()
diff --git a/tests/handling/test_delays.py b/tests/handling/test_delays.py
index 3b00dd74..1f1a8070 100644
--- a/tests/handling/test_delays.py
+++ b/tests/handling/test_delays.py
@@ -51,7 +51,6 @@ async def test_delayed_handlers_progress(
assert handlers.delete_mock.call_count == (1 if cause_reason == Reason.DELETE else 0)
assert handlers.resume_mock.call_count == (1 if cause_reason == Reason.RESUME else 0)
- assert not k8s_mocked.sleep.called
assert k8s_mocked.patch.called
fname = f'{cause_reason}_fn'
@@ -71,21 +70,21 @@ async def test_delayed_handlers_progress(
], ids=['fast', 'slow'])
async def test_delayed_handlers_sleep(
registry, settings, handlers, resource, cause_mock, cause_reason,
- caplog, assert_logs, k8s_mocked, now, delayed_iso, delay):
+ caplog, assert_logs, k8s_mocked, now, delayed_iso, delay, looptime):
caplog.set_level(logging.DEBUG)
# Simulate the original persisted state of the resource.
# Make sure the finalizer is added since there are mandatory deletion handlers.
- started_dt = datetime.datetime.fromisoformat('2000-01-01T00:00:00') # long time ago is fine.
- delayed_dt = datetime.datetime.fromisoformat(delayed_iso)
+ started_dt = '2020-01-01T00:00:00'
+ delayed_dt = delayed_iso
event_type = None if cause_reason == Reason.RESUME else 'irrelevant'
event_body = {
'metadata': {'finalizers': [settings.persistence.finalizer]},
'status': {'kopf': {'progress': {
- 'create_fn': HandlerState(started=started_dt, delayed=delayed_dt).as_in_storage(),
- 'update_fn': HandlerState(started=started_dt, delayed=delayed_dt).as_in_storage(),
- 'delete_fn': HandlerState(started=started_dt, delayed=delayed_dt).as_in_storage(),
- 'resume_fn': HandlerState(started=started_dt, delayed=delayed_dt).as_in_storage(),
+ 'create_fn': HandlerState.from_storage(dict(started=started_dt, delayed=delayed_dt)).as_in_storage(),
+ 'update_fn': HandlerState.from_storage(dict(started=started_dt, delayed=delayed_dt)).as_in_storage(),
+ 'delete_fn': HandlerState.from_storage(dict(started=started_dt, delayed=delayed_dt)).as_in_storage(),
+ 'resume_fn': HandlerState.from_storage(dict(started=started_dt, delayed=delayed_dt)).as_in_storage(),
}}}
}
cause_mock.reason = cause_reason
@@ -113,8 +112,7 @@ async def test_delayed_handlers_sleep(
assert 'dummy' in k8s_mocked.patch.call_args_list[-1][1]['payload']['status']['kopf']
# The duration of sleep should be as expected.
- assert k8s_mocked.sleep.called
- assert k8s_mocked.sleep.call_args_list[0][0][0] == delay
+ assert looptime == delay
assert_logs([
r"Sleeping for ([\d\.]+|[\d\.]+ \(capped [\d\.]+\)) seconds",
diff --git a/tests/handling/test_error_handling.py b/tests/handling/test_error_handling.py
index 0a3f412b..9f57ae72 100644
--- a/tests/handling/test_error_handling.py
+++ b/tests/handling/test_error_handling.py
@@ -16,7 +16,7 @@
@pytest.mark.parametrize('cause_type', HANDLER_REASONS)
async def test_fatal_error_stops_handler(
registry, settings, handlers, extrahandlers, resource, cause_mock, cause_type,
- caplog, assert_logs, k8s_mocked):
+ caplog, assert_logs, k8s_mocked, looptime):
caplog.set_level(logging.DEBUG)
name1 = f'{cause_type}_fn'
@@ -44,7 +44,7 @@ async def test_fatal_error_stops_handler(
assert handlers.delete_mock.call_count == (1 if cause_type == Reason.DELETE else 0)
assert handlers.resume_mock.call_count == (1 if cause_type == Reason.RESUME else 0)
- assert not k8s_mocked.sleep.called
+ assert looptime == 0
assert k8s_mocked.patch.called
patch = k8s_mocked.patch.call_args_list[0][1]['payload']
@@ -61,7 +61,7 @@ async def test_fatal_error_stops_handler(
@pytest.mark.parametrize('cause_type', HANDLER_REASONS)
async def test_retry_error_delays_handler(
registry, settings, handlers, extrahandlers, resource, cause_mock, cause_type,
- caplog, assert_logs, k8s_mocked):
+ caplog, assert_logs, k8s_mocked, looptime):
caplog.set_level(logging.DEBUG)
name1 = f'{cause_type}_fn'
@@ -89,7 +89,7 @@ async def test_retry_error_delays_handler(
assert handlers.delete_mock.call_count == (1 if cause_type == Reason.DELETE else 0)
assert handlers.resume_mock.call_count == (1 if cause_type == Reason.RESUME else 0)
- assert not k8s_mocked.sleep.called
+ assert looptime == 0
assert k8s_mocked.patch.called
patch = k8s_mocked.patch.call_args_list[0][1]['payload']
@@ -107,7 +107,7 @@ async def test_retry_error_delays_handler(
@pytest.mark.parametrize('cause_type', HANDLER_REASONS)
async def test_arbitrary_error_delays_handler(
registry, settings, handlers, extrahandlers, resource, cause_mock, cause_type,
- caplog, assert_logs, k8s_mocked):
+ caplog, assert_logs, k8s_mocked, looptime):
caplog.set_level(logging.DEBUG)
name1 = f'{cause_type}_fn'
@@ -135,7 +135,7 @@ async def test_arbitrary_error_delays_handler(
assert handlers.delete_mock.call_count == (1 if cause_type == Reason.DELETE else 0)
assert handlers.resume_mock.call_count == (1 if cause_type == Reason.RESUME else 0)
- assert not k8s_mocked.sleep.called
+ assert looptime == 0
assert k8s_mocked.patch.called
patch = k8s_mocked.patch.call_args_list[0][1]['payload']
diff --git a/tests/handling/test_multistep.py b/tests/handling/test_multistep.py
index 93403ac1..a0340a40 100644
--- a/tests/handling/test_multistep.py
+++ b/tests/handling/test_multistep.py
@@ -18,7 +18,7 @@
@pytest.mark.parametrize('cause_type', HANDLER_REASONS)
async def test_1st_step_stores_progress_by_patching(
registry, settings, handlers, extrahandlers,
- resource, cause_mock, cause_type, k8s_mocked, deletion_ts):
+ resource, cause_mock, cause_type, k8s_mocked, looptime, deletion_ts):
name1 = f'{cause_type}_fn'
name2 = f'{cause_type}_fn2'
@@ -46,7 +46,7 @@ async def test_1st_step_stores_progress_by_patching(
assert handlers.delete_mock.call_count == (1 if cause_type == Reason.DELETE else 0)
assert handlers.resume_mock.call_count == (1 if cause_type == Reason.RESUME else 0)
- assert not k8s_mocked.sleep.called
+ assert looptime == 0
assert k8s_mocked.patch.called
patch = k8s_mocked.patch.call_args_list[0][1]['payload']
@@ -74,7 +74,7 @@ async def test_1st_step_stores_progress_by_patching(
@pytest.mark.parametrize('cause_type', HANDLER_REASONS)
async def test_2nd_step_finishes_the_handlers(caplog,
registry, settings, handlers, extrahandlers,
- resource, cause_mock, cause_type, k8s_mocked, deletion_ts):
+ resource, cause_mock, cause_type, k8s_mocked, looptime, deletion_ts):
name1 = f'{cause_type}_fn'
name2 = f'{cause_type}_fn2'
@@ -106,7 +106,7 @@ async def test_2nd_step_finishes_the_handlers(caplog,
assert extrahandlers.delete_mock.call_count == (1 if cause_type == Reason.DELETE else 0)
assert extrahandlers.resume_mock.call_count == (1 if cause_type == Reason.RESUME else 0)
- assert not k8s_mocked.sleep.called
+ assert looptime == 0
assert k8s_mocked.patch.called
patch = k8s_mocked.patch.call_args_list[0][1]['payload']
diff --git a/tests/handling/test_no_handlers.py b/tests/handling/test_no_handlers.py
index c62d9dad..50c12f31 100644
--- a/tests/handling/test_no_handlers.py
+++ b/tests/handling/test_no_handlers.py
@@ -46,7 +46,6 @@ async def test_skipped_with_no_handlers(
event_queue=asyncio.Queue(),
)
- assert not k8s_mocked.sleep.called
assert k8s_mocked.patch.called
# The patch must contain ONLY the last-seen update, and nothing else.
@@ -102,6 +101,5 @@ async def test_stealth_mode_with_mismatching_handlers(
event_queue=asyncio.Queue(),
)
- assert not k8s_mocked.sleep.called
assert not k8s_mocked.patch.called
assert not caplog.messages # total stealth mode!
diff --git a/tests/handling/test_retrying_limits.py b/tests/handling/test_retrying_limits.py
index 6bc4e7bb..9893fd8f 100644
--- a/tests/handling/test_retrying_limits.py
+++ b/tests/handling/test_retrying_limits.py
@@ -20,7 +20,7 @@
], ids=['slow'])
async def test_timed_out_handler_fails(
registry, settings, handlers, extrahandlers, resource, cause_mock, cause_type,
- caplog, assert_logs, k8s_mocked, now, ts):
+ caplog, assert_logs, k8s_mocked, looptime, now, ts):
caplog.set_level(logging.DEBUG)
name1 = f'{cause_type}_fn'
@@ -54,7 +54,7 @@ async def test_timed_out_handler_fails(
assert not handlers.resume_mock.called
# Progress is reset, as the handler is not going to retry.
- assert not k8s_mocked.sleep.called
+ assert looptime == 0
assert k8s_mocked.patch.called
patch = k8s_mocked.patch.call_args_list[0][1]['payload']
@@ -71,7 +71,7 @@ async def test_timed_out_handler_fails(
@pytest.mark.parametrize('cause_type', HANDLER_REASONS)
async def test_retries_limited_handler_fails(
registry, settings, handlers, extrahandlers, resource, cause_mock, cause_type,
- caplog, assert_logs, k8s_mocked):
+ caplog, assert_logs, k8s_mocked, looptime):
caplog.set_level(logging.DEBUG)
name1 = f'{cause_type}_fn'
@@ -104,7 +104,7 @@ async def test_retries_limited_handler_fails(
assert not handlers.resume_mock.called
# Progress is reset, as the handler is not going to retry.
- assert not k8s_mocked.sleep.called
+ assert looptime == 0
assert k8s_mocked.patch.called
patch = k8s_mocked.patch.call_args_list[0][1]['payload']
diff --git a/tests/lifecycles/test_handler_selection.py b/tests/lifecycles/test_handler_selection.py
index aca6bdac..6b0a2e79 100644
--- a/tests/lifecycles/test_handler_selection.py
+++ b/tests/lifecycles/test_handler_selection.py
@@ -12,7 +12,7 @@
kopf.lifecycles.shuffled,
kopf.lifecycles.asap,
])
-def test_with_empty_input(lifecycle):
+async def test_with_empty_input(lifecycle):
state = State.from_scratch()
handlers = []
selected = lifecycle(handlers, state=state)
@@ -79,7 +79,7 @@ def test_shuffled_takes_them_all():
assert set(selected) == {handler1, handler2, handler3}
-def test_asap_takes_the_first_one_when_no_retries(mocker):
+async def test_asap_takes_the_first_one_when_no_retries(mocker):
handler1 = mocker.Mock(id='id1', spec_set=['id'])
handler2 = mocker.Mock(id='id2', spec_set=['id'])
handler3 = mocker.Mock(id='id3', spec_set=['id'])
@@ -92,7 +92,7 @@ def test_asap_takes_the_first_one_when_no_retries(mocker):
assert selected[0] is handler1
-def test_asap_takes_the_least_retried(mocker):
+async def test_asap_takes_the_least_retried(mocker):
handler1 = mocker.Mock(id='id1', spec_set=['id'])
handler2 = mocker.Mock(id='id2', spec_set=['id'])
handler3 = mocker.Mock(id='id3', spec_set=['id'])
diff --git a/tests/peering/test_freeze_mode.py b/tests/peering/test_freeze_mode.py
index 3e42b59c..6428880c 100644
--- a/tests/peering/test_freeze_mode.py
+++ b/tests/peering/test_freeze_mode.py
@@ -1,3 +1,5 @@
+import asyncio
+
import freezegun
import pytest
@@ -7,7 +9,7 @@
async def test_other_peering_objects_are_ignored(
- mocker, k8s_mocked, settings,
+ mocker, k8s_mocked, settings, looptime,
peering_resource, peering_namespace):
status = mocker.Mock()
@@ -30,12 +32,12 @@ async def test_other_peering_objects_are_ignored(
)
assert not status.items.called
assert not k8s_mocked.patch.called
- assert k8s_mocked.sleep.call_count == 0
+ assert looptime == 0
@freezegun.freeze_time('2020-12-31T23:59:59.123456')
async def test_toggled_on_for_higher_priority_peer_when_initially_off(
- k8s_mocked, caplog, assert_logs, settings,
+ k8s_mocked, caplog, assert_logs, settings, looptime, event_loop,
peering_resource, peering_namespace):
event = bodies.RawEvent(
@@ -54,13 +56,15 @@ async def test_toggled_on_for_higher_priority_peer_when_initially_off(
settings.peering.priority = 100
conflicts_found = aiotoggles.Toggle(False)
- k8s_mocked.sleep.return_value = 1 # as if interrupted by stream pressure
+ stream_pressure = asyncio.Event()
+ event_loop.call_later(1.23, stream_pressure.set)
caplog.set_level(0)
assert conflicts_found.is_off()
await process_peering_event(
raw_event=event,
conflicts_found=conflicts_found,
+ stream_pressure=stream_pressure,
autoclean=False,
namespace=peering_namespace,
resource=peering_resource,
@@ -68,8 +72,7 @@ async def test_toggled_on_for_higher_priority_peer_when_initially_off(
settings=settings,
)
assert conflicts_found.is_on()
- assert k8s_mocked.sleep.call_count == 1
- assert 9 < k8s_mocked.sleep.call_args[0][0][0] < 10
+ assert looptime == 1.23
assert not k8s_mocked.patch.called
assert_logs(["Pausing operations in favour of"], prohibited=[
"Possibly conflicting operators",
@@ -80,7 +83,7 @@ async def test_toggled_on_for_higher_priority_peer_when_initially_off(
@freezegun.freeze_time('2020-12-31T23:59:59.123456')
async def test_ignored_for_higher_priority_peer_when_already_on(
- k8s_mocked, caplog, assert_logs, settings,
+ k8s_mocked, caplog, assert_logs, settings, looptime, event_loop,
peering_resource, peering_namespace):
event = bodies.RawEvent(
@@ -99,13 +102,15 @@ async def test_ignored_for_higher_priority_peer_when_already_on(
settings.peering.priority = 100
conflicts_found = aiotoggles.Toggle(True)
- k8s_mocked.sleep.return_value = 1 # as if interrupted by stream pressure
+ stream_pressure = asyncio.Event()
+ event_loop.call_later(1.23, stream_pressure.set)
caplog.set_level(0)
assert conflicts_found.is_on()
await process_peering_event(
raw_event=event,
conflicts_found=conflicts_found,
+ stream_pressure=stream_pressure,
autoclean=False,
namespace=peering_namespace,
resource=peering_resource,
@@ -113,8 +118,7 @@ async def test_ignored_for_higher_priority_peer_when_already_on(
settings=settings,
)
assert conflicts_found.is_on()
- assert k8s_mocked.sleep.call_count == 1
- assert 9 < k8s_mocked.sleep.call_args[0][0][0] < 10
+ assert looptime == 1.23
assert not k8s_mocked.patch.called
assert_logs([], prohibited=[
"Possibly conflicting operators",
@@ -126,7 +130,7 @@ async def test_ignored_for_higher_priority_peer_when_already_on(
@freezegun.freeze_time('2020-12-31T23:59:59.123456')
async def test_toggled_off_for_lower_priority_peer_when_initially_on(
- k8s_mocked, caplog, assert_logs, settings,
+ k8s_mocked, caplog, assert_logs, settings, looptime, event_loop,
peering_resource, peering_namespace):
event = bodies.RawEvent(
@@ -145,13 +149,15 @@ async def test_toggled_off_for_lower_priority_peer_when_initially_on(
settings.peering.priority = 100
conflicts_found = aiotoggles.Toggle(True)
- k8s_mocked.sleep.return_value = 1 # as if interrupted by stream pressure
+ stream_pressure = asyncio.Event()
+ event_loop.call_later(1.23, stream_pressure.set)
caplog.set_level(0)
assert conflicts_found.is_on()
await process_peering_event(
raw_event=event,
conflicts_found=conflicts_found,
+ stream_pressure=stream_pressure,
autoclean=False,
namespace=peering_namespace,
resource=peering_resource,
@@ -159,8 +165,7 @@ async def test_toggled_off_for_lower_priority_peer_when_initially_on(
settings=settings,
)
assert conflicts_found.is_off()
- assert k8s_mocked.sleep.call_count == 1
- assert k8s_mocked.sleep.call_args[0][0] == []
+ assert looptime == 0
assert not k8s_mocked.patch.called
assert_logs(["Resuming operations after the pause"], prohibited=[
"Possibly conflicting operators",
@@ -171,7 +176,7 @@ async def test_toggled_off_for_lower_priority_peer_when_initially_on(
@freezegun.freeze_time('2020-12-31T23:59:59.123456')
async def test_ignored_for_lower_priority_peer_when_already_off(
- k8s_mocked, caplog, assert_logs, settings,
+ k8s_mocked, caplog, assert_logs, settings, looptime, event_loop,
peering_resource, peering_namespace):
event = bodies.RawEvent(
@@ -190,13 +195,15 @@ async def test_ignored_for_lower_priority_peer_when_already_off(
settings.peering.priority = 100
conflicts_found = aiotoggles.Toggle(False)
- k8s_mocked.sleep.return_value = 1 # as if interrupted by stream pressure
+ stream_pressure = asyncio.Event()
+ event_loop.call_later(1.23, stream_pressure.set)
caplog.set_level(0)
assert conflicts_found.is_off()
await process_peering_event(
raw_event=event,
conflicts_found=conflicts_found,
+ stream_pressure=stream_pressure,
autoclean=False,
namespace=peering_namespace,
resource=peering_resource,
@@ -204,8 +211,7 @@ async def test_ignored_for_lower_priority_peer_when_already_off(
settings=settings,
)
assert conflicts_found.is_off()
- assert k8s_mocked.sleep.call_count == 1
- assert k8s_mocked.sleep.call_args[0][0] == []
+ assert looptime == 0
assert not k8s_mocked.patch.called
assert_logs([], prohibited=[
"Possibly conflicting operators",
@@ -217,7 +223,7 @@ async def test_ignored_for_lower_priority_peer_when_already_off(
@freezegun.freeze_time('2020-12-31T23:59:59.123456')
async def test_toggled_on_for_same_priority_peer_when_initially_off(
- k8s_mocked, caplog, assert_logs, settings,
+ k8s_mocked, caplog, assert_logs, settings, looptime, event_loop,
peering_resource, peering_namespace):
event = bodies.RawEvent(
@@ -236,13 +242,15 @@ async def test_toggled_on_for_same_priority_peer_when_initially_off(
settings.peering.priority = 100
conflicts_found = aiotoggles.Toggle(False)
- k8s_mocked.sleep.return_value = 1 # as if interrupted by stream pressure
+ stream_pressure = asyncio.Event()
+ event_loop.call_later(1.23, stream_pressure.set)
caplog.set_level(0)
assert conflicts_found.is_off()
await process_peering_event(
raw_event=event,
conflicts_found=conflicts_found,
+ stream_pressure=stream_pressure,
autoclean=False,
namespace=peering_namespace,
resource=peering_resource,
@@ -250,8 +258,7 @@ async def test_toggled_on_for_same_priority_peer_when_initially_off(
settings=settings,
)
assert conflicts_found.is_on()
- assert k8s_mocked.sleep.call_count == 1
- assert 9 < k8s_mocked.sleep.call_args[0][0][0] < 10
+ assert looptime == 1.23
assert not k8s_mocked.patch.called
assert_logs([
"Possibly conflicting operators",
@@ -264,7 +271,7 @@ async def test_toggled_on_for_same_priority_peer_when_initially_off(
@freezegun.freeze_time('2020-12-31T23:59:59.123456')
async def test_ignored_for_same_priority_peer_when_already_on(
- k8s_mocked, caplog, assert_logs, settings,
+ k8s_mocked, caplog, assert_logs, settings, looptime, event_loop,
peering_resource, peering_namespace):
event = bodies.RawEvent(
@@ -283,13 +290,15 @@ async def test_ignored_for_same_priority_peer_when_already_on(
settings.peering.priority = 100
conflicts_found = aiotoggles.Toggle(True)
- k8s_mocked.sleep.return_value = 1 # as if interrupted by stream pressure
+ stream_pressure = asyncio.Event()
+ event_loop.call_later(1.23, stream_pressure.set)
caplog.set_level(0)
assert conflicts_found.is_on()
await process_peering_event(
raw_event=event,
conflicts_found=conflicts_found,
+ stream_pressure=stream_pressure,
autoclean=False,
namespace=peering_namespace,
resource=peering_resource,
@@ -297,8 +306,7 @@ async def test_ignored_for_same_priority_peer_when_already_on(
settings=settings,
)
assert conflicts_found.is_on()
- assert k8s_mocked.sleep.call_count == 1
- assert 9 < k8s_mocked.sleep.call_args[0][0][0] < 10
+ assert looptime == 1.23
assert not k8s_mocked.patch.called
assert_logs([
"Possibly conflicting operators",
@@ -312,7 +320,7 @@ async def test_ignored_for_same_priority_peer_when_already_on(
@freezegun.freeze_time('2020-12-31T23:59:59.123456')
@pytest.mark.parametrize('priority', [100, 101])
async def test_resumes_immediately_on_expiration_of_blocking_peers(
- k8s_mocked, caplog, assert_logs, settings, priority,
+ k8s_mocked, caplog, assert_logs, settings, priority, looptime, event_loop,
peering_resource, peering_namespace):
event = bodies.RawEvent(
@@ -331,13 +339,14 @@ async def test_resumes_immediately_on_expiration_of_blocking_peers(
settings.peering.priority = 100
conflicts_found = aiotoggles.Toggle(True)
- k8s_mocked.sleep.return_value = None # as if finished sleeping uninterrupted
+ stream_pressure = asyncio.Event()
caplog.set_level(0)
assert conflicts_found.is_on()
await process_peering_event(
raw_event=event,
conflicts_found=conflicts_found,
+ stream_pressure=stream_pressure,
autoclean=False,
namespace=peering_namespace,
resource=peering_resource,
@@ -345,6 +354,5 @@ async def test_resumes_immediately_on_expiration_of_blocking_peers(
settings=settings,
)
assert conflicts_found.is_on()
- assert k8s_mocked.sleep.call_count == 1
- assert 9 < k8s_mocked.sleep.call_args[0][0][0] < 10
+ assert looptime == 9.876544
assert k8s_mocked.patch.called
diff --git a/tests/persistence/test_iso8601_times.py b/tests/persistence/test_iso8601_times.py
new file mode 100644
index 00000000..22d5a10e
--- /dev/null
+++ b/tests/persistence/test_iso8601_times.py
@@ -0,0 +1,63 @@
+import datetime
+
+import pytest
+
+from kopf._core.actions.progression import format_iso8601, parse_iso8601
+
+TZ = datetime.timezone.utc
+
+
+@pytest.mark.parametrize('basetime, val, expected', [
+ (datetime.datetime(2000, 1, 1), None, None),
+ # TZ-naive:
+ (datetime.datetime(2000, 1, 1), 0, '2000-01-01T00:00:00.000000'),
+ (datetime.datetime(2000, 1, 1), 123.456789, '2000-01-01T00:02:03.456789'),
+ (datetime.datetime(2000, 1, 1), -123.456789, '1999-12-31T23:57:56.543211'),
+ (datetime.datetime(2000, 1, 1, 9, 8, 7, 654321), 0, '2000-01-01T09:08:07.654321'),
+ (datetime.datetime(2000, 1, 1, 9, 8, 7, 654321), 123.456789, '2000-01-01T09:10:11.111110'),
+ (datetime.datetime(2000, 1, 1, 9, 8, 7, 654321), -123.456789, '2000-01-01T09:06:04.197532'),
+ # TZ-aware:
+ (datetime.datetime(2000, 1, 1, tzinfo=TZ), 0, '2000-01-01T00:00:00.000000+00:00'),
+ (datetime.datetime(2000, 1, 1, tzinfo=TZ), 123.456789, '2000-01-01T00:02:03.456789+00:00'),
+ (datetime.datetime(2000, 1, 1, tzinfo=TZ), -123.456789, '1999-12-31T23:57:56.543211+00:00'),
+ (datetime.datetime(2000, 1, 1, 9, 8, 7, 654321, tzinfo=TZ), 0, '2000-01-01T09:08:07.654321+00:00'),
+ (datetime.datetime(2000, 1, 1, 9, 8, 7, 654321, tzinfo=TZ), 123.456789, '2000-01-01T09:10:11.111110+00:00'),
+ (datetime.datetime(2000, 1, 1, 9, 8, 7, 654321, tzinfo=TZ), -123.456789, '2000-01-01T09:06:04.197532+00:00'),
+])
+def test_format_iso8601(val, basetime, expected):
+ result = format_iso8601(val, basetime)
+ assert result == expected
+
+
+@pytest.mark.parametrize('basetime, val, expected', [
+ (datetime.datetime(2000, 1, 1), None, None),
+ # TZ-naive:
+ (datetime.datetime(2000, 1, 1), '2000-01-01T00:00:00.000000', 0),
+ (datetime.datetime(2000, 1, 1), '2000-01-01T00:02:03.456789', 123.456789),
+ (datetime.datetime(2000, 1, 1), '1999-12-31T23:57:56.543211', -123.456789),
+ (datetime.datetime(2000, 1, 1, 9, 8, 7, 654321), '2000-01-01T09:08:07.654321', 0),
+ (datetime.datetime(2000, 1, 1, 9, 8, 7, 654321), '2000-01-01T09:10:11.111110', 123.456789),
+ (datetime.datetime(2000, 1, 1, 9, 8, 7, 654321), '2000-01-01T09:06:04.197532', -123.456789),
+ # TZ-aware:
+ (datetime.datetime(2000, 1, 1, tzinfo=TZ), '2000-01-01T00:00:00.000000+00:00', 0),
+ (datetime.datetime(2000, 1, 1, tzinfo=TZ), '2000-01-01T00:02:03.456789+00:00', 123.456789),
+ (datetime.datetime(2000, 1, 1, tzinfo=TZ), '1999-12-31T23:57:56.543211+00:00', -123.456789),
+ (datetime.datetime(2000, 1, 1, 9, 8, 7, 654321, tzinfo=TZ), '2000-01-01T09:08:07.654321+00:00', 0),
+ (datetime.datetime(2000, 1, 1, 9, 8, 7, 654321, tzinfo=TZ), '2000-01-01T09:10:11.111110+00:00', 123.456789),
+ (datetime.datetime(2000, 1, 1, 9, 8, 7, 654321, tzinfo=TZ), '2000-01-01T09:06:04.197532+00:00', -123.456789),
+])
+def test_parse_iso8601(val, basetime, expected):
+ result = parse_iso8601(val, basetime)
+ assert result == expected
+
+
+def test_format_on_stretched_interval_with_precision_loss():
+ seconds_in_100yr = (datetime.datetime(2099, 1, 1) - datetime.datetime(2000, 1, 1)).total_seconds()
+ result = format_iso8601(seconds_in_100yr, datetime.datetime(2000, 1, 1))
+ assert result == '2099-01-01T00:00:00.000000'
+
+
+def test_parse_on_stretched_interval_with_precision_loss():
+ seconds_in_100yr = (datetime.datetime(2099, 1, 1) - datetime.datetime(2000, 1, 1)).total_seconds()
+ result = parse_iso8601('2099-01-01T00:00:00.000000', datetime.datetime(2000, 1, 1))
+ assert result == seconds_in_100yr
diff --git a/tests/persistence/test_states.py b/tests/persistence/test_states.py
index c39daa1e..e094d182 100644
--- a/tests/persistence/test_states.py
+++ b/tests/persistence/test_states.py
@@ -1,3 +1,5 @@
+import asyncio
+
import datetime
from unittest.mock import Mock
@@ -40,7 +42,7 @@ def handler():
#
-def test_created_empty_from_scratch():
+async def test_created_empty_from_scratch():
state = State.from_scratch()
assert len(state) == 0
assert state.purpose is None
@@ -57,7 +59,7 @@ def test_created_empty_from_scratch():
({'status': {'kopf': {}}}),
({'status': {'kopf': {'progress': {}}}}),
])
-def test_created_empty_from_empty_storage_with_handlers(storage, handler, body):
+async def test_created_empty_from_empty_storage_with_handlers(storage, handler, body):
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
assert len(state) == 0
assert state.purpose is None
@@ -73,7 +75,7 @@ def test_created_empty_from_empty_storage_with_handlers(storage, handler, body):
({'status': {'kopf': {'progress': {'some-id': {'success': True}}}}}),
({'status': {'kopf': {'progress': {'some-id': {'failure': True}}}}}),
])
-def test_created_empty_from_filled_storage_without_handlers(storage, handler, body):
+async def test_created_empty_from_filled_storage_without_handlers(storage, handler, body):
state = State.from_storage(body=Body(body), handlers=[], storage=storage)
assert len(state) == 0
assert state.purpose is None
@@ -89,21 +91,21 @@ def test_created_empty_from_filled_storage_without_handlers(storage, handler, bo
#
-def test_created_from_storage_as_passive(storage, handler):
+async def test_created_from_storage_as_passive(storage, handler):
body = {'status': {'kopf': {'progress': {'some-id': {}}}}}
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
assert len(state) == 1
assert state['some-id'].active is False
-def test_created_from_handlers_as_active(storage, handler):
+async def test_created_from_handlers_as_active(storage, handler):
state = State.from_scratch()
state = state.with_handlers([handler])
assert len(state) == 1
assert state['some-id'].active is True
-def test_switched_from_passive_to_active(storage, handler):
+async def test_switched_from_passive_to_active(storage, handler):
body = {'status': {'kopf': {'progress': {'some-id': {'purpose': None}}}}}
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state = state.with_handlers([handler])
@@ -111,7 +113,7 @@ def test_switched_from_passive_to_active(storage, handler):
assert state['some-id'].active is True
-def test_passed_through_with_outcomes_when_passive(storage, handler):
+async def test_passed_through_with_outcomes_when_passive(storage, handler):
body = {'status': {'kopf': {'progress': {'some-id': {'purpose': None}}}}}
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state = state.with_outcomes({'some-id': Outcome(final=True)})
@@ -119,7 +121,7 @@ def test_passed_through_with_outcomes_when_passive(storage, handler):
assert state['some-id'].active is False
-def test_passed_through_with_outcomes_when_active(storage, handler):
+async def test_passed_through_with_outcomes_when_active(storage, handler):
state = State.from_scratch()
state = state.with_handlers([handler])
state = state.with_outcomes({'some-id': Outcome(final=True)})
@@ -127,14 +129,14 @@ def test_passed_through_with_outcomes_when_active(storage, handler):
assert state['some-id'].active is True
-def test_passive_states_are_not_used_in_done_calculation(storage, handler):
+async def test_passive_states_are_not_used_in_done_calculation(storage, handler):
body = {'status': {'kopf': {'progress': {'some-id': {}}}}}
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
assert len(state) == 1
assert state.done is True # because the unfinished handler state is ignored
-def test_active_states_are_used_in_done_calculation(storage, handler):
+async def test_active_states_are_used_in_done_calculation(storage, handler):
body = {'status': {'kopf': {'progress': {'some-id': {}}}}}
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state = state.with_handlers([handler])
@@ -143,7 +145,7 @@ def test_active_states_are_used_in_done_calculation(storage, handler):
@freezegun.freeze_time(TS0)
-def test_passive_states_are_not_used_in_delays_calculation(storage, handler):
+async def test_passive_states_are_not_used_in_delays_calculation(storage, handler):
body = {'status': {'kopf': {'progress': {'some-id': {'delayed': TS1_ISO}}}}}
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
assert len(state) == 1
@@ -151,7 +153,7 @@ def test_passive_states_are_not_used_in_delays_calculation(storage, handler):
@freezegun.freeze_time(TS0)
-def test_active_states_are_used_in_delays_calculation(storage, handler):
+async def test_active_states_are_used_in_delays_calculation(storage, handler):
body = {'status': {'kopf': {'progress': {'some-id': {'delayed': TS1_ISO}}}}}
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state = state.with_handlers([handler])
@@ -164,7 +166,7 @@ def test_active_states_are_used_in_delays_calculation(storage, handler):
#
-def test_created_from_purposeless_storage(storage, handler):
+async def test_created_from_purposeless_storage(storage, handler):
body = {'status': {'kopf': {'progress': {'some-id': {'purpose': None}}}}}
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
assert len(state) == 1
@@ -173,7 +175,7 @@ def test_created_from_purposeless_storage(storage, handler):
@pytest.mark.parametrize('reason', HANDLER_REASONS)
-def test_created_from_purposeful_storage(storage, handler, reason):
+async def test_created_from_purposeful_storage(storage, handler, reason):
body = {'status': {'kopf': {'progress': {'some-id': {'purpose': reason.value}}}}}
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
assert len(state) == 1
@@ -182,7 +184,7 @@ def test_created_from_purposeful_storage(storage, handler, reason):
@pytest.mark.parametrize('reason', HANDLER_REASONS)
-def test_enriched_with_handlers_keeps_the_original_purpose(handler, reason):
+async def test_enriched_with_handlers_keeps_the_original_purpose(handler, reason):
state = State.from_scratch()
state = state.with_purpose(reason)
state = state.with_handlers([handler])
@@ -190,7 +192,7 @@ def test_enriched_with_handlers_keeps_the_original_purpose(handler, reason):
@pytest.mark.parametrize('reason', HANDLER_REASONS)
-def test_enriched_with_outcomes_keeps_the_original_purpose(reason):
+async def test_enriched_with_outcomes_keeps_the_original_purpose(reason):
state = State.from_scratch()
state = state.with_purpose(reason)
state = state.with_outcomes({})
@@ -198,7 +200,7 @@ def test_enriched_with_outcomes_keeps_the_original_purpose(reason):
@pytest.mark.parametrize('reason', HANDLER_REASONS)
-def test_repurposed_before_handlers(handler, reason):
+async def test_repurposed_before_handlers(handler, reason):
state = State.from_scratch()
state = state.with_purpose(reason).with_handlers([handler])
assert len(state) == 1
@@ -207,7 +209,7 @@ def test_repurposed_before_handlers(handler, reason):
@pytest.mark.parametrize('reason', HANDLER_REASONS)
-def test_repurposed_after_handlers(handler, reason):
+async def test_repurposed_after_handlers(handler, reason):
state = State.from_scratch()
state = state.with_handlers([handler]).with_purpose(reason)
assert len(state) == 1
@@ -216,7 +218,7 @@ def test_repurposed_after_handlers(handler, reason):
@pytest.mark.parametrize('reason', HANDLER_REASONS)
-def test_repurposed_with_handlers(handler, reason):
+async def test_repurposed_with_handlers(handler, reason):
state = State.from_scratch()
state = state.with_handlers([handler]).with_purpose(reason, handlers=[handler])
assert len(state) == 1
@@ -225,7 +227,7 @@ def test_repurposed_with_handlers(handler, reason):
@pytest.mark.parametrize('reason', HANDLER_REASONS)
-def test_repurposed_not_affecting_the_existing_handlers_from_scratch(handler, reason):
+async def test_repurposed_not_affecting_the_existing_handlers_from_scratch(handler, reason):
state = State.from_scratch()
state = state.with_handlers([handler]).with_purpose(reason).with_handlers([handler])
assert len(state) == 1
@@ -234,7 +236,7 @@ def test_repurposed_not_affecting_the_existing_handlers_from_scratch(handler, re
@pytest.mark.parametrize('reason', HANDLER_REASONS)
-def test_repurposed_not_affecting_the_existing_handlers_from_storage(storage, handler, reason):
+async def test_repurposed_not_affecting_the_existing_handlers_from_storage(storage, handler, reason):
body = {'status': {'kopf': {'progress': {'some-id': {'purpose': None}}}}}
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state = state.with_handlers([handler]).with_purpose(reason).with_handlers([handler])
@@ -258,7 +260,7 @@ def test_repurposed_not_affecting_the_existing_handlers_from_storage(storage, ha
# All combinations except for same-to-same (it is not an "extra" then).
(a, b) for a in HANDLER_REASONS for b in HANDLER_REASONS if a != b
])
-def test_with_handlers_irrelevant_to_the_purpose(
+async def test_with_handlers_irrelevant_to_the_purpose(
storage, handler, body, expected_extras, stored_reason, processed_reason):
body['status']['kopf']['progress']['some-id']['purpose'] = stored_reason.value
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
@@ -277,7 +279,7 @@ def test_with_handlers_irrelevant_to_the_purpose(
(StateCounters(1, 0, 0), True, [], {'status': {'kopf': {'progress': {'some-id': {'success': True}}}}}),
])
@pytest.mark.parametrize('reason', HANDLER_REASONS)
-def test_with_handlers_relevant_to_the_purpose(
+async def test_with_handlers_relevant_to_the_purpose(
storage, handler, body, expected_counts, expected_done, expected_delays, reason):
body['status']['kopf']['progress']['some-id']['purpose'] = reason.value
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
@@ -296,7 +298,7 @@ def test_with_handlers_relevant_to_the_purpose(
])
@pytest.mark.parametrize('reason', HANDLER_REASONS)
@freezegun.freeze_time(TS0)
-def test_with_handlers_relevant_to_the_purpose_and_delayed(
+async def test_with_handlers_relevant_to_the_purpose_and_delayed(
storage, handler, body, expected_counts, expected_done, expected_delays, reason):
body['status']['kopf']['progress']['some-id']['delayed'] = TS1_ISO
body['status']['kopf']['progress']['some-id']['purpose'] = reason.value
@@ -311,7 +313,7 @@ def test_with_handlers_relevant_to_the_purpose_and_delayed(
@pytest.mark.parametrize('reason', [Reason.CREATE, Reason.UPDATE, Reason.RESUME])
@freezegun.freeze_time(TS0)
-def test_issue_601_deletion_supersedes_other_processing(storage, reason):
+async def test_issue_601_deletion_supersedes_other_processing(storage, reason):
body = {'status': {'kopf': {'progress': {
'fn1': {'purpose': reason.value, 'failure': True},
@@ -349,7 +351,7 @@ def test_issue_601_deletion_supersedes_other_processing(storage, reason):
@freezegun.freeze_time(TS0)
-def test_started_from_scratch(storage, handler):
+async def test_started_from_scratch(storage, handler):
patch = Patch()
state = State.from_scratch()
state = state.with_handlers([handler])
@@ -365,7 +367,7 @@ def test_started_from_scratch(storage, handler):
(TSA_ISO, {'status': {'kopf': {'progress': {'some-id': {'started': TSA_ISO}}}}}),
])
@freezegun.freeze_time(TS0)
-def test_started_from_storage(storage, handler, body, expected):
+async def test_started_from_storage(storage, handler, body, expected):
patch = Patch()
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state.store(body=Body({}), patch=patch, storage=storage)
@@ -379,13 +381,20 @@ def test_started_from_storage(storage, handler, body, expected):
(TSB_ISO, {'status': {'kopf': {'progress': {'some-id': {'started': TSB_ISO}}}}}),
(TSA_ISO, {'status': {'kopf': {'progress': {'some-id': {'started': TSA_ISO}}}}}),
])
-def test_started_from_storage_is_preferred_over_from_scratch(storage, handler, body, expected):
- with freezegun.freeze_time(TS0):
- state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
- with freezegun.freeze_time(TS1):
- state = state.with_handlers([handler])
+@freezegun.freeze_time(TS0)
+async def test_started_from_storage_is_preferred_over_from_scratch(storage, handler, body, expected, looptime):
+ # TODO: freeze_time is useless here now, since we switched to the loop time.
+ # but we check for the real UTC timestamp, which are UTC-clock-dependent.
+ # => so we have to sync both the loop time & the frozen time (somehow).
+ # and abandon the TSx timestamps concept completely.
+
patch = Patch()
- state.store(body=Body({}), patch=patch, storage=storage)
+ state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
+ with freezegun.freeze_time() as ft: # for some different time
+ ft.tick(123)
+ await asyncio.sleep(123)
+ state = state.with_handlers([handler])
+ state.store(body=Body({}), patch=patch, storage=storage)
assert patch['status']['kopf']['progress']['some-id']['started'] == expected
@@ -401,7 +410,7 @@ def test_started_from_storage_is_preferred_over_from_scratch(storage, handler, b
(TS0 - TSA, {'status': {'kopf': {'progress': {'some-id': {'started': TSA_ISO}}}}}),
])
@freezegun.freeze_time(TS0)
-def test_runtime(storage, handler, expected, body):
+async def test_runtime(storage, handler, expected, body):
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state = state.with_handlers([handler])
result = state[handler.id].runtime
@@ -421,7 +430,7 @@ def test_runtime(storage, handler, expected, body):
(True , {'status': {'kopf': {'progress': {'some-id': {'success': True}}}}}),
(True , {'status': {'kopf': {'progress': {'some-id': {'failure': True}}}}}),
])
-def test_finished_flag(storage, handler, expected, body):
+async def test_finished_flag(storage, handler, expected, body):
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state = state.with_handlers([handler])
result = state[handler.id].finished
@@ -456,7 +465,7 @@ def test_finished_flag(storage, handler, expected, body):
(True , {'status': {'kopf': {'progress': {'some-id': {'delayed': TSA_ISO, 'failure': None}}}}}),
])
@freezegun.freeze_time(TS0)
-def test_sleeping_flag(storage, handler, expected, body):
+async def test_sleeping_flag(storage, handler, expected, body):
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state = state.with_handlers([handler])
result = state[handler.id].sleeping
@@ -491,7 +500,7 @@ def test_sleeping_flag(storage, handler, expected, body):
(False, {'status': {'kopf': {'progress': {'some-id': {'delayed': TSA_ISO, 'failure': None}}}}}),
])
@freezegun.freeze_time(TS0)
-def test_awakened_flag(storage, handler, expected, body):
+async def test_awakened_flag(storage, handler, expected, body):
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state = state.with_handlers([handler])
result = state[handler.id].awakened
@@ -505,9 +514,14 @@ def test_awakened_flag(storage, handler, expected, body):
(None, {'status': {'kopf': {'progress': {}}}}),
(None, {'status': {'kopf': {'progress': {'some-id': {}}}}}),
(None, {'status': {'kopf': {'progress': {'some-id': {'delayed': None}}}}}),
- (TS0, {'status': {'kopf': {'progress': {'some-id': {'delayed': TS0_ISO}}}}}),
+ (1000, {'status': {'kopf': {'progress': {'some-id': {'delayed': TS0_ISO}}}}}),
+ (1001, {'status': {'kopf': {'progress': {'some-id': {'delayed': TS1_ISO}}}}}),
+ (999.876544, {'status': {'kopf': {'progress': {'some-id': {'delayed': TSB_ISO}}}}}),
+ (1000.876543, {'status': {'kopf': {'progress': {'some-id': {'delayed': TSA_ISO}}}}}),
])
-def test_awakening_time(storage, handler, expected, body):
+@freezegun.freeze_time(TS0)
+@pytest.mark.looptime(start=1000)
+async def test_awakening_time(storage, handler, expected, body):
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state = state.with_handlers([handler])
result = state[handler.id].delayed
@@ -522,7 +536,7 @@ def test_awakening_time(storage, handler, expected, body):
(0, {'status': {'kopf': {'progress': {'some-id': {'retries': None}}}}}),
(6, {'status': {'kopf': {'progress': {'some-id': {'retries': 6}}}}}),
])
-def test_get_retry_count(storage, handler, expected, body):
+async def test_get_retry_count(storage, handler, expected, body):
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state = state.with_handlers([handler])
result = state[handler.id].retries
@@ -535,7 +549,7 @@ def test_get_retry_count(storage, handler, expected, body):
({}, 1, TS1_ISO),
])
@freezegun.freeze_time(TS0)
-def test_set_awake_time(storage, handler, expected, body, delay):
+async def test_set_awake_time(storage, handler, expected, body, delay):
patch = Patch()
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state = state.with_handlers([handler])
@@ -558,7 +572,7 @@ def test_set_awake_time(storage, handler, expected, body, delay):
(6, TS1_ISO, 1, {'status': {'kopf': {'progress': {'some-id': {'retries': 5}}}}}),
])
@freezegun.freeze_time(TS0)
-def test_set_retry_time(storage, handler, expected_retries, expected_delayed, body, delay):
+async def test_set_retry_time(storage, handler, expected_retries, expected_delayed, body, delay):
patch = Patch()
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state = state.with_handlers([handler])
@@ -573,7 +587,7 @@ def test_set_retry_time(storage, handler, expected_retries, expected_delayed, bo
#
-def test_subrefs_added_to_empty_state(storage, handler):
+async def test_subrefs_added_to_empty_state(storage, handler):
body = {}
patch = Patch()
outcome_subrefs = ['sub2/b', 'sub2/a', 'sub2', 'sub1', 'sub3']
@@ -586,7 +600,7 @@ def test_subrefs_added_to_empty_state(storage, handler):
assert patch['status']['kopf']['progress']['some-id']['subrefs'] == expected_subrefs
-def test_subrefs_added_to_preexisting_subrefs(storage, handler):
+async def test_subrefs_added_to_preexisting_subrefs(storage, handler):
body = {'status': {'kopf': {'progress': {'some-id': {'subrefs': ['sub9/2', 'sub9/1']}}}}}
patch = Patch()
outcome_subrefs = ['sub2/b', 'sub2/a', 'sub2', 'sub1', 'sub3']
@@ -599,7 +613,7 @@ def test_subrefs_added_to_preexisting_subrefs(storage, handler):
assert patch['status']['kopf']['progress']['some-id']['subrefs'] == expected_subrefs
-def test_subrefs_ignored_when_not_specified(storage, handler):
+async def test_subrefs_ignored_when_not_specified(storage, handler):
body = {}
patch = Patch()
outcome = Outcome(final=True, subrefs=[])
@@ -620,7 +634,7 @@ def test_subrefs_ignored_when_not_specified(storage, handler):
(6, TS0_ISO, {'status': {'kopf': {'progress': {'some-id': {'retries': 5}}}}}),
])
@freezegun.freeze_time(TS0)
-def test_store_failure(storage, handler, expected_retries, expected_stopped, body):
+async def test_store_failure(storage, handler, expected_retries, expected_stopped, body):
error = Exception('some-error')
patch = Patch()
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
@@ -639,7 +653,7 @@ def test_store_failure(storage, handler, expected_retries, expected_stopped, bod
(6, TS0_ISO, {'status': {'kopf': {'progress': {'some-id': {'retries': 5}}}}}),
])
@freezegun.freeze_time(TS0)
-def test_store_success(storage, handler, expected_retries, expected_stopped, body):
+async def test_store_success(storage, handler, expected_retries, expected_stopped, body):
patch = Patch()
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
state = state.with_handlers([handler])
@@ -657,7 +671,7 @@ def test_store_success(storage, handler, expected_retries, expected_stopped, bod
('string', {'status': {'some-id': 'string'}}),
({'field': 'value'}, {'status': {'some-id': {'field': 'value'}}}),
])
-def test_store_result(handler, expected_patch, result):
+async def test_store_result(handler, expected_patch, result):
patch = Patch()
outcomes = {handler.id: Outcome(final=True, result=result)}
deliver_results(outcomes=outcomes, patch=patch)
@@ -669,7 +683,7 @@ def test_store_result(handler, expected_patch, result):
#
-def test_purge_progress_when_exists_in_body(storage, handler):
+async def test_purge_progress_when_exists_in_body(storage, handler):
body = {'status': {'kopf': {'progress': {'some-id': {'retries': 5}}}}}
patch = Patch()
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
@@ -677,7 +691,7 @@ def test_purge_progress_when_exists_in_body(storage, handler):
assert patch == {'status': {'kopf': {'progress': {'some-id': None}}}}
-def test_purge_progress_when_already_empty_in_body_and_patch(storage, handler):
+async def test_purge_progress_when_already_empty_in_body_and_patch(storage, handler):
body = {}
patch = Patch()
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
@@ -685,7 +699,7 @@ def test_purge_progress_when_already_empty_in_body_and_patch(storage, handler):
assert not patch
-def test_purge_progress_when_already_empty_in_body_but_not_in_patch(storage, handler):
+async def test_purge_progress_when_already_empty_in_body_but_not_in_patch(storage, handler):
body = {}
patch = Patch({'status': {'kopf': {'progress': {'some-id': {'retries': 5}}}}})
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
@@ -693,7 +707,7 @@ def test_purge_progress_when_already_empty_in_body_but_not_in_patch(storage, han
assert not patch
-def test_purge_progress_when_known_at_restoration_only(storage, handler):
+async def test_purge_progress_when_known_at_restoration_only(storage, handler):
body = {'status': {'kopf': {'progress': {'some-id': {'retries': 5}}}}}
patch = Patch()
state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
@@ -701,7 +715,7 @@ def test_purge_progress_when_known_at_restoration_only(storage, handler):
assert patch == {'status': {'kopf': {'progress': {'some-id': None}}}}
-def test_purge_progress_when_known_at_purge_only(storage, handler):
+async def test_purge_progress_when_known_at_purge_only(storage, handler):
body = {'status': {'kopf': {'progress': {'some-id': {'retries': 5}}}}}
patch = Patch()
state = State.from_storage(body=Body(body), handlers=[], storage=storage)
@@ -709,7 +723,7 @@ def test_purge_progress_when_known_at_purge_only(storage, handler):
assert patch == {'status': {'kopf': {'progress': {'some-id': None}}}}
-def test_purge_progress_cascades_to_subrefs(storage, handler):
+async def test_purge_progress_cascades_to_subrefs(storage, handler):
body = {'status': {'kopf': {'progress': {
'some-id': {'subrefs': ['sub1', 'sub2', 'sub3']},
'sub1': {},
@@ -727,7 +741,7 @@ def test_purge_progress_cascades_to_subrefs(storage, handler):
}}}}
-def test_original_body_is_not_modified_by_storing(storage, handler):
+async def test_original_body_is_not_modified_by_storing(storage, handler):
body = Body({})
patch = Patch()
state = State.from_storage(body=body, handlers=[handler], storage=storage)