##
import asyncio
-import base64
-import binascii
import logging
import os
-import re
from n2vc.exceptions import (
N2VCBadArgumentsException,
N2VCException,
N2VCConnectionException,
N2VCExecutionException,
- N2VCInvalidCertificate,
# N2VCNotFound,
MethodNotImplemented,
JujuK8sProxycharmNotSupported,
from n2vc.n2vc_conn import N2VCConnector
from n2vc.n2vc_conn import obj_to_dict, obj_to_yaml
from n2vc.libjuju import Libjuju
+from n2vc.utils import base64_to_cacert
class N2VCJujuConnector(N2VCConnector):
# TODO: Verify ca_cert is valid before using. VCA will crash
# if the ca_cert isn't formatted correctly.
- def base64_to_cacert(b64string):
- """Convert the base64-encoded string containing the VCA CACERT.
-
- The input string....
-
- """
- try:
- cacert = base64.b64decode(b64string).decode("utf-8")
-
- cacert = re.sub(r"\\n", r"\n", cacert,)
- except binascii.Error as e:
- self.log.debug("Caught binascii.Error: {}".format(e))
- raise N2VCInvalidCertificate(message="Invalid CA Certificate")
-
- return cacert
self.ca_cert = vca_config.get("ca_cert")
if self.ca_cert:
# return public key if exists
return output["pubkey"] if "pubkey" in output else output
+ async def get_metrics(self, model_name: str, application_name: str) -> dict:
+ return await self.libjuju.get_metrics(model_name, application_name)
+
async def add_relation(
self, ee_id_1: str, ee_id_2: str, endpoint_1: str, endpoint_2: str
):