Repository: apache/airflow
File: providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py (Python)

Code before:
    )
    @patch(KUB_OP_PATH.format("patch_already_checked"))
    @patch(KUB_OP_PATH.format("process_pod_deletion"))
    def test_process_duplicate_label_pods_with_no_start_time_or_creation_timestamp(
        self,
        process_pod_deletion_mock,
        patch_already_checked_mock,
        on_finish_action,
    ):
        k = KubernetesPodOperator(
            namespace="default",
            image="ubuntu:22.04",
            cmds=["bash", "-cx"],
            arguments=["echo 12"],
            name="test",
            task_id="task",
            do_xcom_push=False,
            reattach_on_restart=False,
            on_finish_action=on_finish_action,
        )
        context = create_context(k)
        pod_1 = k.get_or_create_pod(pod_request_obj=k.build_pod_request_obj(context), context=context)
        pod_2 = k.get_or_create_pod(pod_request_obj=k.build_pod_request_obj(context), context=context)

        pod_1.status = {"start_time": None}
        pod_2.status = {"start_time": None}
        pod_1.metadata.creation_timestamp = None
        pod_2.metadata.creation_timestamp = None
        pod_2.metadata.labels.update({"try_number": "2"})

        result = k.process_duplicate_label_pods([pod_1, pod_2])

        patch_already_checked_mock.assert_called_once_with(pod_1, reraise=False)
        process_pod_deletion_mock.assert_not_called()
        assert result.metadata.name == pod_2.metadata.name

    @patch(KUB_OP_PATH.format("patch_already_checked"))
    @patch(KUB_OP_PATH.format("process_pod_deletion"))
    def test_process_duplicate_label_pods__pod_removed_if_delete_pod(
        self, process_pod_deletion_mock, patch_already_checked_mock
    ):
        now = datetime.datetime.now()
        k = KubernetesPodOperator(
            namespace="default",
            image="ubuntu:22.04",
            cmds=["bash", "-cx"],
            arguments=["echo 12"],
            name="test",
            task_id="task",
            do_xcom_push=False,
            reattach_on_restart=False,

Review comment (suggestion):
Setting `pod.status` to a plain dictionary may not correctly mock the V1Pod object's status attribute. The kubernetes client's V1Pod expects a V1PodStatus object. Consider using `V1PodStatus(start_time=None)` instead to ensure the test accurately represents the actual API behavior when serialized with `to_dict()`.
```suggestion
        pod_1.status = V1PodStatus(start_time=None)
        pod_2.status = V1PodStatus(start_time=None)
```

Code after:
        pod_1 = k.get_or_create_pod(pod_request_obj=k.build_pod_request_obj(context), context=context)
        pod_2 = k.get_or_create_pod(pod_request_obj=k.build_pod_request_obj(context), context=context)

        pod_1.status = {"start_time": now}
        pod_2.status = {"start_time": now + datetime.timedelta(seconds=60)}
        pod_2.metadata.labels.update({"try_number": "2"})

        result = k.process_duplicate_label_pods([pod_1, pod_2])

        patch_already_checked_mock.assert_called_once_with(pod_1, reraise=False)
        process_pod_deletion_mock.assert_not_called()
        assert result.metadata.name == pod_2.metadata.name

    @pytest.mark.parametrize(
        "on_finish_action", [OnFinishAction.KEEP_POD, OnFinishAction.DELETE_SUCCEEDED_POD]
    )
    @patch(KUB_OP_PATH.format("patch_already_checked"))
    @patch(KUB_OP_PATH.format("process_pod_deletion"))
    def test_process_duplicate_label_pods_with_start_time_none(
        self,
        process_pod_deletion_mock,
        patch_already_checked_mock,
        on_finish_action,
    ):
        now = datetime.datetime.now()
        k = KubernetesPodOperator(
            namespace="default",
            image="ubuntu:22.04",
            cmds=["bash", "-cx"],
            arguments=["echo 12"],
            name="test",
            task_id="task",
            do_xcom_push=False,
            reattach_on_restart=False,
            on_finish_action=on_finish_action,
        )
        context = create_context(k)
        pod_1 = k.get_or_create_pod(pod_request_obj=k.build_pod_request_obj(context), context=context)
        pod_2 = k.get_or_create_pod(pod_request_obj=k.build_pod_request_obj(context), context=context)

        pod_1.status = {"start_time": None}
        pod_2.status = {"start_time": None}
        pod_1.metadata.creation_timestamp = now
        pod_2.metadata.creation_timestamp = now + datetime.timedelta(seconds=60)
        pod_2.metadata.labels.update({"try_number": "2"})

        result = k.process_duplicate_label_pods([pod_1, pod_2])

        patch_already_checked_mock.assert_called_once_with(pod_1, reraise=False)
        process_pod_deletion_mock.assert_not_called()
        assert result.metadata.name == pod_2.metadata.name