content: import base64
  import ssl
  import textwrap
  
  import pytest
  
  from kopf.clients.auth import reauthenticated_request, vault_var, APIContext
  from kopf.structs.credentials import Vault, ConnectionInfo
  
  # These are Minikube's locally geenrated certificates (CN=minikubeCA).
  # They are not in any public use, and are regenerated regularly.
  SAMPLE_MINIKUBE_CA = textwrap.dedent('''
  -----BEGIN CERTIFICATE-----
  MIIC5zCCAc+gAwIBAgIBATANBgkqhkiG9w0BAQsFADAVMRMwEQYDVQQDEwptaW5p
  a3ViZUNBMB4XDTE5MDUyMTA5MTgzNloXDTI5MDUxOTA5MTgzNlowFTETMBEGA1UE
  AxMKbWluaWt1YmVDQTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBANNU
  9eyjlDfWZNSTXbdM9uebYseWWo6KeGGQ/OISrnVc6+AjuIv/fxtdCc8nVyLXBWu2
  dlDjBqOsG2WfY7m1RVwsF0L2G8pgNZJ4eOww/PyDZzzIcB911eWiry528YB2PZQu
  sN6sUItSZrHsin3dkcEZMUKcvVOY3FaNqXukCoMywZBO7QlLZasHhCCaanMFjxBx
  WiqB4gxcyTlGRBSoa49agSW2r45873xmJ+JglI/tNjeobGLynYwrDvRWmrhVOFAj
  QeiMr5lkzVO5cC2t84WdEihXVqFcZQUe0jfRHmmgpUxJRtiJMv7yudgnK2ALdhOY
  eDVtV1wIyWgLpF2lZk8CAwEAAaNCMEAwDgYDVR0PAQH/BAQDAgKkMB0GA1UdJQQW
  MBQGCCsGAQUFBwMCBggrBgEFBQcDATAPBgNVHRMBAf8EBTADAQH/MA0GCSqGSIb3
  DQEBCwUAA4IBAQBdsqvuvK+8RJ5xqwGkdpSAK1U2LrZ3Hm0MzXoEo8GH79F1yubv
  Ig1VRLHDIDY1d/fKrK4a7uulSFTFvpt6AGSB/225wJVBQUAALH1lPkTXq5TDi5jE
  NqoXk3d61qf9StUEc1YehL6ZgkSknNU7ksAe5Ht0lfJlSa3DmACkI4CZJ1F5cztk
  m2p3RZYkxizY2i/9P34f59F3XCNUSOW52aJgLhnMugEM0baOTHN0mcYZRGmrunT6
  fs/5eZq6ZrXBu0nIkEZkEWAM2WoqDGxMlUao5IOnf289HyBxJTFGte5tysg1sJF0
  JCH4VcilJllzUki594R1Yv8O5qtxkXXfXQNR
  -----END CERTIFICATE-----
  ''').strip()
  
  SAMPLE_MINIKUBE_CERT = textwrap.dedent('''
  -----BEGIN CERTIFICATE-----
  MIIDADCCAeigAwIBAgIBAjANBgkqhkiG9w0BAQsFADAVMRMwEQYDVQQDEwptaW5p
  a3ViZUNBMB4XDTE5MTExMzA5NDMzNFoXDTIwMTExMzA5NDMzNFowMTEXMBUGA1UE
  ChMOc3lzdGVtOm1hc3RlcnMxFjAUBgNVBAMTDW1pbmlrdWJlLXVzZXIwggEiMA0G
  CSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQCut0LegwHP4kJp9Uf89vjuslIMi6hv
  BiBNhSr8wCZ9uuFUN2dvBnCPXX/xvxYxpBKh0WUKX7sYfKdaMjxVr1ndnwu63e5A
  CW7919uRH3fhVV7rTntlO+rUeyHXNlSQue8oVlvO+8D+Qzlna02axt/5PwsPGD/G
  lw7Ti0f/LjmmqTB0T6yCsyOH90d7pQ0yuiOyDK072Ns1vTf6hkrkiNQaRhUjEtqm
  rMq/A4xpb7h+z6LWlRyv6/DBJsLmpDS99hqZbbj1U6IJ56r0JpQuq7CJxeE/F2t5
  I1i0k8TJ4CQRrHvXovl6wI+zmOTmKMy5uDdKUEkfe6vck+x16gV2RK79AgMBAAGj
  PzA9MA4GA1UdDwEB/wQEAwIFoDAdBgNVHSUEFjAUBggrBgEFBQcDAQYIKwYBBQUH
  AwIwDAYDVR0TAQH/BAIwADANBgkqhkiG9w0BAQsFAAOCAQEAbo/PxDz7K9dCAcxi
  S0RdZvOu2RYHqRmc/vN1B5JDHLcYcCet4iuJ44BuWZc/Oe1sXdy6DOZO2UQHAF0F
  5cAaBnkqpz/yvUzkE3NvdXpU/9leo3O6XeKKi0Fi+hv09nhh/tJgh/XDxWfAAkeG
  I3AQCkcHrqFrpBMxuWUXlnexwsvvbdEVkpVMVwSRfpsxLfxV62HCU90EU0823UKW
  V1npcRXBtxK/jqWZbLd5buRul+V6PKa4KRY22d8it+9MDAMgPFQosG1ShhQropul
  VJUvAQ14dPpiyQN4FRI3MljVykFe2cWI0rVwoboy9TEniaMPqr3MqujOlUv7KTpk
  lKRP5w==
  -----END CERTIFICATE-----
  ''').strip()
  
  SAMPLE_MINIKUBE_PKEY = textwrap.dedent('''
  -----BEGIN RSA PRIVATE KEY-----
  MIIEowIBAAKCAQEArrdC3oMBz+JCafVH/Pb47rJSDIuobwYgTYUq/MAmfbrhVDdn
  bwZwj11/8b8WMaQSodFlCl+7GHynWjI8Va9Z3Z8Lut3uQAlu/dfbkR934VVe6057
  ZTvq1Hsh1zZUkLnvKFZbzvvA/kM5Z2tNmsbf+T8LDxg/xpcO04tH/y45pqkwdE+s
  grMjh/dHe6UNMrojsgytO9jbNb03+oZK5IjUGkYVIxLapqzKvwOMaW+4fs+i1pUc
  r+vwwSbC5qQ0vfYamW249VOiCeeq9CaULquwicXhPxdreSNYtJPEyeAkEax716L5
  esCPs5jk5ijMubg3SlBJH3ur3JPsdeoFdkSu/QIDAQABAoIBAAk8FoS8V/Qs+WWw
  WUW9qBq1wjB3kUeNA1gVmdgSL/alUhOpegYcSQbK4mBcwUeObI2xC64osTLyI8ZY
  sWe2BQH5zhzqbhVkakFwj2J0T1nRsVquo0cOi7L/byJ49K9RpJp1NhUSqXjHBNm6
  ijeMG3qJIoSBu507jsUPr5aFUvbEFCby+VvU/UljS1dK+5wm7QcGgRrcXc8ZCIuk
  1P5YX6Tr9RYzYNjc/zB8czIyoISSSTk9uroYzvuMCgYTQ4WzWvX8UwchdfZyV7Gq
  kdKjG6IkGxvlS0Lc8534LRlH8iJR+wPAlYKHBDa0Rc3qZFsvgjnbDCCa6YqLoNNn
  ltsz7wECgYEA5IrnX8EHXyeaYnyiVI2xk6QTkmLrtovd1Ue2GwQ9BJgtaDrT0xil
  UV5NV4VUu5Zid7cqmIyFyh+7jjIex+dpfXT94+wr1HXxNYbLHeCnCZWcomxaD8pU
  Bh8B8wSRgjOFW33q6APDkFJPO92O96B+BczSgOFmEvuk8Kj7aYFahWUCgYEAw7Ta
  YiD66I+eK8B2+lWfPNfpIddW7D3Dn9cSW2RVazMyinTsVBm3p63kpCQeVaaX5key
  WiyY6phTvIfJ45pTzrS/kpA2zGcB1FFnB1xvM0bzpbIxTOQBGQUH1mSQmNT6+0VZ
  +GdILRKedp4qg7BLh7VElSCYGVy1Yr62Rp01FbkCgYB3QZxWtQ05tBq1hb/XS1D8
  b8PewUuqp/WL063NDzsf6KDZIMlkABpUCVdmciay9FhRi/zoOXue60wdeT3ipni/
  hIrvok+EwD6r5bib0JyZPb7MaqncT4Hk581GmH2taWEPSveHNl+YMbsyy/xMby0T
  rbuykOuIwFNjWWpHtb4cmQKBgGF4MUuuIUiyPpSLxrXm7ufeoL26AhCmskdpVjsu
  PVymowVSNmGsbUuVz8nwMyt1TTHjg3BlxcMRGqNK/cHdmt/YJZFZQfGLW93irO19
  m+Rt8esUVHl3FRTg7IZaj6mOaXG7mJOe3NOV8lYhcAsmQnfUT9P158q54ZzMXvvM
  UCQBAoGBAMTbVdbybvzqOSIKGeqIVlX7L2Fp55lYC8MfTwQzjPeJMQ8YdeNquv+M
  hweBrRg1DXxdaicfpCuqITU1WkqHd/NGNQX2h8VleiR6t22dZb8nBplO1l1e3XgY
  lUXVsCYgw8yNCm10xGCelpJ4nxxPhf5apbz4F3nGORGfsv5C+x++
  -----END RSA PRIVATE KEY-----
  ''').strip()
  
  
  @reauthenticated_request
  async def fn(context: APIContext):
      return context.session
  
  
  @pytest.fixture(autouse=True)
  def vault():
      vault = Vault()
      vault_var.set(vault)
      return vault
  
  
  @pytest.fixture
  def cabase64(tmpdir):
      return base64.encodebytes(SAMPLE_MINIKUBE_CA.encode('ascii'))
  
  
  @pytest.fixture
  def certbase64(tmpdir):
      return base64.encodebytes(SAMPLE_MINIKUBE_CERT.encode('ascii'))
  
  
  @pytest.fixture
  def pkeybase64(tmpdir):
      return base64.encodebytes(SAMPLE_MINIKUBE_PKEY.encode('ascii'))
  
  
  @pytest.fixture
  def cafile(tmpdir):
      path = tmpdir / 'ca.crt'
      path.write_text(SAMPLE_MINIKUBE_CA, encoding='utf-8')
      return str(path)
  
  
  @pytest.fixture
  def certfile(tmpdir):
      path = tmpdir / 'client.crt'
      path.write_text(SAMPLE_MINIKUBE_CERT, encoding='utf-8')
      return str(path)
  
  
  @pytest.fixture
  def pkeyfile(tmpdir):
      path = tmpdir / 'client.key'
      path.write_text(SAMPLE_MINIKUBE_PKEY, encoding='utf-8')
      return str(path)
  
  
  async def test_basic_auth(vault):
      await vault.populate({
          'id': ConnectionInfo(
              server='http://localhost',
              username='username',
              password='password',
          ),
      })
      session = await fn()
  
      async with session:
          assert session._default_auth.login == 'username'
          assert session._default_auth.password == 'password'
          assert 'Authorization' not in session._default_headers
  
  
  async def test_header_with_token_only(vault):
      await vault.populate({
          'id': ConnectionInfo(
              server='http://localhost',
              token='token',
          ),
      })
      session = await fn()
  
      async with session:
          assert session._default_auth is None
          assert session._default_headers['Authorization'] == 'Bearer token'
  
  
  async def test_header_with_schema_only(vault):
      await vault.populate({
          'id': ConnectionInfo(
              server='http://localhost',
              scheme='Digest xyz',
          ),
      })
      session = await fn()
  
      async with session:
          assert session._default_auth is None
          assert session._default_headers['Authorization'] == 'Digest xyz'
  
  
  async def test_header_with_schema_and_token(vault):
      await vault.populate({
          'id': ConnectionInfo(
              server='http://localhost',
              scheme='Digest',
              token='xyz',
          ),
      })
      session = await fn()
  
      async with session:
          assert session._default_auth is None
          assert session._default_headers['Authorization'] == 'Digest xyz'
  
  
  async def test_ca_insecure(vault, cafile):
      await vault.populate({
          'id': ConnectionInfo(
              server='http://localhost',
              insecure=True,
          ),
      })
      session = await fn()
  
      async with session:
          ctx = session.connector._ssl
          assert ctx.verify_mode == ssl.CERT_NONE
  
  
  async def test_ca_as_path(vault, cafile):
      await vault.populate({
          'id': ConnectionInfo(
              server='http://localhost',
              ca_path=cafile,
          ),
      })
      session = await fn()
  
      async with session:
          ctx = session.connector._ssl
          assert len(ctx.get_ca_certs()) == 1
          assert ctx.cert_store_stats()['x509'] == 1
          assert ctx.cert_store_stats()['x509_ca'] == 1
  
  
  async def test_ca_as_data(vault, cabase64):
      await vault.populate({
          'id': ConnectionInfo(
              server='http://localhost',
              ca_data=cabase64,
          ),
      })
      session = await fn()
  
      async with session:
          ctx = session.connector._ssl
          assert len(ctx.get_ca_certs()) == 1
          assert ctx.cert_store_stats()['x509'] == 1
          assert ctx.cert_store_stats()['x509_ca'] == 1
  
  
  # TODO: find a way to test that the client certificates/pkeys are indeed loaded.
  # TODO: currently, we only test that the parsing/loading does not fail at all.
  async def test_clientcert_as_path(vault, cafile, certfile, pkeyfile):
      await vault.populate({
          'id': ConnectionInfo(
              server='http://localhost',
              ca_path=cafile,
              certificate_path=certfile,
              private_key_path=pkeyfile,
          ),
      })
      session = await fn()
  
      async with session:
          pass
  
  
  async def test_clientcert_as_data(vault, cafile, certbase64, pkeybase64):
      await vault.populate({
          'id': ConnectionInfo(
              server='http://localhost',
              ca_path=cafile,
              certificate_data=certbase64,
              private_key_data=pkeybase64,
          ),
      })
      session = await fn()
  
      async with session:
          pass
  