HEX

File: //lib/python3/dist-packages/awscli/customizations/eks/__pycache__/kubeconfig.cpython-310.pyc
o

.&�aH%�@s�ddlZddlZddlZddlZddlmZddlmZddlm	Z	m
Z
Gdd�de�ZGdd�de�ZGd	d
�d
e�Z
dd�ZGd
d�de�ZGdd�de�ZGdd�de�ZGdd�de�ZGdd�de�ZdS)�N��OrderedDict)�EKSError)�ordered_yaml_load�ordered_yaml_dumpc@�eZdZdZdS)�KubeconfigErrorz& Base class for all kubeconfig errors.N��__name__�
__module__�__qualname__�__doc__�rr�F/usr/lib/python3/dist-packages/awscli/customizations/eks/kubeconfig.pyr�rc@r)�KubeconfigCorruptedErrorz+ Raised when a kubeconfig cannot be parsed.Nr	rrrrrrrc@r)�KubeconfigInaccessableErrorz< Raised when a kubeconfig cannot be opened for read/writing.Nr	rrrrr!rrc	Cs(tddgfdgfdddt�fdgfg�S)N)�
apiVersion�v1�clusters�contexts)�current-context�)�kind�Config�preferences�usersrrrrr�_get_new_kubeconfig_content%s�rc@s&eZdZddd�Zdd�Zdd�ZdS)	�
KubeconfigNcCs||_|dur
t�}||_dS�N)�pathr�content)�selfr r!rrr�__init__2s
zKubeconfig.__init__cCs
t|j�S)z+ Return the stored content in yaml format. )rr!�r"rrr�dump_content8s
zKubeconfig.dump_contentcCs&d|jvrdS|dd�|jdD�vS)zg
        Return true if this kubeconfig contains an entry
        For the passed cluster name.
        rFcSsg|]}|d�qS)�namer)�.0�clusterrrr�
<listcomp>Cs�z*Kubeconfig.has_cluster.<locals>.<listcomp>)r!)r"r&rrr�has_cluster<s

�zKubeconfig.has_clusterr)r
rrr#r%r*rrrrr1s
rc@s,eZdZdd�Zdd�Zdd�Zdd�Zd	S)
�KubeconfigValidatorcCstdd�j|_dSr)rr!�_validation_contentr$rrrr#HszKubeconfigValidator.__init__cCs*t|t�s	td��|�|�|�|�dS)z�
        Raises KubeconfigCorruptedError if the passed content is invalid

        :param config: The config to validate
        :type config: Kubeconfig
        z(Internal error: Not a Kubeconfig object.N)�
isinstancerr�_validate_config_types�_validate_list_entry_types)r"�configrrr�validate_configMs

z#KubeconfigValidator.validate_configcCsxt|jt�s
td��|j��D]*\}}||jvr9|j|dur9t|j|t|��s9td�|t|j|�t|����qdS)z�
        Raises KubeconfigCorruptedError if any of the entries in config
        are the wrong type

        :param config: The config to validate
        :type config: Kubeconfig
        zContent not a dictionary.Nz%{0} is wrong type:{1} (Should be {2}))r-r!�dictrr,�items�type�format)r"r0�key�valuerrrr.Zs"
�����z*KubeconfigValidator._validate_config_typescCs\|j��D]&\}}||jvr+t|j|�tkr+|j|D]}t|t�s*td�|���qqdS)z�
        Raises KubeconfigCorruptedError if any lists in config contain objects
        which are not dictionaries

        :param config: The config to validate
        :type config: Kubeconfig
        zEntry in {0} not a dictionary.N)	r,r3r!r4�listr-rrr5)r"r0r6r7�elementrrrr/qs

����z.KubeconfigValidator._validate_list_entry_typesN)r
rrr#r1r.r/rrrrr+Gs

