How to use test_timedelta method in pandera

Best Python code snippet using pandera_python

test_execution.py

Source:test_execution.py Github

copy

Full Screen

1import datetime2import pytest3import pytz4from flytekit.models import common as _common_models5from flytekit.models import execution as _execution6from flytekit.models import literals as _literals7from flytekit.models.core import execution as _core_exec8from flytekit.models.core import identifier as _identifier9from tests.flytekit.common import parameterizers as _parameterizers10_INPUT_MAP = _literals.LiteralMap(11 {"a": _literals.Literal(scalar=_literals.Scalar(primitive=_literals.Primitive(integer=1)))}12)13_OUTPUT_MAP = _literals.LiteralMap(14 {"b": _literals.Literal(scalar=_literals.Scalar(primitive=_literals.Primitive(integer=2)))}15)16def test_execution_closure_with_output():17 test_datetime = datetime.datetime(year=2022, month=1, day=1, tzinfo=pytz.UTC)18 test_timedelta = datetime.timedelta(seconds=10)19 test_outputs = _execution.LiteralMapBlob(values=_OUTPUT_MAP, uri="http://foo/")20 obj = _execution.ExecutionClosure(21 phase=_core_exec.WorkflowExecutionPhase.SUCCEEDED,22 started_at=test_datetime,23 duration=test_timedelta,24 outputs=test_outputs,25 )26 assert obj.phase == _core_exec.WorkflowExecutionPhase.SUCCEEDED27 assert obj.started_at == test_datetime28 assert obj.duration == test_timedelta29 assert obj.outputs == test_outputs30 obj2 = _execution.ExecutionClosure.from_flyte_idl(obj.to_flyte_idl())31 assert obj2 == obj32 assert obj2.phase == _core_exec.WorkflowExecutionPhase.SUCCEEDED33 assert obj2.started_at == test_datetime34 assert obj2.duration == test_timedelta35 assert obj2.outputs == test_outputs36def test_execution_closure_with_error():37 test_datetime = datetime.datetime(year=2022, month=1, day=1, tzinfo=pytz.UTC)38 test_timedelta = datetime.timedelta(seconds=10)39 test_error = _core_exec.ExecutionError(40 code="foo", message="bar", error_uri="http://foobar", kind=_core_exec.ExecutionError.ErrorKind.USER41 )42 obj = _execution.ExecutionClosure(43 phase=_core_exec.WorkflowExecutionPhase.SUCCEEDED,44 started_at=test_datetime,45 duration=test_timedelta,46 error=test_error,47 )48 assert obj.phase == _core_exec.WorkflowExecutionPhase.SUCCEEDED49 assert obj.started_at == test_datetime50 assert obj.duration == test_timedelta51 assert obj.error == test_error52 obj2 = _execution.ExecutionClosure.from_flyte_idl(obj.to_flyte_idl())53 assert obj2 == obj54 assert obj2.phase == _core_exec.WorkflowExecutionPhase.SUCCEEDED55 assert obj2.started_at == test_datetime56 assert obj2.duration == test_timedelta57 assert obj2.error == test_error58def test_execution_closure_with_abort_metadata():59 test_datetime = datetime.datetime(year=2022, month=1, day=1, tzinfo=pytz.UTC)60 test_timedelta = datetime.timedelta(seconds=10)61 abort_metadata = _execution.AbortMetadata(cause="cause", principal="skinner")62 obj = _execution.ExecutionClosure(63 phase=_core_exec.WorkflowExecutionPhase.SUCCEEDED,64 started_at=test_datetime,65 duration=test_timedelta,66 abort_metadata=abort_metadata,67 )68 assert obj.phase == _core_exec.WorkflowExecutionPhase.SUCCEEDED69 assert obj.started_at == test_datetime70 assert obj.duration == test_timedelta71 assert obj.abort_metadata == abort_metadata72 obj2 = _execution.ExecutionClosure.from_flyte_idl(obj.to_flyte_idl())73 assert obj2 == obj74 assert obj2.phase == _core_exec.WorkflowExecutionPhase.SUCCEEDED75 assert obj2.started_at == test_datetime76 assert obj2.duration == test_timedelta77 assert obj2.abort_metadata == abort_metadata78def test_system_metadata():79 obj = _execution.SystemMetadata(execution_cluster="my_cluster")80 assert obj.execution_cluster == "my_cluster"81 obj2 = _execution.SystemMetadata.from_flyte_idl(obj.to_flyte_idl())82 assert obj == obj283 assert obj2.execution_cluster == "my_cluster"84def test_execution_metadata():85 scheduled_at = datetime.datetime.now()86 system_metadata = _execution.SystemMetadata(execution_cluster="my_cluster")87 parent_node_execution = _identifier.NodeExecutionIdentifier(88 node_id="node_id",89 execution_id=_identifier.WorkflowExecutionIdentifier(90 project="project",91 domain="domain",92 name="parent",93 ),94 )95 reference_execution = _identifier.WorkflowExecutionIdentifier(96 project="project",97 domain="domain",98 name="reference",99 )100 obj = _execution.ExecutionMetadata(101 _execution.ExecutionMetadata.ExecutionMode.MANUAL,102 "tester",103 1,104 scheduled_at=scheduled_at,105 parent_node_execution=parent_node_execution,106 reference_execution=reference_execution,107 system_metadata=system_metadata,108 )109 assert obj.mode == _execution.ExecutionMetadata.ExecutionMode.MANUAL110 assert obj.principal == "tester"111 assert obj.nesting == 1112 assert obj.scheduled_at == scheduled_at113 assert obj.parent_node_execution == parent_node_execution114 assert obj.reference_execution == reference_execution115 assert obj.system_metadata == system_metadata116 obj2 = _execution.ExecutionMetadata.from_flyte_idl(obj.to_flyte_idl())117 assert obj == obj2118 assert obj2.mode == _execution.ExecutionMetadata.ExecutionMode.MANUAL119 assert obj2.principal == "tester"120 assert obj2.nesting == 1121 assert obj2.scheduled_at == scheduled_at122 assert obj2.parent_node_execution == parent_node_execution123 assert obj2.reference_execution == reference_execution124 assert obj2.system_metadata == system_metadata125@pytest.mark.parametrize("literal_value_pair", _parameterizers.LIST_OF_SCALAR_LITERALS_AND_PYTHON_VALUE)126def test_execution_spec(literal_value_pair):127 literal_value, _ = literal_value_pair128 obj = _execution.ExecutionSpec(129 _identifier.Identifier(_identifier.ResourceType.LAUNCH_PLAN, "project", "domain", "name", "version"),130 _execution.ExecutionMetadata(_execution.ExecutionMetadata.ExecutionMode.MANUAL, "tester", 1),131 notifications=_execution.NotificationList(132 [133 _common_models.Notification(134 [_core_exec.WorkflowExecutionPhase.ABORTED],135 pager_duty=_common_models.PagerDutyNotification(recipients_email=["a", "b", "c"]),136 )137 ]138 ),139 raw_output_data_config=_common_models.RawOutputDataConfig(output_location_prefix="raw_output"),140 max_parallelism=100,141 )142 assert obj.launch_plan.resource_type == _identifier.ResourceType.LAUNCH_PLAN143 assert obj.launch_plan.domain == "domain"144 assert obj.launch_plan.project == "project"145 assert obj.launch_plan.name == "name"146 assert obj.launch_plan.version == "version"147 assert obj.metadata.mode == _execution.ExecutionMetadata.ExecutionMode.MANUAL148 assert obj.metadata.nesting == 1149 assert obj.metadata.principal == "tester"150 assert obj.notifications.notifications[0].phases == [_core_exec.WorkflowExecutionPhase.ABORTED]151 assert obj.notifications.notifications[0].pager_duty.recipients_email == [152 "a",153 "b",154 "c",155 ]156 assert obj.disable_all is None157 assert obj.max_parallelism == 100158 assert obj.raw_output_data_config.output_location_prefix == "raw_output"159 obj2 = _execution.ExecutionSpec.from_flyte_idl(obj.to_flyte_idl())160 assert obj == obj2161 assert obj2.launch_plan.resource_type == _identifier.ResourceType.LAUNCH_PLAN162 assert obj2.launch_plan.domain == "domain"163 assert obj2.launch_plan.project == "project"164 assert obj2.launch_plan.name == "name"165 assert obj2.launch_plan.version == "version"166 assert obj2.metadata.mode == _execution.ExecutionMetadata.ExecutionMode.MANUAL167 assert obj2.metadata.nesting == 1168 assert obj2.metadata.principal == "tester"169 assert obj2.notifications.notifications[0].phases == [_core_exec.WorkflowExecutionPhase.ABORTED]170 assert obj2.notifications.notifications[0].pager_duty.recipients_email == [171 "a",172 "b",173 "c",174 ]175 assert obj2.disable_all is None176 assert obj2.max_parallelism == 100177 assert obj2.raw_output_data_config.output_location_prefix == "raw_output"178 obj = _execution.ExecutionSpec(179 _identifier.Identifier(_identifier.ResourceType.LAUNCH_PLAN, "project", "domain", "name", "version"),180 _execution.ExecutionMetadata(_execution.ExecutionMetadata.ExecutionMode.MANUAL, "tester", 1),181 disable_all=True,182 )183 assert obj.launch_plan.resource_type == _identifier.ResourceType.LAUNCH_PLAN184 assert obj.launch_plan.domain == "domain"185 assert obj.launch_plan.project == "project"186 assert obj.launch_plan.name == "name"187 assert obj.launch_plan.version == "version"188 assert obj.metadata.mode == _execution.ExecutionMetadata.ExecutionMode.MANUAL189 assert obj.metadata.nesting == 1190 assert obj.metadata.principal == "tester"191 assert obj.notifications is None192 assert obj.disable_all is True193 obj2 = _execution.ExecutionSpec.from_flyte_idl(obj.to_flyte_idl())194 assert obj == obj2195 assert obj2.launch_plan.resource_type == _identifier.ResourceType.LAUNCH_PLAN196 assert obj2.launch_plan.domain == "domain"197 assert obj2.launch_plan.project == "project"198 assert obj2.launch_plan.name == "name"199 assert obj2.launch_plan.version == "version"200 assert obj2.metadata.mode == _execution.ExecutionMetadata.ExecutionMode.MANUAL201 assert obj2.metadata.nesting == 1202 assert obj2.metadata.principal == "tester"203 assert obj2.notifications is None204 assert obj2.disable_all is True205def test_workflow_execution_data_response():206 input_blob = _common_models.UrlBlob("in", 1)207 output_blob = _common_models.UrlBlob("out", 2)208 obj = _execution.WorkflowExecutionGetDataResponse(input_blob, output_blob, _INPUT_MAP, _OUTPUT_MAP)209 obj2 = _execution.WorkflowExecutionGetDataResponse.from_flyte_idl(obj.to_flyte_idl())210 assert obj == obj2211 assert obj2.inputs == input_blob212 assert obj2.outputs == output_blob213 assert obj2.full_inputs == _INPUT_MAP214 assert obj2.full_outputs == _OUTPUT_MAP215def test_node_execution_data_response():216 input_blob = _common_models.UrlBlob("in", 1)217 output_blob = _common_models.UrlBlob("out", 2)218 obj = _execution.NodeExecutionGetDataResponse(input_blob, output_blob, _INPUT_MAP, _OUTPUT_MAP)219 obj2 = _execution.NodeExecutionGetDataResponse.from_flyte_idl(obj.to_flyte_idl())220 assert obj == obj2221 assert obj2.inputs == input_blob222 assert obj2.outputs == output_blob223 assert obj2.full_inputs == _INPUT_MAP224 assert obj2.full_outputs == _OUTPUT_MAP225def test_task_execution_data_response():226 input_blob = _common_models.UrlBlob("in", 1)227 output_blob = _common_models.UrlBlob("out", 2)228 obj = _execution.TaskExecutionGetDataResponse(input_blob, output_blob, _INPUT_MAP, _OUTPUT_MAP)229 obj2 = _execution.TaskExecutionGetDataResponse.from_flyte_idl(obj.to_flyte_idl())230 assert obj == obj2231 assert obj2.inputs == input_blob232 assert obj2.outputs == output_blob233 assert obj2.full_inputs == _INPUT_MAP234 assert obj2.full_outputs == _OUTPUT_MAP235def test_abort_metadata():236 obj = _execution.AbortMetadata(cause="cause", principal="skinner")237 assert obj.cause == "cause"238 assert obj.principal == "skinner"239 obj2 = _execution.AbortMetadata.from_flyte_idl(obj.to_flyte_idl())240 assert obj == obj2241 assert obj2.cause == "cause"...

