Adapt unit tests

tests/data/clientkey.pem

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDRLCtJpkXjq5G1
+32sQ2Q0cWguEosNGG6/sYjJlHItAn+STItmf8o3fHTo9eZc+uTCpEYiazqSVOIsA
+c5xGW7EOcPjDn9Vsl6Klnd76R/C12mr4aEUkra2cNsdxJhyu51Pj4ifkZMpoJJCI
+Nm02Tx66rWWGHG/bUX+m/hnsIRGfaJ8C4NbhLwSRmkyuxn8E2sPDvydx4RVfTYfj
+XZCE1aJoNIzJ6towk9BEsI+ACOo3pybJj+wAO/kZg3tpr3i8HEZFE7aZvPJ6KBWx
+UVWJSZVPJ1POdGPOLmiDXtGUwJvUcAbw0WHsJIh7nP6mbvyaYNiYGAyVlIFF8PJ+
+cYiXYFxDAgMBAAECggEACIp5KgS0DdvPk1GQGZwDQQkcN0o+FvrCcDCCRkaoGPdu
+aeOeZz2MNXQIEMKKPnFpXz3sCgYuCjnI0zflRaowzJooTjSUjl6SsZiEpkuRtJs6
+ivIXAKxmzl4ePqyDt1CSyFdPsa+75Ay6KhWu1+zbIFw8LQG0P4xQXg06Gb4v7gZQ
+qvmKRKLC/7hcV6X6+s2bOuM6mB7EL319aUklPMoJH8hsWT9Fw8voeyUz5VXLUTs1
+JS6TP0FoXEbtg/G06oPzTomLWe+7Aq+g6PG/HflzcjqvJ30wrAdVXNiWfgkjgDxl
+7FBk5GcPe08i5jMoUdxtCfer+U1lBGiLz+1MsPUU4QKBgQD0Qa1otFrMj1IZfrA6
+ofZ0O5JTvPygQuu+2bLPSYH9c+hJCPf50JA/w5n/nxkFIr0yFhm+EnHSrMrgpglf
+A2Sj+sAKvmajlIYgvjKmXxUpWX5FZ2nWh/o43a7jPHT/mee59V1tTsgCoamhGMy1
+n1XHVvRVV1RjCQu8DEC7PP2VBQKBgQDbOqz/YMW2bqLichCokYU3YIacSHgin3ls
+jQ7FQYTTFQ8GwDg7iZBbZ6vkNDIIYTRY671QMXU8dTUuMk/rZfrO0trz2L8RDrUg
+5p7epDgcx4mLtRIwjcLJE7T4Cwh/Zchhk+mzGn75MC53ckSqwfSYfMQ/YI8bYwMK
+AxzYO/xupwKBgQCceIJc48SS7HEckfLU7LJTzWG0sQlopNYegZkxfxZ9xcWVG+C3
+MOtnXaeGgGXny1RGBLBi+a/e1QB2Hwc7zZGoFlb1yvk3a0rtKMqLl7eXsJPaZCAB
+5UPfL+v799u/bdlrYAqEnoY0YVmoMJna84Jg24xUK0iM1NumkHbbO3v6/QKBgCsJ
+z+Cq3OW9vph6EC8nsmF2v7Z6u6sAZ9QZtSaggDT4U2Td46w8i2yGY8Z/QLtIagBy
+902BCCUAVZpmIi5ybNShrH5mtMvieUimPdYzoxhzS9tzhsila/IRvltbvyVTlA0j
+/qM8tmSxQs4MTtK/FQfCprxSdoXH4Fbc2ZLR4/LNAoGAA9OlHzTdXpAKQgXZG0Pf
+N9NWdvIWWuKB4l3U2e6CU7jUmT2GH3lknMecZo95sjgI2dZ3pqF0swslKtXUvfgF
++hNbLaLzkRLDls/Q/tQoIyM82LMWFI3rMQPXbbL1dSrVErUft7FwVwco+2Civu0z
+CAn6r0KHHHiBb4Scgk6NfoI=
+-----END PRIVATE KEY-----

tests/data/clientreq.pem