r+c@seZdZddd�Zdd�ZdS)�KubeconfigLoaderNcCs|durt�}||_dSr)r+�
_validator)r"�	validatorrrrr#�s
zKubeconfigLoader.__init__c
Cs�zt|d��}t|�}Wd�n1swYWn5ty=}z|jtjkr,d}ntd�|���WYd}~nd}~wtjyP}zt	d�|���d}~wwt
||�}|j�|�|S)aP
        Loads the kubeconfig found at the given path.
        If no file is found at the given path,
        Generate a new kubeconfig to write back.
        If the kubeconfig is valid, loads the content from it.
        If the kubeconfig is invalid, throw the relevant exception.

        :param path: The path to load a kubeconfig from
        :type path: string

        :raises KubeconfigInaccessableError: if the kubeconfig can't be opened
        :raises KubeconfigCorruptedError: if the kubeconfig is invalid

        :return: The loaded kubeconfig
        :rtype: Kubeconfig
        �rNz&Can't open kubeconfig for reading: {0}z'YamlError while loading kubeconfig: {0})
�openr�IOError�errno�ENOENTrr5�yaml�	YAMLErrorrrr;r1)r"r �stream�loaded_content�e�
loaded_configrrr�load_kubeconfig�s,
��������
z KubeconfigLoader.load_kubeconfigr)r
rrr#rHrrrrr:�s
r:c@seZdZdd�ZdS)�KubeconfigWriterc
Cs�tj�|j�}zt�|�Wnty-}z|jtjkr#td�|���WYd}~nd}~wwz.t�	t�
|jtjtjBtj
Bd�d��}t|j|�Wd�WdS1sUwYWdSttfyq}ztd�|���d}~ww)a
        Write config to disk.
        OK if the file doesn't exist.

        :param config: The kubeconfig to write
        :type config: Kubeconfig

        :raises KubeconfigInaccessableError: if the kubeconfig
        can't be opened for writing
        z'Can't create directory for writing: {0}Ni�zw+z&Can't open kubeconfig for writing: {0})�osr �dirname�makedirs�OSErrorr@�EEXISTrr5�fdopenr>�O_CREAT�O_RDWR�O_TRUNCrr!r?)r"r0�	directoryrFrDrrr�write_kubeconfig�s:������&����z!KubeconfigWriter.write_kubeconfigN)r
rrrTrrrrrI�srIc@s(eZdZdd�Zddd�Zddd�ZdS)	�KubeconfigAppendercCs�||jvr
g|j|<|j|}t|t�std�|t|�t���d}t|�D]\}}d|vr?d|vr?|d|dkr?|||<d}q%|sG|�|�||j|<|S)z�
        Insert entry into the array at content[key]
        Overwrite an existing entry if they share the same name

        :param config: The kubeconfig to insert an entry into
        :type config: Kubeconfig
        z1Tried to insert into {0},which is a {1} not a {2}Fr&T)r!r-r8rr5r4�	enumerate�append)r"r0r6�entry�array�found�counter�existing_entryrrr�insert_entry�s,



���

zKubeconfigAppender.insert_entryNcCs4tdtd|dfd|dfg�fd|p|dfg�S)zE Generate a context to associate cluster and user with a given alias.�contextr(r&�userr)r"r(r_�aliasrrr�
_make_context�s

��z KubeconfigAppender._make_contextcCsL|j|||d�}|�|d|�|�|d|�|�|d|�|d|jd<|S)ag
        Insert the passed cluster entry and user entry,
        then make a context to associate them
        and set current-context to be the new context.
        Returns the new context

        :param config: the Kubeconfig to insert the pair into
        :type config: Kubeconfig

        :param cluster: the cluster entry
        :type cluster: OrderedDict

        :param user: the user entry
        :type user: OrderedDict

        :param alias: the alias for the context; defaults top user entry name
        :type context: str

        :return: The generated context
        :rtype: OrderedDict
        )r`rrrr&r)rar]r!)r"r0r(r_r`r^rrr�insert_cluster_user_pair�sz+KubeconfigAppender.insert_cluster_user_pairr)r
rrr]rarbrrrrrU�s

rU)rJrB�loggingr@�botocore.compatr�$awscli.customizations.eks.exceptionsr�&awscli.customizations.eks.ordered_yamlrrrrrr�objectrr+r:rIrUrrrr�<module>s 
;*!