File: //usr/lib64/python3.9/site-packages/rhsm/__pycache__/utils.cpython-39.opt-1.pyc
a
X�ZhBI �
@ s$ U d dl Z d dlZd dlZd dlZd dlZd dlZd dlZd dlmZm Z m
Z
m Z m Z m
Z
mZ d dlZd dlZd dlmZ dZeed< G dd� de j�Zeeed�d d
�Zeed �d d
�ZG dd� de �Z G dd� de �Z G dd� de �Z G dd� de �Z G dd� de �Z!G dd� de �Z"G dd� de �Z#G d d � d e �Z$eed �d d �Z%eed �d!d"�Z&d?eeeeeee eeeeef d#�d$d%�Z'e(d&�d'd(�Z)e e ed)�d*d+�Z*dd&�d,d-�Z+eed.�d/d0�Z,e-e-d1�d2d3�Z.eed4�d5d6�Z/e-e-d1�d7d8�Z0G d9d:� d:�Z1G d;d<� d<�Z2G d=d>� d>e1�Z3dS )@� N)�Callable�List�Optional�TextIO�Tuple�Union� Generator)�DEFAULT_PROXY_PORTT�PROGRESS_MESSAGESc @ s$ e Zd ZdZdZdZdZdZdZdS )�COLORz[91mz[93mz[92mz[95mz[94m�[0mN) �__name__�
__module__� __qualname__ZREDZYELLOWZGREENZMAGENTAZBLUE�RESET� r r �0/usr/lib64/python3.9/site-packages/rhsm/utils.pyr & s r )�text�color�returnc C s |j | tjj S �N)�valuer r )r r r r r �colorize/ s r )�urir c C s t �dd| �S )z'Remove the scheme component from a URI.z^[A-Za-z][A-Za-z0-9+-.]*://� )�re�sub)r r r r �
remove_scheme3 s r c @ s e Zd Zdeed�dd�ZdS )�ServerUrlParseErrorN�� serverurl�msgc C s || _ || _d S r r )�selfr r! r r r �__init__9 s z ServerUrlParseError.__init__)N)r
r r �strr# r r r r r 8 s r c @ s e Zd ZdS )�ServerUrlParseErrorEmptyN�r
r r r r r r r% > s r% c @ s e Zd ZdS )�ServerUrlParseErrorNoneNr& r r r r r' B s r' c @ s e Zd ZdS )�ServerUrlParseErrorPortNr& r r r r r( F s r( c @ s e Zd ZdS )�ServerUrlParseErrorPathNr& r r r r r) J s r) c @ s e Zd ZdS )�ServerUrlParseErrorSchemeNr& r r r r r* N s r* c @ s e Zd ZdS )� ServerUrlParseErrorJustSchemeNr& r r r r r+ R s r+ c @ s e Zd ZdZdS )� UnsupportedOperationExceptionz�Thrown when a call is made that is unsupported in the current
state. For example, if a call is made to a deprecated API when
a newer API is available.
N)r
r r �__doc__r r r r r, V s r, )�urlr c C s, d}d}t �|| �rdS t �|| �r(dS dS )z�Check a url for an invalid or unuseful schema.
Don't allow urls to start with :/ http/ https/ non http/httpsm or http(s) with single /
:params url: URL string to check
:returns: True if the url scheme is "bad"
z(https?[:/])|(:/)|(\S+://)z https?://FT)r �match)r. Z match_badZ
match_goodr r r �has_bad_scheme_ s r0 c C s* t �d| �}|sdS |�d�s&t| ��dS )Nzhttps?://(\S+)?F� T)r r/ �groupr+ )r. r/ r r r �has_good_schemeq s
r3 )�local_server_entry�default_hostname� default_port�default_prefix�default_username�default_passwordr c
C s� | dkrt | d��| du r$t| d��d}t| �r8t| ��t| �rD| }|sPd| }tj�|�}|j}|du sr|dkrv|}|j } | du r�|} |j
}
|
du r�|}
z
|j } W n t y� t
| ��Y n0 | du r�|} nt| �} |j} | du s�| dkr�|} || |
| | fS )a
Parse hostname, port, and webapp prefix from the string a user entered.
Expected format: username:password@hostname:port/prefix
:param local_server_entry: URL of a candlepin server
:param default_hostname: default hostname
:param default_port: default port
:param default_prefix: default prefix (e.g. /candlepin)
:param default_username: not encrypted default username
:param default_password: not encrypted dfault password
:return: a tuple of (username, password, hostname, port, path)
r zServer entry is emptyNzNo server entry providedz
https://%s)r% r' r0 r* r3 �urllib�parse�urlparse�username�password�hostname�port�
ValueErrorr( r$ �path)
r4 r5 r6 r7 r8 r9 Zgood_url�resultr= r> r? r@ �prefixr r r � parse_url{ s@
rE �r c C s� ddddd�} d}g d�}|D ]"}t �|�}|r t|td�} qBq |r�|d | d< |d | d < |d
| d < |d d u s�|d dkr�d | d
< nt|d �| d
< | S )Nr )�proxy_username�proxy_hostname�
proxy_port�proxy_passwordr )Z HTTPS_PROXYZ https_proxyZ
HTTP_PROXYZ
http_proxy)r6 r rG r1 rJ � rH � rI )�os�getenvrE r �int)Z the_proxy�infoZenv_varsZevZ
proxy_infor r r �get_env_proxy_info� s( �
rQ )�argvr c C s t j�| d �}|S )z�Attempt to get a meaningful command name from argv.
This handles cases where argv[0] isn't helpful (for
example, '/usr/bin/python' or '__main__.py'.
r )rM rB �basename)rR Zcmd_name_stringr r r �cmd_name� s rT c C sZ t j�d�pt j�d�} | durV| dkrVd�dd� | �d�D ��} | t jd< | t jd< dS )z�
This fixes no_proxy/NO_PROXY environment to not include leading
asterisk, because there is some imperfection in proxy_bypass_environment.
�no_proxyZNO_PROXYN�*�,c S s g | ]}|� d ��qS )z *)�lstrip)�.0�itemr r r �
<listcomp> � z fix_no_proxy.<locals>.<listcomp>)rM �environ�get�join�split)rU r r r � fix_no_proxy� s
ra )�funcr c s � fdd�}|S )Nc s� d }d }d }z�t tjd�}tj}tj}|t_|t_� | i |��W |d urN|t_|d ur\|t_|d url|�� S |d ur||t_|d ur�|�� n.|d ur�|t_|d ur�|t_|d ur�|�� 0 d S )N�w)�openrM �devnull�sys�stdout�stderr�close)�args�kwargsre rg rh �rb r r �wrapper s: �
�z suppress_output.<locals>.wrapperr )rb rm r rl r �suppress_output
s rn )�clsr c s" | j � d| _� fdd�}|| _ | S )z�Decorate a class to make it singleton.
This decorator is inherited: subclasses will be singletons as well,
but they won't be the same singleton as their parent.
Nc s@ t | j| �s:� tju r"� | �| _n� | g|�R i |��| _| jS r )�
isinstance� _instance�object�__new__)Zklsrj rk �Z __orig_new__r r rs + s
zsingleton.<locals>.__new__)rs rq )ro rs r rt r � singleton! s
ru )�fnr c s< t �� � _d� _t�� �� fdd��}� fdd�}||_|S )zrDecorate a function to make it callable just once.
All further calls do not do anything and return None.
Fc sr � j jddd�}|s&td� j� d���z:� jr>W � j �� d S d� _� | i |��}|W � j �� S � j �� 0 d S )NT� �?��blocking�timeout�.Could not acquire call_once lock for function �.��_call_once_lock�acquire� RuntimeErrorr
�_called�release)rj rk �lockedZ fn_result�rv r r rm @ s
�
�zcall_once.<locals>.wrapperc sL � j jddd�} | s&td� j� d���zd� _W � j �� n � j �� 0 dS )z�Reset function state to allow another call to it.
This is used for testing purposes, there is no need to call this in
production code itself.
Trw rx r{ r| FNr} )r� r� r r �_resetP s zcall_once.<locals>._reset)� threading�RLockr~ r� � functools�wrapsr� )rv rm r� r r� r � call_once8 s
r� c C st t �� | _d| _tdd� d�| _dd�dd�}dd�d d
�}| d�d d �}dd�d
d�}|| _|| _|| _|| _ | S )z�Decorate a class to make it thread-safe lock.
It will provide read-only 'locked' attribute, functions lock() and unlock()
and __enter__/__exit__ methods for context manager functionality.
Fc S s | j S r )�_locked�r" r r r �<lambda>r r\ zlock.<locals>.<lambda>)�fgetNrF c S s | j �� d| _dS )z�Lock using RLock.
One thread can acquire the lock multiple times, but other threads cannot
until the lock is completely unlocked.
TN)�_lockr r� r� r r r �lockt s
zlock.<locals>.lockc S s: z,z| j �� W n ty" Y n0 W d| _nd| _0 dS )zUnlock the lock.FN)r� r� r� r� r� r r r �unlock} s
zlock.<locals>.unlockc S s | � � | S r )r� r� r r r � __enter__� s zlock.<locals>.__enter__c S s | � � d S r )r� )r" �exc_typeZexc_valZexc_tbr r r �__exit__� s zlock.<locals>.__exit__)
r� r� r� r� �propertyr� r� r� r� r� )ro r� r� r� r� r r r r� d s
r� c @ sZ e Zd ZdZee d�dd�Zdd�dd�Zdd�d d
�Zdd�d d �Z dd�d
d�Z
dS )�
StatusMessagea Class for temporary reporting.
While you can call 'print()' and 'clean()' methods directly, the easier
way is to use context manager: 'with StatusMessage("Fetching data"):'.
This object will print the description to stdout when called. When the
context manager exists (either because the code finished or because an
error got raised inside), printed output will be cleared before the
program continues with its execution. This ensures that the message will
disappear after it's no longer valid.
)� descriptionc C s� |d u r d}|| _ d}d}|� | j � |� �| _d| _tj�� }|�dd�dkrTd| _ts^d| _tj �
� snd| _t j �d d
�r�d| _d S ) NzTransmitting dataz[3mr F�rhsmZprogress_messages�0TZSUBMAN_DEBUG_PRINT_REQUESTr )
�raw_textr �quietr� �configZget_config_parserr^ r
rf rg �isattyrM r] )r" r� ZCURSIVEr r� r r r r# � s
zStatusMessage.__init__NrF c C s | j r
d S t| jdd� d S )N�
��end)r� �printr r� r r r r� � s zStatusMessage.printc C s | j r
d S tddd� d S �Nz[0Kr� r� �r� r� r� r r r �clean� s zStatusMessage.cleanc C s | � � d S r )r� r� r r r r� � s zStatusMessage.__enter__c C s | � � |r� d S r )r� �r" Z
error_typeZ error_value� tracebackr r r r� � s zStatusMessage.__exit__) r
r r r- r r$ r# r� r� r� r� r r r r r� � s r� c @ sv e Zd ZU dZdZee ed< dZee ed< dZ ee ed< dZ
ee ed < d
Z ee ed < e
e Z ee ed < d
S )�StatusSpinnerStyleag Class for spinner animations.
While subscription-manager may not be using all spinners used below,
they have been defined here, so they can be easily used in the future.
The default spinner, LINE, has been chosen for several reasons:
- it is small (only one character wide),
- it has small loop cycle (so quickly switching several status messages
looks like it is one spinner with several messages, even though every
message has its own spinner),
- it is only made of ASCII characters (which makes it renderable on
all TTYs, not just rich terminal emulators in GUI).
)�|�/�-�\�LINE)u ⠋u ⠙u ⠸u ⠴u ⠦u ⠇�BRAILLE)
u ⠧ u ⠏ u ⠋⠁u ⠉⠉u ⠈⠙u ⠹u ⠼u ⠠⠴u ⠤⠤u ⠦⠄� WIDE_BRAILLE)�[ ]�[= ]�[== ]�[=== ]�[====]�[ ===]�[ ==]�[ =]� BAR_FORWARD)r� r� r� r� r� r� r� r� � BAR_BACKWARD�
BAR_BOUNCEN)
r
r r r- r� r r$ �__annotations__r� r� r� r� r� r r r r r� � s
r� c s� e Zd ZdZejddd�ee ee ee dd�� fdd�Z
e e eddf d �d
d ��Z
e ed �d d
��Zejedd�dd
��Zdd �dd�Zdd �dd�Zdd �dd�Zdd �dd�Zdd �dd�Z� ZS )�LiveStatusMessagea� Class for temporary reporting, with activity spinner.
While you can call 'print()' and 'clean()' methods directly, the easier
way is to use context manager: 'with LiveStatusMessage("Fetching data"):'.
This object will print the description to stdout when called. When the
context manager exists (either because the code finished or because an
error got raised inside), printed output will be cleared before the
program continues with its execution. This ensures that the message will
disappear after it's no longer valid.
You can set several loading styles (via class StatusSpinnerStyle) or choose
the speed in which the animation is played (per frame, in seconds).
�BEFOREg333333�?)�style� placement�speedN)r� r� r� r� r c sZ t � �|� | j| _d| _d| _d | _d| _|| _|| _ |dvrPt
d|� d���|| _ d S )NFr T)r� ZAFTERzString z is not valid spinner placement.) �superr# r� r �busy�_loops�_thread�_cursor�frames�delayrA r� )r" r� r� r� r� �� __class__r r r# � s zLiveStatusMessage.__init__rF c c s | j | jt| j � V q dS )z(Get next frame of the spinner animation.N)r� r� �lenr� r r r �
spinner_frame s z LiveStatusMessage.spinner_framec C s | j S )z Get cursor visibility state.)r� r� r r r �cursor s zLiveStatusMessage.cursor)�enabler c C sL t |�tur tdt |�����|r6tddd� d| _ntddd� d| _dS ) z�Enable or disable cursor.
For more information on these rarely used shell escape codes, see
https://en.wikipedia.org/wiki/ANSI_escape_code#CSI_(Control_Sequence_Introducer)_sequences
zExpected bool, got z[?25hr r� Tz[?25lFN)�type�boolrA r� r� )r" r� r r r r� " s c C s4 d| _ | jrd S d| _tj| jd�| _| j�� d S )NTF)�target)r� r� r� r� �Thread�loopr� �startr� r r r r� 2 s zLiveStatusMessage.__enter__c C s6 d| _ | jr|r� d S d| _| jj| jd� |r2� d S )NFT)rz )r� r� r� r� r_ r� r� r r r r� : s zLiveStatusMessage.__exit__c C sL | j r
d S t| j�}| jdkr.|d | j }n| jd | }t|dd� d S )Nr� � r� r� )r� �nextr� r� r r� )r" �frame�liner r r r� E s
zLiveStatusMessage.printc C s | j r
d S tddd� d S r� r� r� r r r r� P s zLiveStatusMessage.cleanc C s6 | j r2| �� | jd7 _t�| j� | �� q dS )z*Show pretty animation while we fetch data.r1 N)r� r� r� �time�sleepr� r� r� r r r r� U s
zLiveStatusMessage.loop)r
r r r- r� r� r r$ r �floatr# r� r r� r� r� �setterr� r� r� r� r� �
__classcell__r r r� r r� � s, �� r� )NNNNN)4�enumr� rM r rf r� r� �typingr r r r r r r � urllib.parser: Z rhsm.configr� r r
r� r� �Enumr r$ r r � Exceptionr r% r' r( r) r* r+ r, r0 r3 rE �dictrQ rT ra rn r� ru r� r� r� r� r� r r r r �<module> s`
$ �� N#
,25