@@ -0,0 +1,22 @@
+-----BEGIN CERTIFICATE-----
+MIIDjzCCAnegAwIBAgIUHA9xE40tm3defw2dP7AaSfiJD18wDQYJKoZIhvcNAQEL
+BQAwVzELMAkGA1UEBhMCREUxDzANBgNVBAgMBkJlcmxpbjEPMA0GA1UEBwwGQmVy
+bGluMRAwDgYDVQQKDAdFeGFtcGxlMRQwEgYDVQQDDAtleGFtcGxlLmNvbTAeFw0y
+NDEyMTQxNzE1MDdaFw0yNTAxMTMxNzE1MDdaMFcxCzAJBgNVBAYTAkRFMQ8wDQYD
+VQQIDAZCZXJsaW4xDzANBgNVBAcMBkJlcmxpbjEQMA4GA1UECgwHRXhhbXBsZTEU
+MBIGA1UEAwwLZXhhbXBsZS5jb20wggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEK
+AoIBAQDRLCtJpkXjq5G132sQ2Q0cWguEosNGG6/sYjJlHItAn+STItmf8o3fHTo9
+eZc+uTCpEYiazqSVOIsAc5xGW7EOcPjDn9Vsl6Klnd76R/C12mr4aEUkra2cNsdx
+Jhyu51Pj4ifkZMpoJJCINm02Tx66rWWGHG/bUX+m/hnsIRGfaJ8C4NbhLwSRmkyu
+xn8E2sPDvydx4RVfTYfjXZCE1aJoNIzJ6towk9BEsI+ACOo3pybJj+wAO/kZg3tp
+r3i8HEZFE7aZvPJ6KBWxUVWJSZVPJ1POdGPOLmiDXtGUwJvUcAbw0WHsJIh7nP6m
+bvyaYNiYGAyVlIFF8PJ+cYiXYFxDAgMBAAGjUzBRMB0GA1UdDgQWBBQc6QMw+fKF
+zVTfSTgvORkkEr6+OzAfBgNVHSMEGDAWgBQc6QMw+fKFzVTfSTgvORkkEr6+OzAP
+BgNVHRMBAf8EBTADAQH/MA0GCSqGSIb3DQEBCwUAA4IBAQC7BroYEqjM+zZGbSwl
+IvuclDQDFRmRQRA7RKiicptJfUDxmOzdk/COh2wFZ1+mCVNamwexLdSETooDd/9s
+Qm0LaR3KzpUer+6WYq2y8ijxT2wzsr/Yejq8v1GkejezuedzbMvdihsDvJI2YUuo
+3yHlTWifLse7dn6Ru07P7hnxTfFrav/z6H5PtuXKJ76FUbqLZieRc6Eve5gwsSTc
+h9+DjpdVD8rz88PurLNFA1C72ZU384W/xnYhNICatQ26DiHO5cWbHPdwVhOenM/F
+Ow5qn/FcM3T5lOamgakdflcn/sFxXJjOj2aWnWRnYOp/zea3WCpYYIR59SSCa1EF
+xxDd
+-----END CERTIFICATE-----

tests/test_async_client.py

@@ -64,15 +64,16 @@ def test_namespace(client: lightkube.Client, kubeconfig_ns):
 
 @unittest.mock.patch('httpx.AsyncClient')
 @unittest.mock.patch('lightkube.config.client_adapter.user_auth')
-def test_client_httpx_attributes(user_auth, httpx_async_client, kubeconfig):
+@unittest.mock.patch('lightkube.config.client_adapter.verify_cluster')
+def test_client_httpx_attributes(verify_cluster, user_auth, httpx_async_client, kubeconfig):
     config = KubeConfig.from_file(kubeconfig)
     single_conf = config.get()
     lightkube.AsyncClient(config=single_conf, trust_env=False)