Full Screen

Full Screen

test_unittest.py

Source:test_unittest.py Github

copy

Full Screen

1from date import Date, TimeDelta2import unittest3class MyTestCase(unittest.TestCase):4 def test_init(self):5 for i in (-1, 1, 32):6 date = i7 for j in (-1, 1, 13):8 month = j9 for k in (0, 1):10 if i == 1 and j == 1 and k == 1:11 pass12 else:13 year = k14 with self.assertRaises(ValueError):15 some_date = Date(date, month, year)16 date1, month1, year1 = 1, 1, 117 with self.assertRaises(ValueError):18 Date(date1, month1)19 with self.assertRaises(ValueError):20 Date("2.2")21 self.assertEqual(str(Date("2.2.2")), "02.02.0002")22 def test_repr(self):23 date = 124 month = 125 year = 126 test_date = Date(date, month, year)27 test_timedelta = TimeDelta(date, month, year)28 self.assertEqual(test_date.__repr__(), f"Date({date}, {month}, {year})", "Should be 'Date(1, 1, 1)'")29 self.assertEqual(test_timedelta.__repr__(), f"TimeDelta({date},"30 f" {month}, {year})", "Should be 'TimeDelta(1, 1, 1)'")31 def test_str(self):32 day, month, year = 1, 1, 133 test_date = Date(day, month, year)34 test_timedelta = TimeDelta(day, month, year)35 self.assertEqual(str(test_date), "01.01.0001")36 self.assertEqual(str(test_timedelta), f'{day} day(s), {month} month(s), {year} year(s)')37 day, month, year = 11, 1, 138 test_date = Date(day, month, year)39 test_timedelta = TimeDelta(day, month, year)40 self.assertEqual(str(test_date), "11.01.0001")41 self.assertEqual(str(test_timedelta), f'{day} day(s), {month} month(s), {year} year(s)')42 day, month, year = 11, 11, 143 test_date = Date(day, month, year)44 test_timedelta = TimeDelta(day, month, year)45 self.assertEqual(str(test_date), "11.11.0001")46 self.assertEqual(str(test_timedelta), f'{day} day(s), {month} month(s), {year} year(s)')47 day, month, year = 11, 11, 1148 test_date = Date(day, month, year)49 test_timedelta = TimeDelta(day, month, year)50 self.assertEqual(str(test_date), "11.11.0011")51 self.assertEqual(str(test_timedelta), f'{day} day(s), {month} month(s), {year} year(s)')52 year = 11153 test_date = Date(day, month, year)54 test_timedelta = TimeDelta(day, month, year)55 self.assertEqual(str(test_date), "11.11.0111")56 self.assertEqual(str(test_timedelta), f'{day} day(s), {month} month(s), {year} year(s)')57 year = 111158 test_date = Date(day, month, year)59 test_timedelta = TimeDelta(day, month, year)60 self.assertEqual(str(test_date), "11.11.1111")61 self.assertEqual(str(test_timedelta), f'{day} day(s), {month} month(s), {year} year(s)')62 def test_is_leap_year(self):63 answers = []64 for year in (1, 100, 400, 1000, 2001):65 test_date = Date(11, 11, year)66 answers.append(test_date.is_leap_year(year))67 self.assertEqual(answers, [False, False, True, False, False])68 with self.assertRaises(ValueError):69 test_date.is_leap_year("f")70 def test_get_max_days(self):71 j = 172 for i in (31, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31):73 test_date = Date(11, j, 400)74 self.assertEqual(test_date.get_max_day(test_date.month, test_date.year), i)75 j += 176 j = 177 for i in (31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31):78 test_date = Date(11, j, 401)79 self.assertEqual(test_date.get_max_day(test_date.month, test_date.year), i)80 j += 181 def test_is_valid_date(self):82 with self.assertRaises(ValueError):83 Date(29, 2, 401)84 def test_setters(self):85 test_date = Date(11, 11, 1111)86 test_date.day = 1287 test_date.month = 1288 test_date.year = 111289 test_date = Date(1, 1, 1)90 def test_setters_incorrect(self):91 test_date = Date(1, 1, 1)92 with self.assertRaises(ValueError):93 test_date.day = 3294 with self.assertRaises(ValueError):95 test_date.month = 1396 with self.assertRaises(ValueError):97 test_date.year = 098 def test_sub(self):99 test_date1 = Date(11, 11, 1111)100 test_date2 = Date(12, 12, 1112)101 self.assertEqual(test_date1 - test_date2, -398)102 test_date2 = Date(12, 12, 400)103 self.assertEqual(test_date1 - test_date2, 259654)104 def test_sub_incorrect(self):105 test_date1 = Date(11, 11, 1111)106 test_date2 = "OLOLO"107 res = test_date1.__sub__(test_date2)108 self.assertEqual(res, NotImplemented)109 def test_add(self):110 test_date = Date(11, 11, 1111)111 timedelta = TimeDelta(1, 1, 1)112 summ = test_date + timedelta113 self.assertEqual(str(summ), "12.12.1112")114 test_date = Date(28, 2, 400)115 timedelta = TimeDelta(2, 11, 400)116 summ = test_date + timedelta117 self.assertEqual(str(summ), "01.02.0801")118 test_date = Date(1, 10, 1)119 timedelta = TimeDelta(30, 1, 0)120 summ = test_date + timedelta121 self.assertEqual(str(summ), "01.12.0001")122 def test_iadd(self):123 test_date = Date(11, 11, 1111)124 timedelta = TimeDelta(1, 1, 1)125 test_date += timedelta126 self.assertEqual(str(test_date), "12.12.1112")127 test_date = Date(28, 2, 400)128 timedelta = TimeDelta(2, 9, 400)129 test_date += timedelta130 self.assertEqual(str(test_date), "01.12.0800")131 test_date = Date(31, 12, 400)132 timedelta = TimeDelta(1, 0, 0)133 test_date += timedelta134 self.assertEqual(str(test_date), "01.01.0401")135if __name__ == '__main__':...

