How to use test_timedelta method in pandera

Best Python code snippet using pandera_python

test_execution.py

Source:test_execution.py

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"...

test_unittest.py

Source:test_unittest.py

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__':...

test_date.py

Source:test_date.py

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...

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.