+    verify_cluster.assert_called_once_with(single_conf.cluster, single_conf.user, single_conf.abs_file, trust_env=False)
     httpx_async_client.assert_called_once_with(
         timeout=None,
         base_url=single_conf.cluster.server,
-        verify=True,
-        cert=None,
+        verify=verify_cluster.return_value,
         auth=user_auth.return_value,
         trust_env=False,
         transport=None,

tests/test_client.py

@@ -1,6 +1,7 @@
 from collections import namedtuple
 import unittest.mock
 import warnings
+from ssl import SSLContext
 
 warnings.filterwarnings("ignore", category=DeprecationWarning)
 
@@ -103,15 +104,16 @@ def test_client_default_config_construction(mock_kube_config):
 
 @unittest.mock.patch('httpx.Client')
 @unittest.mock.patch('lightkube.config.client_adapter.user_auth')
-def test_client_httpx_attributes(user_auth, httpx_client, kubeconfig):
+@unittest.mock.patch('lightkube.config.client_adapter.verify_cluster')
+def test_client_httpx_attributes(verify_cluster, user_auth, httpx_client, kubeconfig):
     config = KubeConfig.from_file(kubeconfig)
     single_conf = config.get()
     lightkube.Client(config=single_conf, trust_env=False)
+    verify_cluster.assert_called_once_with(single_conf.cluster, single_conf.user, single_conf.abs_file, trust_env=False)
     httpx_client.assert_called_once_with(
         timeout=None,
         base_url=single_conf.cluster.server,
-        verify=True,
-        cert=None,
+        verify=verify_cluster.return_value,
         auth=user_auth.return_value,
         trust_env=False,
         transport=None,

tests/test_client_adapter.py

@@ -1,3 +1,7 @@
+import base64
+import shutil
+import ssl
+import unittest
 from pathlib import Path
 from unittest.mock import Mock
 from lightkube.config import kubeconfig, client_adapter
@@ -18,40 +22,49 @@ def single_conf(cluster=None, user=None, fname=None):
 
 
 def test_verify_cluster_insecure():
-    cfg = single_conf(cluster=models.Cluster(insecure=True))
-    verify = client_adapter.verify_cluster(cfg.cluster, cfg.abs_file)
-    assert verify is False
+    cfg = single_conf(cluster=models.Cluster(insecure=True), user=models.User())
+    verify = client_adapter.verify_cluster(cfg.cluster, cfg.user, cfg.abs_file)
+    assert verify.verify_mode is ssl.CERT_NONE
+    assert not verify.check_hostname
 
 
 def test_verify_cluster_secure():
-    cfg = single_conf(cluster=models.Cluster())
-    verify = client_adapter.verify_cluster(cfg.cluster, cfg.abs_file)
-    assert verify is True
+    cfg = single_conf(cluster=models.Cluster(), user=models.User())
+    verify = client_adapter.verify_cluster(cfg.cluster, cfg.user, cfg.abs_file)
+    assert verify.verify_mode is ssl.CERT_REQUIRED
 
 
-def test_verify_cluster_ca(tmpdir):
+def get_issuer_mata(data: dict):
+    return {d[0][0]: d[0][1] for d in data['issuer']}
+
+def test_verify_cluster_ca_path(tmpdir):
     tmpdir = Path(tmpdir)
-    cluster = models.Cluster(certificate_auth="ca.pem")
-    cfg = single_conf(cluster=cluster, fname=tmpdir.joinpath("kubeconf"))
-    verify = client_adapter.verify_cluster(cfg.cluster, cfg.abs_file)
-    assert verify == tmpdir.joinpath("ca.pem")
+    data_dir = Path(__file__).parent.joinpath('data')
+    shutil.copy(data_dir.joinpath("clientreq.pem"), tmpdir.joinpath("clientreq.pem"))
+    cluster = models.Cluster(certificate_auth="clientreq.pem")
+    cfg = single_conf(cluster=cluster, user=models.User(), fname=tmpdir.joinpath("kubeconf"))
+    verify = client_adapter.verify_cluster(cfg.cluster, cfg.user, cfg.abs_file)
+    assert get_issuer_mata(verify.get_ca_certs()[0])["organizationName"] == "Example"
 
     # fname not provided
-    cfg = single_conf(cluster=cluster)
+    cfg = single_conf(cluster=models.Cluster(certificate_auth="clientreq.pem"), user=models.User())
     with pytest.raises(ConfigError):
-        client_adapter.verify_cluster(cfg.cluster, cfg.abs_file)
+        client_adapter.verify_cluster(cfg.cluster, cfg.user, cfg.abs_file)
 
     # cert path absolute
-    cluster.certificate_auth = tmpdir.joinpath("ca.pem")
-    verify = client_adapter.verify_cluster(cfg.cluster, cfg.abs_file)
-    assert verify == tmpdir.joinpath("ca.pem")
+    cluster = models.Cluster(certificate_auth=str(data_dir.joinpath("clientreq.pem")))
+    verify = client_adapter.verify_cluster(cluster, cfg.user, cfg.abs_file)
+    assert get_issuer_mata(verify.get_ca_certs()[0])["organizationName"] == "Example"
 
 
 def test_verify_cluster_ca_data():
-    cluster = models.Cluster(certificate_auth_data="dGVzdCBkZWNvZGluZw==")
-    cfg = single_conf(cluster=cluster)
-    verify = client_adapter.verify_cluster(cfg.cluster, cfg.abs_file)
-    assert Path(verify).read_text() == "test decoding"
+    data_dir = Path(__file__).parent.joinpath('data')
+    cert_data = base64.b64encode(data_dir.joinpath("clientreq.pem").read_bytes()).decode("utf8")
+
+    cluster = models.Cluster(certificate_auth_data=cert_data)
+    cfg = single_conf(cluster=cluster, user=models.User())
+    verify = client_adapter.verify_cluster(cfg.cluster, cfg.user, cfg.abs_file)
+    assert get_issuer_mata(verify.get_ca_certs()[0])["organizationName"] == "Example"
 
 
 def test_user_cert_missing():
@@ -73,6 +86,23 @@ def test_user_cert_data():
     assert Path(certs[1]).read_text() == "key"
 
 
+@unittest.mock.patch('ssl.create_default_context')
+def test_verify_cluster_ca_and_cert(create_default_context):
+    data_dir = Path(__file__).parent.joinpath('data')
+    cluster = models.Cluster(certificate_auth=str(data_dir.joinpath("clientreq.pem")))
+    cfg = single_conf(cluster=cluster, user=models.User(
+        client_cert=str(data_dir.joinpath("clientreq.pem")),
+        client_key=str(data_dir.joinpath("clientkey.pem"))
+    ))
+    verify = client_adapter.verify_cluster(cluster, cfg.user, cfg.abs_file)
+    assert verify is create_default_context.return_value
+    create_default_context.assert_called_once_with(cafile=str(data_dir.joinpath("clientreq.pem")))
+    create_default_context.return_value.load_cert_chain.assert_called_once_with(
+        str(data_dir.joinpath("clientreq.pem")),
+        str(data_dir.joinpath("clientkey.pem"))
+    )
+
+
 def test_user_auth_missing():
     assert client_adapter.user_auth(None) is None