Full Screen

Full Screen

test_date.py

Source:test_date.py Github

copy

Full Screen

1import pytest2from date import Date, TimeDelta3@pytest.mark.parametrize("day, month, year", [4 (-1, -1, -1),5 (1, -1, -1),6 (1, 1, -1)7])8def test_init_date_incorrect(day, month, year):9 with pytest.raises(ValueError):10 Date(day, month, year)11 with pytest.raises(ValueError):12 Date("30.02.1")13@pytest.mark.parametrize("day, month, year", [14 (-1, -1, -1),15 (1, -1, -1),16 (1, 1, -1),17])18def test_init_timedelta(day, month, year):19 test_timedelta = TimeDelta(day, month, year)20 assert test_timedelta.days == day21 assert test_timedelta.months == month22 assert test_timedelta.years == year23def test_init_timedelta_incorrect():24 with pytest.raises(ValueError):25 test_timedelta = TimeDelta(None, None, None)26def test_repr_str_timedelta():27 test_timedelta = TimeDelta(1, 1, 1)28 assert repr(test_timedelta) == "TimeDelta(1, 1, 1)"29 assert str(test_timedelta) == "1 day(s), 1 month(s), 1 year(s)"30def test_correct_str_date():31 assert str(Date("3.3.3")) == "03.03.0003"32def test_incorrect_date():33 with pytest.raises(ValueError):34 Date("3.3")35 with pytest.raises(ValueError):36 Date(3, 3)37def test_str_repr_date():38 test_date = Date(1, 1, 1)39 assert repr(test_date) == "Date(1, 1, 1)"40 assert str(test_date) == "01.01.0001"41@pytest.mark.parametrize("year", [1, 100, 1000, 2001])42def test_is_not_leap(year):43 test_date = Date(11, 2, year)44 assert test_date.is_leap_year(test_date.year) == False45@pytest.mark.parametrize("year", [4, 400, 2000, 16])46def test_is_leap(year):47 test_date = Date(11, 2, year)48 assert test_date.is_leap_year(test_date.year) == True49@pytest.mark.parametrize("day", [0, 32])50def test_is_valid(day):51 test_date = Date(1, 1, 1)52 with pytest.raises(ValueError):53 test_date.is_valid_date(day, 1, 1)54def test_day_setter():55 test_date = Date(1, 1, 1)56 test_date.day = 457 with pytest.raises(ValueError):58 test_date.day = 3359def test_month_setter():60 test_date = Date(1, 1, 1)61 test_date.month = 462 with pytest.raises(ValueError):63 test_date.month = 3364def test_year_setter():65 test_date = Date(1, 1, 1)66 test_date.year = 467 with pytest.raises(ValueError):68 test_date.year = 069@pytest.mark.parametrize("date1, date2, expected", [70 (Date(11, 11, 1111), Date(12, 11, 1111), -1),71 (Date(12, 11, 1111), Date(12, 11, 1111), 0),72 (Date(12, 12, 1111), Date(12, 11, 1111), 31),73 (Date(3, 1, 2222), Date(12, 11, 1111), 405473),74 (Date(1, 1, 1), Date(1, 1, 2), -365),75 (Date(1, 1, 4), Date(1, 1, 5), -366),76])77def test_sub(date1, date2, expected):78 assert date1 - date2 == expected79def test_some_test():80 test_date1 = Date(11, 11, 1111)81 test_date2 = "OLOLO"82 res = test_date1.__sub__(test_date2)83 assert res == NotImplemented84@pytest.mark.parametrize("date1, date2, expected", [85 (Date(11, 11, 1111), TimeDelta(2000, 1, 1), "03.06.1118"),86 (Date(28, 2, 400), TimeDelta(2, 11, 400), "01.02.0801"),87 (Date(1, 10, 1), TimeDelta(30, 1, 0), "01.12.0001"),88])89def test_add(date1, date2, expected):90 assert str(date1 + date2) == expected91@pytest.mark.parametrize("date1, date2, expected", [92 (Date(11, 11, 1111), TimeDelta(1, 1, 1), "12.12.1112"),93 (Date(28, 2, 400), TimeDelta(2, 9, 400), "01.12.0800"),94 (Date(31, 12, 400), TimeDelta(31, 12, 400), "31.01.0802"),95])96def test_iadd(date1, date2, expected):97 date1 += date2...

Full Screen

Full Screen

Automation Testing Tutorials

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run pandera automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful