diff --git a/satella/instrumentation/memory/conditions.py b/satella/instrumentation/memory/conditions.py index e54ab40f0cfbd637b7c940989e6bf67b840630c5..7ccb0a74dde86930d012efcf69c10cf3ae9ea2bf 100644 --- a/satella/instrumentation/memory/conditions.py +++ b/satella/instrumentation/memory/conditions.py @@ -1,4 +1,5 @@ import functools +import operator import typing as tp from abc import ABCMeta, abstractmethod @@ -40,35 +41,32 @@ class ZerothSeverity(BaseCondition): class OperationJoin(BaseCondition): __slots__ = ('conditions',) + _OPERATOR = lambda y, z: y and z # pylint: disable=invalid-name + _STARTING_VALUE = False # pylint: disable=invalid-name + def __init__(self, *conditions: BaseCondition): self.conditions = conditions def can_fire(self, local_memory_data, local_maximum_consume: tp.Optional[int]) -> bool: - return functools.reduce(self.OPERATOR, ( + return functools.reduce(self._OPERATOR, ( condition.can_fire(local_memory_data, local_maximum_consume) for condition in - self.conditions), self.STARTING_VALUE) + self.conditions), self._STARTING_VALUE) class Any(OperationJoin): """This is true if one of the arguments is True""" __slots__ = () - @staticmethod - def OPERATOR(a, b): - return a or b - - STARTING_VALUE = False + _OPERATOR = operator.or_ + _STARTING_VALUE = False class All(OperationJoin): """This is true if all arguments are True""" __slots__ = () - @staticmethod - def OPERATOR(a, b): - return a and b - - STARTING_VALUE = True + _OPERATOR = operator.and_ + _STARTING_VALUE = True class Not(BaseCondition): diff --git a/tests/test_instrumentation/test_memory.py b/tests/test_instrumentation/test_memory.py index 71dbbf30defb287c80f62153a16a4de579c938ca..6254539dd2131c3038b58a21d4789df27c9fee5b 100644 --- a/tests/test_instrumentation/test_memory.py +++ b/tests/test_instrumentation/test_memory.py @@ -129,7 +129,7 @@ class TestMemory(unittest.TestCase): self.assertEqual(a['mem_normal'], 1) a['level_2_engaged'] = True time.sleep(3) - self.assertEqual(MemoryPressureManager().objects_to_cleanup_on_entered[1], []) + self.assertLessEqual(len(MemoryPressureManager().objects_to_cleanup_on_entered[1]), 1) self.assertEqual(MemoryPressureManager().severity_level, 2) self.assertEqual(a['cancelled'], 1) self.assertEqual(a['times_entered_1'], 2)