Use env var connections instead of mocks in kube system tests

[amoghrajesh]

Why?

While working on #54083, I happened to notice that the kube system tests started failing with:

../task-sdk/src/airflow/sdk/bases/operator.py:403: in wrapper
    return func(self, *args, **kwargs)
../providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:642: in execute
    return self.execute_sync(context)
../providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:650: in execute_sync
    self.pod_request_obj = self.build_pod_request_obj(context)
../providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:1289: in build_pod_request_obj
    "airflow_kpo_in_cluster": str(self.hook.is_in_cluster),
../providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:307: in is_in_cluster
    self.api_client  # so we can determine if we are in_cluster or not
/opt/hostedtoolcache/Python/3.10.18/x64/lib/python3.10/functools.py:981: in __get__
    val = self.func(instance)
../providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:315: in api_client
    return self.get_conn()
../providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:215: in get_conn
    in_cluster = self._coalesce_param(self.in_cluster, self._get_field("in_cluster"))
../providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:208: in _get_field
    if field_name in self.conn_extras:
/opt/hostedtoolcache/Python/3.10.18/x64/lib/python3.10/functools.py:981: in __get__
    val = self.func(instance)
../providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:189: in conn_extras
    connection = self.get_connection(self.conn_id)
../providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py:180: in get_connection
    return super().get_connection(conn_id)  # type: ignore[return-value]
../task-sdk/src/airflow/sdk/bases/hook.py:61: in get_connection
    conn = Connection.get(conn_id)
../task-sdk/src/airflow/sdk/definitions/connection.py:154: in get
    return _get_connection(conn_id)

The reason for this is that those tests used a mock on get_connection which is mid-way to the entire flow of getting a connection (models connection had an internal get_connection defined on connection backends and this test was bypassing it). The current mock_get_connection fixture uses mock.patch to return a minimal Connection object with only conn_id set.

A better OR more complete pattern would be to actual create the connection using env vars on the host machine as part of the test setup that runs, that would ensure that no unneccessary mocking is performed and Connection retrieval flow works normally (backend > env > api server)

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[amoghrajesh]

@potiuk I think you will like this one :)

[jason810496]

Nice cleanup!

[potiuk]

nice indeed :)

[amoghrajesh]

Ah, had to include the pytest plugin.

[gopidesupavan]

cool :) thanks for updating

[amoghrajesh]

@potiuk getting a clarification here, these tests are separate distros out of providers but use provider code, so do we cherry pick these to v3-0-x branches?

[potiuk]

the kubetnetes_tests are more than provider tests in general - they run a complete e2e tests with "airflow" installed with PROD image using "helm chart" - which means we are testing the whole deployment

However - some of the "kubernetes_test" are only using that K8S setup to run "KPO" on running k8s instance (and the fact that that instance was installed to run airflow - does not matter"

To be on the safe-side - I'd cherry-pick, also because it makes it easier to cherry-pick changes that should for sure be tested on v3-0-test.

[amoghrajesh]

the kubetnetes_tests are more than provider tests in general - they run a complete e2e tests with "airflow" installed with PROD image using "helm chart" - which means we are testing the whole deployment

However - some of the "kubernetes_test" are only using that K8S setup to run "KPO" on running k8s instance (and the fact that that instance was installed to run airflow - does not matter"

To be on the safe-side - I'd cherry-pick, also because it makes it easier to cherry-pick changes that should for sure be tested on v3-0-test.

Yeah i had similar thoughts too regarding this. Let me manually do it

[amoghrajesh]

Heres the cherry pick #54320