test lightkube trust_env=False

tests/integration/ha_tests/helpers.py

@@ -284,7 +284,7 @@ async def count_writes(
 
             # Translate the service hostname to an IP address.
             model = ops_test.model.info
-            client = Client(namespace=model.name)
+            client = Client(namespace=model.name, trust_env=False)
             service = client.get(Pod, name=host.split(".")[0])
             ip = service.status.podIP
 
@@ -349,7 +349,7 @@ async def fetch_cluster_members(ops_test: OpsTest):
     def get_host_ip(host: str) -> str:
         # Translate the pod hostname to an IP address.
         model = ops_test.model.info
-        client = Client(namespace=model.name)
+        client = Client(namespace=model.name, trust_env=False)
         pod = client.get(Pod, name=host.split(".")[0])
         return pod.status.podIP
 

tests/integration/ha_tests/test_replication.py

@@ -64,7 +64,7 @@ async def test_reelection(ops_test: OpsTest, continuous_writes, primary_start_ti
 
     # Kill the current primary.
     primary_name = await get_primary(ops_test, app)
-    client = Client(namespace=ops_test.model.info.name)
+    client = Client(namespace=ops_test.model.info.name, trust_env=False)
     client.delete(Pod, name=primary_name.replace("/", "-"))
 
     # Wait and get the primary again (which can be any unit, including the previous primary).

tests/integration/ha_tests/test_upgrade.py

@@ -81,7 +81,7 @@ async def test_pre_upgrade_check(ops_test: OpsTest) -> None:
             assert primary_name == f"{DATABASE_APP_NAME}/0", "Primary unit not set to unit 0"
 
     logger.info("Assert partition is set to 2")
-    client = Client()
+    client = Client(trust_env=False)
     stateful_set = client.get(
         res=StatefulSet, namespace=ops_test.model.info.name, name=DATABASE_APP_NAME
     )

tests/integration/ha_tests/test_upgrade_from_stable.py

@@ -79,7 +79,7 @@ async def test_pre_upgrade_check(ops_test: OpsTest) -> None:
             assert primary_name == f"{DATABASE_APP_NAME}/0", "Primary unit not set to unit 0"
 
     logger.info("Assert partition is set to 2")
-    client = Client()
+    client = Client(trust_env=False)
     stateful_set = client.get(
         res=StatefulSet, namespace=ops_test.model.info.name, name=DATABASE_APP_NAME
     )

tests/integration/helpers.py

@@ -365,7 +365,7 @@ def get_existing_k8s_resources(namespace: str, application: str) -> set:
         set of existing charm/Patroni specific k8s resources.
     """
     # Create a k8s API client instance.
-    client = Client(namespace=namespace)
+    client = Client(namespace=namespace, trust_env=False)
 
     # Retrieve the resources created by the charm and Patroni.
     resources = set()

tests/integration/new_relations/helpers.py

@@ -79,7 +79,7 @@ async def build_connection_string(
 
     # Translate the service hostname to an IP address.
     model = ops_test.model.info
-    client = AsyncClient(namespace=model.name)
+    client = AsyncClient(namespace=model.name, trust_env=False)
     service = await client.get(Service, name=host.split(".")[0])
     ip = service.spec.clusterIP
 

tests/integration/test_charm.py

@@ -64,7 +64,7 @@ async def test_application_created_required_resources(ops_test: OpsTest) -> None
 @pytest.mark.parametrize("unit_id", UNIT_IDS)
 async def test_labels_consistency_across_pods(ops_test: OpsTest, unit_id: int) -> None:
     model = ops_test.model.info
-    client = AsyncClient(namespace=model.name)
+    client = AsyncClient(namespace=model.name, trust_env=False)
     pod = await client.get(Pod, name=f"postgresql-k8s-{unit_id}")
     # Ensures that the correct kubernetes labels are set
     # (these ones guarantee the correct working of replication).
@@ -214,7 +214,7 @@ async def test_cluster_is_stable_after_leader_deletion(ops_test: OpsTest) -> Non
 
     # Delete the primary pod.
     model = ops_test.model.info
-    client = AsyncClient(namespace=model.name)
+    client = AsyncClient(namespace=model.name, trust_env=False)
     await client.delete(Pod, name=primary.replace("/", "-"))
     logger.info(f"deleted pod {primary}")
 
@@ -300,7 +300,7 @@ async def test_persist_data_through_failure(ops_test: OpsTest):
 
     # Cause a machine failure by killing a unit in k8s
     model = ops_test.model.info
-    client = AsyncClient(namespace=model.name)
+    client = AsyncClient(namespace=model.name, trust_env=False)
     await client.delete(Pod, name=primary.replace("/", "-"))
     logger.info("primary pod deleted")