Best Python code snippet using slash
test_stopping_criteria.py
Source:test_stopping_criteria.py
...95 assert str(a_and_a) == "(True and True)"96 assert str(b_and_b) == "(False and False)"97 assert str(a_and_b) == "(True and False)"98 assert str(b_and_a) == "(False and True)"99 assert a_and_a.is_met(None)100 assert not b_and_b.is_met(None)101 assert not a_and_b.is_met(None)102 assert not b_and_a.is_met(None)103def test_criterion_combination_or():104 a = AlwaysStopStoppingCriterion()105 b = NeverStopStoppingCriterion()106 a_or_a = _OrStoppingCriterion(a, a)107 b_or_b = _OrStoppingCriterion(b, b)108 a_or_b = _OrStoppingCriterion(a, b)109 b_or_a = _OrStoppingCriterion(b, a)110 assert str(a_or_a) == "(True or True)"111 assert str(b_or_b) == "(False or False)"112 assert str(a_or_b) == "(True or False)"113 assert str(b_or_a) == "(False or True)"114 assert a_or_a.is_met(None)115 assert not b_or_b.is_met(None)116 assert a_or_b.is_met(None)117 assert b_or_a.is_met(None)118def test_criterion_always():119 a = AlwaysStopStoppingCriterion()120 assert a.is_met(None)121def test_criterion_never():122 a = NeverStopStoppingCriterion()123 assert not a.is_met(None)124def test_criterion_toggleable_init_default():125 a = ToggleableStoppingCriterion()126 assert not a.is_met(None)127def test_criterion_toggleable_init_met():128 a = ToggleableStoppingCriterion(met=True)129 assert a.is_met(None)130def test_criterion_toggleable_init_not_met():131 a = ToggleableStoppingCriterion(met=False)132 assert not a.is_met(None)133def test_criterion_toggleable_set_on_off_init_met():134 a = ToggleableStoppingCriterion(met=True)135 assert a.is_met(None)136 a.on()137 assert a.is_met(None)138 a.on()139 assert a.is_met(None)140 a.off()141 assert not a.is_met(None)142def test_criterion_toggleable_set_on_off_init_not_met():143 a = ToggleableStoppingCriterion(met=False)144 assert not a.is_met(None)145 a.on()146 assert a.is_met(None)147 a.on()148 assert a.is_met(None)149 a.off()150 assert not a.is_met(None)151@pytest.mark.parametrize("is_met_expected", [True, False])152def test_criterion_custom(is_met_expected):153 # Assigning to a variable in a closure would redefine the scope, so rather use a list as a holding.154 was_called = [False]155 algo_expected = "ABC"156 def criterion_fn(algo):157 was_called[0] = True158 assert algo == algo_expected159 return is_met_expected160 criterion = CustomStoppingCriterion(criterion_fn, "Name")161 assert str(criterion) == "Name"162 assert criterion.is_met(algo_expected) == is_met_expected163 assert was_called[0]164def test_criterion_iter_count_lower():165 algo = SimpleNamespace(curr_iter=1)166 criterion = IterCountStoppingCriterion(max_iter=2)167 assert not criterion.is_met(algo)168def test_criterion_iter_count_higher():169 algo = SimpleNamespace(curr_iter=3)170 criterion = IterCountStoppingCriterion(max_iter=2)171 assert criterion.is_met(algo)172def test_criterion_iter_count_equal():173 algo = SimpleNamespace(curr_iter=2)174 criterion = IterCountStoppingCriterion(max_iter=2)175 assert criterion.is_met(algo)176def test_criterion_sample_count_lower():177 algo = SimpleNamespace(sample_count=1)178 criterion = SampleCountStoppingCriterion(max_sample_count=2)179 assert not criterion.is_met(algo)180def test_criterion_sample_count_higher():181 algo = SimpleNamespace(sample_count=3)182 criterion = SampleCountStoppingCriterion(max_sample_count=2)183 assert criterion.is_met(algo)184def test_criterion_sample_count_equal():185 algo = SimpleNamespace(sample_count=2)186 criterion = SampleCountStoppingCriterion(max_sample_count=2)187 assert criterion.is_met(algo)188# noinspection PyTypeChecker189def test_criterion_rollout_based_no_sampler():190 algo = SimpleNamespace()191 criterion = MinReturnStoppingCriterion(return_threshold=None)192 with pytest.raises(pyrado.ValueErr):193 criterion.is_met(algo)194# noinspection PyTypeChecker195def test_criterion_rollout_based_wrong_sampler():196 sampler = SimpleNamespace()197 algo = SimpleNamespace(sampler=sampler)198 criterion = MinReturnStoppingCriterion(return_threshold=None)199 with pytest.raises(pyrado.TypeErr):200 criterion.is_met(algo)201# noinspection PyTypeChecker202@pytest.mark.parametrize(203 ["statistic", "expected"],204 [205 (ReturnStatistic.min, 1),206 (ReturnStatistic.max, 6),207 (ReturnStatistic.median, 2),208 (ReturnStatistic.mean, 3),209 (ReturnStatistic.variance, 14 / 3),210 ],211)212def test_criterion_return_statistic_based_check_min(statistic, expected):213 rollout_a = SimpleNamespace(undiscounted_return=lambda: 6)214 rollout_b = SimpleNamespace(undiscounted_return=lambda: 2)215 rollout_c = SimpleNamespace(undiscounted_return=lambda: 1)216 sampler = RolloutSavingWrapper(MockSampler([rollout_a, rollout_b, rollout_c]))217 sampler.sample()218 algo = SimpleNamespace(sampler=sampler)219 criterion = ExposingReturnStatisticBasedStoppingCriterion(return_statistic=statistic)220 criterion.is_met(algo)221 assert np.isclose(criterion.return_statistic_value, expected)222# noinspection PyTypeChecker223def test_criterion_rollout_based_min_min_return_lower():224 rollout_a = SimpleNamespace(undiscounted_return=lambda: 1)225 sampler = RolloutSavingWrapper(MockSampler([rollout_a]))226 sampler.sample()227 algo = SimpleNamespace(sampler=sampler)228 criterion = MinReturnStoppingCriterion(return_threshold=2)229 assert not criterion.is_met(algo)230# noinspection PyTypeChecker231def test_criterion_rollout_based_min_min_return_higher():232 rollout_a = SimpleNamespace(undiscounted_return=lambda: 3)233 sampler = RolloutSavingWrapper(MockSampler([rollout_a]))234 sampler.sample()235 algo = SimpleNamespace(sampler=sampler)236 criterion = MinReturnStoppingCriterion(return_threshold=2)237 assert criterion.is_met(algo)238# noinspection PyTypeChecker239def test_criterion_rollout_based_min_min_return_equal():240 rollout_a = SimpleNamespace(undiscounted_return=lambda: 2)241 sampler = RolloutSavingWrapper(MockSampler([rollout_a]))242 sampler.sample()243 algo = SimpleNamespace(sampler=sampler)244 criterion = MinReturnStoppingCriterion(return_threshold=2)245 assert criterion.is_met(algo)246@pytest.mark.parametrize(247 ["num_iter", "expected"], [(None, [1, 2, 3]), (1, [3]), (2, [2, 3]), (3, [1, 2, 3]), (4, None)]248)249def test_criterion_rollout_based_convergence_subset(num_iter, expected):250 criterion = ConvergenceStoppingCriterion(num_iter=num_iter)251 criterion._return_statistic_history = [1, 2, 3]252 assert criterion._get_relevant_return_statistic_subset() == expected253# noinspection PyTypeChecker254def test_criterion_rollout_based_convergence_history_filling():255 rollouts = [SimpleNamespace(undiscounted_return=(lambda k: lambda: k)(n)) for n in range(10)]256 mock_sampler = MockSampler()257 sampler = RolloutSavingWrapper(mock_sampler)258 algo = SimpleNamespace(sampler=sampler)259 criterion = ConvergenceStoppingCriterion()260 for i, rollout in enumerate(rollouts):261 mock_sampler.step_sequences = [rollout]262 sampler.sample()263 criterion.is_met(algo)264 assert criterion._return_statistic_history == np.arange(i + 1).tolist()265def test_criterion_rollout_based_convergence_regress_constant_zero():266 criterion = ConvergenceStoppingCriterion()267 criterion._return_statistic_history = np.zeros(10).tolist()268 assert np.isclose(criterion._compute_convergence_probability(), 1.0)269def test_criterion_rollout_based_convergence_regress_constant_one():270 criterion = ConvergenceStoppingCriterion()271 criterion._return_statistic_history = np.ones(10).tolist()272 assert np.isclose(criterion._compute_convergence_probability(), 1.0)273def test_criterion_rollout_based_convergence_regress_not_constant():274 criterion = ConvergenceStoppingCriterion()275 criterion._return_statistic_history = np.arange(10).tolist()276 assert np.isclose(criterion._compute_convergence_probability(), 0.0)277def test_criterion_rollout_based_convergence_regress_random():278 criterion = ConvergenceStoppingCriterion()279 criterion._return_statistic_history = np.random.default_rng(seed=5).normal(loc=0.0, scale=0.001, size=10000)280 assert criterion._compute_convergence_probability() > 0.9281# noinspection PyTypeChecker282def test_criterion_rollout_based_convergence_lower():283 rollout = SimpleNamespace(undiscounted_return=lambda: 0)284 sampler = RolloutSavingWrapper(MockSampler([rollout]))285 sampler.sample()286 algo = SimpleNamespace(sampler=sampler)287 criterion = ConvergenceStoppingCriterion(convergence_probability_threshold=0.5)288 criterion._compute_convergence_probability = lambda: 0.0289 assert not criterion.is_met(algo)290# noinspection PyTypeChecker291def test_criterion_rollout_based_convergence_higher():292 rollout = SimpleNamespace(undiscounted_return=lambda: 0)293 sampler = RolloutSavingWrapper(MockSampler([rollout]))294 sampler.sample()295 algo = SimpleNamespace(sampler=sampler)296 criterion = ConvergenceStoppingCriterion(convergence_probability_threshold=0.5)297 criterion._compute_convergence_probability = lambda: 1.0298 assert criterion.is_met(algo)299# noinspection PyTypeChecker300def test_criterion_rollout_based_convergence_equal():301 rollout = SimpleNamespace(undiscounted_return=lambda: 0)302 sampler = RolloutSavingWrapper(MockSampler([rollout]))303 sampler.sample()304 algo = SimpleNamespace(sampler=sampler)305 criterion = ConvergenceStoppingCriterion(convergence_probability_threshold=0.5)306 criterion._compute_convergence_probability = lambda: 0.5307 assert criterion.is_met(algo)308# noinspection PyTypeChecker309def test_criterion_rollout_based_convergence_none():310 rollout = SimpleNamespace(undiscounted_return=lambda: 0)311 sampler = RolloutSavingWrapper(MockSampler([rollout]))312 sampler.sample()313 algo = SimpleNamespace(sampler=sampler)314 criterion = ConvergenceStoppingCriterion(convergence_probability_threshold=0.5)315 criterion._compute_convergence_probability = lambda: None...
test_ready_to_reschedule_dep.py
Source:test_ready_to_reschedule_dep.py
...58 return reschedule59 def test_should_pass_if_ignore_in_reschedule_period_is_set(self):60 ti = self._get_task_instance(State.UP_FOR_RESCHEDULE)61 dep_context = DepContext(ignore_in_reschedule_period=True)62 assert ReadyToRescheduleDep().is_met(ti=ti, dep_context=dep_context)63 def test_should_pass_if_not_reschedule_mode(self):64 ti = self._get_task_instance(State.UP_FOR_RESCHEDULE)65 del ti.task.reschedule66 assert ReadyToRescheduleDep().is_met(ti=ti)67 def test_should_pass_if_not_in_none_state(self):68 ti = self._get_task_instance(State.UP_FOR_RETRY)69 assert ReadyToRescheduleDep().is_met(ti=ti)70 @patch('airflow.models.taskreschedule.TaskReschedule.query_for_task_instance')71 def test_should_pass_if_no_reschedule_record_exists(self, mock_query_for_task_instance):72 mock_query_for_task_instance.return_value.with_entities.return_value.first.return_value = []73 ti = self._get_task_instance(State.NONE)74 assert ReadyToRescheduleDep().is_met(ti=ti)75 @patch('airflow.models.taskreschedule.TaskReschedule.query_for_task_instance')76 def test_should_pass_after_reschedule_date_one(self, mock_query_for_task_instance):77 mock_query_for_task_instance.return_value.with_entities.return_value.first.return_value = (78 self._get_task_reschedule(utcnow() - timedelta(minutes=1))79 )80 ti = self._get_task_instance(State.UP_FOR_RESCHEDULE)81 assert ReadyToRescheduleDep().is_met(ti=ti)82 @patch('airflow.models.taskreschedule.TaskReschedule.query_for_task_instance')83 def test_should_pass_after_reschedule_date_multiple(self, mock_query_for_task_instance):84 mock_query_for_task_instance.return_value.with_entities.return_value.first.return_value = [85 self._get_task_reschedule(utcnow() - timedelta(minutes=21)),86 self._get_task_reschedule(utcnow() - timedelta(minutes=11)),87 self._get_task_reschedule(utcnow() - timedelta(minutes=1)),88 ][-1]89 ti = self._get_task_instance(State.UP_FOR_RESCHEDULE)90 assert ReadyToRescheduleDep().is_met(ti=ti)91 @patch('airflow.models.taskreschedule.TaskReschedule.query_for_task_instance')92 def test_should_fail_before_reschedule_date_one(self, mock_query_for_task_instance):93 mock_query_for_task_instance.return_value.with_entities.return_value.first.return_value = (94 self._get_task_reschedule(utcnow() + timedelta(minutes=1))95 )96 ti = self._get_task_instance(State.UP_FOR_RESCHEDULE)97 assert not ReadyToRescheduleDep().is_met(ti=ti)98 @patch('airflow.models.taskreschedule.TaskReschedule.query_for_task_instance')99 def test_should_fail_before_reschedule_date_multiple(self, mock_query_for_task_instance):100 mock_query_for_task_instance.return_value.with_entities.return_value.first.return_value = [101 self._get_task_reschedule(utcnow() - timedelta(minutes=19)),102 self._get_task_reschedule(utcnow() - timedelta(minutes=9)),103 self._get_task_reschedule(utcnow() + timedelta(minutes=1)),104 ][-1]105 ti = self._get_task_instance(State.UP_FOR_RESCHEDULE)106 assert not ReadyToRescheduleDep().is_met(ti=ti)107 def test_mapped_task_should_pass_if_ignore_in_reschedule_period_is_set(self):108 ti = self._get_mapped_task_instance(State.UP_FOR_RESCHEDULE)109 dep_context = DepContext(ignore_in_reschedule_period=True)110 assert ReadyToRescheduleDep().is_met(ti=ti, dep_context=dep_context)111 @patch('airflow.models.taskreschedule.TaskReschedule.query_for_task_instance')112 def test_mapped_task_should_pass_if_not_reschedule_mode(self, mock_query_for_task_instance):113 mock_query_for_task_instance.return_value.with_entities.return_value.first.return_value = []114 ti = self._get_mapped_task_instance(State.UP_FOR_RESCHEDULE)115 del ti.task.reschedule116 assert ReadyToRescheduleDep().is_met(ti=ti)117 def test_mapped_task_should_pass_if_not_in_none_state(self):118 ti = self._get_mapped_task_instance(State.UP_FOR_RETRY)119 assert ReadyToRescheduleDep().is_met(ti=ti)120 @patch('airflow.models.taskreschedule.TaskReschedule.query_for_task_instance')121 def test_mapped_should_pass_if_no_reschedule_record_exists(self, mock_query_for_task_instance):122 mock_query_for_task_instance.return_value.with_entities.return_value.first.return_value = []123 ti = self._get_mapped_task_instance(State.NONE)124 assert ReadyToRescheduleDep().is_met(ti=ti)125 @patch('airflow.models.taskreschedule.TaskReschedule.query_for_task_instance')126 def test_mapped_should_pass_after_reschedule_date_one(self, mock_query_for_task_instance):127 mock_query_for_task_instance.return_value.with_entities.return_value.first.return_value = (128 self._get_mapped_task_reschedule(utcnow() - timedelta(minutes=1))129 )130 ti = self._get_mapped_task_instance(State.UP_FOR_RESCHEDULE)131 assert ReadyToRescheduleDep().is_met(ti=ti)132 @patch('airflow.models.taskreschedule.TaskReschedule.query_for_task_instance')133 def test_mapped_task_should_pass_after_reschedule_date_multiple(self, mock_query_for_task_instance):134 mock_query_for_task_instance.return_value.with_entities.return_value.first.return_value = [135 self._get_mapped_task_reschedule(utcnow() - timedelta(minutes=21)),136 self._get_mapped_task_reschedule(utcnow() - timedelta(minutes=11)),137 self._get_mapped_task_reschedule(utcnow() - timedelta(minutes=1)),138 ][-1]139 ti = self._get_mapped_task_instance(State.UP_FOR_RESCHEDULE)140 assert ReadyToRescheduleDep().is_met(ti=ti)141 @patch('airflow.models.taskreschedule.TaskReschedule.query_for_task_instance')142 def test_mapped_task_should_fail_before_reschedule_date_one(self, mock_query_for_task_instance):143 mock_query_for_task_instance.return_value.with_entities.return_value.first.return_value = (144 self._get_mapped_task_reschedule(utcnow() + timedelta(minutes=1))145 )146 ti = self._get_mapped_task_instance(State.UP_FOR_RESCHEDULE)147 assert not ReadyToRescheduleDep().is_met(ti=ti)148 @patch('airflow.models.taskreschedule.TaskReschedule.query_for_task_instance')149 def test_mapped_task_should_fail_before_reschedule_date_multiple(self, mock_query_for_task_instance):150 mock_query_for_task_instance.return_value.with_entities.return_value.first.return_value = [151 self._get_mapped_task_reschedule(utcnow() - timedelta(minutes=19)),152 self._get_mapped_task_reschedule(utcnow() - timedelta(minutes=9)),153 self._get_mapped_task_reschedule(utcnow() + timedelta(minutes=1)),154 ][-1]155 ti = self._get_mapped_task_instance(State.UP_FOR_RESCHEDULE)...
test_dates.py
Source:test_dates.py
...37 meta = self.model._meta38 self.assertEqual(meta.verbose_name, 'day of month condition')39 self.assertEqual(meta.verbose_name_plural, 'day of month conditions')40 def test_model_is_met_method(self):41 """Test model `is_met()` method"""42 instance = DayOfMonthConditionFactory(day=2)43 # Wrong day44 with freeze_time('2016-01-01'):45 self.assertFalse(instance.is_met())46 # Correct day47 with freeze_time('2016-01-02'):48 self.assertTrue(instance.is_met())49 instance.last_executed = datetime(2016, 1, 2)50 instance.save()51 # Correct day, but the same date as 'last_executed'52 with freeze_time('2016-01-02'):53 self.assertFalse(instance.is_met())54 # Wrong day55 with freeze_time('2016-02-01'):56 self.assertFalse(instance.is_met())57 # Correct day58 with freeze_time('2016-02-02'):59 self.assertTrue(instance.is_met())60 def test_model_str_method(self):61 """Test model `__str__` method"""62 self.assertIn(str(self.instance.day), str(self.instance))63class DayOfWeekConditionTestCase(TestCase):64 """65 Test `conditioner.conditions.dates.DayOfWeekCondition` model66 """67 def setUp(self):68 super().setUp()69 self.model = DayOfWeekCondition70 self.instance = DayOfWeekConditionFactory()71 def test_model_inheritance(self):72 """Test model inheritance"""73 self.assertIsInstance(self.instance, BaseCronCondition)74 def test_model_weekday_field(self):75 """Test model 'weekday' field"""76 field = self.model._meta.get_field('weekday')77 self.assertIsInstance(field, models.PositiveSmallIntegerField)78 self.assertEqual(field.verbose_name, 'day of the week')79 self.assertEqual(field.choices, self.model.WEEKDAY_CHOICES)80 self.assertEqual(field.help_text, "Action will occur every week on that day.")81 def test_model_meta_attributes(self):82 """Test model meta attributes"""83 meta = self.model._meta84 self.assertEqual(meta.verbose_name, 'day of week condition')85 self.assertEqual(meta.verbose_name_plural, 'day of week conditions')86 def test_model_is_met_method(self):87 """Test model `is_met()` method"""88 instance = DayOfWeekConditionFactory(weekday=7)89 # Wrong weekday (January 1 2007 is Monday)90 with freeze_time('2007-01-01'):91 self.assertFalse(instance.is_met())92 # Correct weekday93 with freeze_time('2007-01-07'):94 self.assertTrue(instance.is_met())95 instance.last_executed = datetime(2007, 1, 7)96 instance.save()97 # Correct weekday, but the same date as 'last_executed'98 with freeze_time('2007-01-07'):99 self.assertFalse(instance.is_met())100 # Wrong weekday101 with freeze_time('2007-01-10'):102 self.assertFalse(instance.is_met())103 # Correct weekday104 with freeze_time('2007-01-14'):105 self.assertTrue(instance.is_met())106 def test_model_str_method(self):107 """Test model `__str__` method"""...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!