fix(mypy): Pass parameters to Kubernetes methods conditionally (apache#64242)

* Pass parameters to k8s methods conditionally to fix mypy

Author: amoghrajesh

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py

@@ -1040,15 +1040,19 @@ async def read_logs(
                 # (e.g. binary data, truncated multi-byte sequences).
                 # kubernetes_asyncio's default decoding uses strict UTF-8 which
                 # crashes the task in those cases.
-                raw_resp: ClientResponse = await v1_api.read_namespaced_pod_log(
-                    name=name,
-                    namespace=namespace,
-                    container=container_name,
-                    follow=False,
-                    timestamps=True,
-                    since_seconds=since_seconds,
-                    _preload_content=False,
-                )  # type: ignore  # _preload_content=False makes returning ClientResponse instead of str!
+                kwargs = {
+                    "name": name,
+                    "namespace": namespace,
+                    "follow": False,
+                    "timestamps": True,
+                    "_preload_content": False,
+                }
+                if container_name is not None:
+                    kwargs["container"] = container_name
+                if since_seconds is not None:
+                    kwargs["since_seconds"] = since_seconds
+
+                raw_resp: ClientResponse = await v1_api.read_namespaced_pod_log(**kwargs)  # type: ignore  # _preload_content=False makes returning ClientResponse instead of str!
                 raw_bytes = await raw_resp.read()
                 logs = raw_bytes.decode("utf-8", errors="replace")
                 logs_list: list[str] = logs.splitlines()
@@ -1070,12 +1074,15 @@ async def get_pod_events(
         async with self.get_conn() as connection:
             try:
                 v1_api = async_client.CoreV1Api(connection)
-                events: CoreV1EventList = await v1_api.list_namespaced_event(
-                    field_selector=f"involvedObject.name={name}",
-                    namespace=namespace,
-                    resource_version=resource_version,
-                    resource_version_match="NotOlderThan" if resource_version else None,
-                )
+                kwargs = {
+                    "field_selector": f"involvedObject.name={name}",
+                    "namespace": namespace,
+                }
+                if resource_version is not None:
+                    kwargs["resource_version"] = resource_version
+                    kwargs["resource_version_match"] = "NotOlderThan"
+
+                events: CoreV1EventList = await v1_api.list_namespaced_event(**kwargs)
                 return events
             except HTTPError as e:
                 if hasattr(e, "status") and e.status == 403:

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py

@@ -1186,8 +1186,6 @@ async def test_async_get_pod_events_without_resource_version(
         mock_list_namespaced_event.assert_called_once_with(
             field_selector=f"involvedObject.name={POD_NAME}",
             namespace=NAMESPACE,
-            resource_version=None,
-            resource_version_match=None,
         )
         assert result == mock_events
 

providers/google/tests/unit/google/cloud/hooks/test_kubernetes_engine.py

@@ -538,13 +538,11 @@ async def test_read_logs(self, read_namespaced_pod_log, get_conn_mock, async_hoo
 
         get_conn_mock.assert_called_once_with()
         read_namespaced_pod_log.assert_called_with(
-            _preload_content=False,
             name=POD_NAME,
             namespace=POD_NAMESPACE,
             follow=False,
             timestamps=True,
-            container=None,
-            since_seconds=None,
+            _preload_content=False,
         )
         assert "Test string #1" in logs
         assert "Test string #2" in logs