File manager - Edit - /opt/saltstack/salt/lib/python3.10/site-packages/libcloud/storage/drivers/__pycache__/dummy.cpython-310.pyc
Back
o ;j�G � @ s� d dl Z d dlZd dlZd dlmZ d dlmZ d dl m Z d dl m Z m Z mZ d dlmZmZmZmZ G dd� de�ZG d d � d �ZG d d � d e�Zed krYd dlZe�� dS dS )� N)�FileIO)�b)� LibcloudError)�Object� Container� StorageDriver)�ObjectDoesNotExistError�ContainerIsNotEmptyError�ContainerDoesNotExistError�ContainerAlreadyExistsErrorc @ s. e Zd Zd dd�Zdd� Zdd� Zd d � Zd S ) �DummyFileObject� � c C s || _ || _d S �N�� _yield_count� _chunk_len)�selfZ yield_count� chunk_len� r �R/opt/saltstack/salt/lib/python3.10/site-packages/libcloud/storage/drivers/dummy.py�__init__! s zDummyFileObject.__init__c c s8 � d}|| j k r| �| j�V |d7 }|| j k sd S d S )Nr � )r � _get_chunkr )r �size�ir r r �read% s � �zDummyFileObject.readc C s dd� t �dd�D �}|S )Nc S s g | ]}t |��qS r )�str)�.0�xr r r � <listcomp>- s z.DummyFileObject._get_chunk.<locals>.<listcomp>�a �x )�random�randint)r r �chunkr r r r , s zDummyFileObject._get_chunkc C s | j | j S r r �r r r r �__len__1 s zDummyFileObject.__len__N)r r )�__name__� __module__� __qualname__r r r r' r r r r r s r c @ s> e Zd Zddd�Zdd� Zdd� Zdd � Zd d � Zd d � ZdS )� DummyIteratorNc C s t �� | _|pg | _d| _d S )Nr )�hashlib�md5�hash�_data� _current_item)r �datar r r r 6 s zDummyIterator.__init__c C s | j �� S r )r. � hexdigestr& r r r � get_md5_hash; s zDummyIterator.get_md5_hashc C sB | j t| j�kr t�| j| j }| j�t|�� | j d7 _ |S )Nr )r0 �lenr/ � StopIterationr. �updater )r �valuer r r �next> s zDummyIterator.nextc C s | � � S r )r8 r& r r r �__next__H s zDummyIterator.__next__c C � d S r r r& r r r � __enter__K � zDummyIterator.__enter__c C r: r r )r �typer7 � tracebackr r r �__exit__N r< zDummyIterator.__exit__r ) r( r) r* r r3 r8 r9 r; r? r r r r r+ 5 s r+ c @ s� e Zd ZdZdZdZdd� Zdd� Zdd � Zd'd d �Z d d� Z dd� Z dd� Z dd� Z dd� Zdd� Z d(dd �Zd)d d �Z d*d d �Zd'd!d"�Zd#d$� Zd)d%d&�Zd S )+�DummyStorageDrivera� Dummy Storage driver. >>> from libcloud.storage.drivers.dummy import DummyStorageDriver >>> driver = DummyStorageDriver('key', 'secret') >>> container = driver.create_container(container_name='test container') >>> container <Container: name=test container, provider=Dummy Storage Provider> >>> container.name 'test container' >>> container.extra['object_count'] 0 zDummy Storage Providerzhttp://example.comc C s i | _ dS )z� :param api_key: API key or username to used (required) :type api_key: ``str`` :param api_secret: Secret password to be used (required) :type api_secret: ``str`` :rtype: ``None`` N)� _containers)r Zapi_keyZ api_secretr r r r d s zDummyStorageDriver.__init__c st t � j�}t� fdd�� jD ��}d}� jD ]}� j| d }|�� D ] \}}||j7 }q#qt|�t|�t|�d�S )aa >>> driver = DummyStorageDriver('key', 'secret') >>> driver.get_meta_data()['object_count'] 0 >>> driver.get_meta_data()['container_count'] 0 >>> driver.get_meta_data()['bytes_used'] 0 >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container_name = 'test container 2' >>> container = driver.create_container(container_name=container_name) >>> obj = container.upload_object_via_stream( ... object_name='test object', iterator=DummyFileObject(5, 10), ... extra={}) >>> driver.get_meta_data()['object_count'] 1 >>> driver.get_meta_data()['container_count'] 2 >>> driver.get_meta_data()['bytes_used'] 50 :rtype: ``dict`` c 3 s" � | ] }t � j| d �V qdS )�objectsN)r4 rA )r � containerr& r r � <genexpr>� s � �z3DummyStorageDriver.get_meta_data.<locals>.<genexpr>r rB )�container_count� object_count� bytes_used)r4 rA �sum�itemsr �int)r rE rF rG rC rB �_�objr r& r � get_meta_datan s � ��z DummyStorageDriver.get_meta_datac c s$ � t | j�� �D ]}|d V qdS )a4 >>> driver = DummyStorageDriver('key', 'secret') >>> list(driver.iterate_containers()) [] >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container <Container: name=test container 1, provider=Dummy Storage Provider> >>> container.name 'test container 1' >>> container_name = 'test container 2' >>> container = driver.create_container(container_name=container_name) >>> container <Container: name=test container 2, provider=Dummy Storage Provider> >>> container = driver.create_container( ... container_name='test container 2') ... #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ContainerAlreadyExistsError: >>> container_list=list(driver.iterate_containers()) >>> sorted([c.name for c in container_list]) ['test container 1', 'test container 2'] @inherits: :class:`StorageDriver.iterate_containers` rC N)�listrA �values�r rC r r r �iterate_containers� s � �z%DummyStorageDriver.iterate_containersNc C s8 | � ||�}| �|j�}| j|j d �� }| �||�S )NrB )Z_normalize_prefix_argument� get_container�namerA rO Z _filter_listed_container_objects)r rC �prefixZ ex_prefixrB r r r �iterate_container_objects� s z,DummyStorageDriver.iterate_container_objectsc C s& || j vr t| d|d��| j | d S )a� >>> driver = DummyStorageDriver('key', 'secret') >>> driver.get_container('unknown') #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ContainerDoesNotExistError: >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container <Container: name=test container 1, provider=Dummy Storage Provider> >>> container.name 'test container 1' >>> driver.get_container('test container 1') <Container: name=test container 1, provider=Dummy Storage Provider> @inherits: :class:`StorageDriver.get_container` N��driverr7 �container_namerC )rA r )r rX r r r rR � s z DummyStorageDriver.get_containerc C s, |j | jvrt| d|j d��| j|j d S )a� >>> driver = DummyStorageDriver('key', 'secret') >>> driver.get_container('unknown') #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ContainerDoesNotExistError: >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container <Container: name=test container 1, provider=Dummy Storage Provider> >>> container.name 'test container 1' >>> container.get_cdn_url() 'http://www.test.com/container/test_container_1' @inherits: :class:`StorageDriver.get_container_cdn_url` NrV �cdn_url)rS rA r rP r r r �get_container_cdn_url� s z(DummyStorageDriver.get_container_cdn_urlc C s6 | � |� | j| d }||vrt|d| d��|| S )a� >>> driver = DummyStorageDriver('key', 'secret') >>> driver.get_object('unknown', 'unknown') ... #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ContainerDoesNotExistError: >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container <Container: name=test container 1, provider=Dummy Storage Provider> >>> driver.get_object( ... 'test container 1', 'unknown') #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ObjectDoesNotExistError: >>> obj = container.upload_object_via_stream(object_name='test object', ... iterator=DummyFileObject(5, 10), extra={}) >>> obj.name 'test object' >>> obj.size 50 @inherits: :class:`StorageDriver.get_object` rB N�� object_namer7 rW )rR rA r )r rX r\ �container_objectsr r r � get_object� s z DummyStorageDriver.get_objectc C s@ |j j}| j| d }|j|vrt|jd| d��||j jd S )a >>> driver = DummyStorageDriver('key', 'secret') >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container <Container: name=test container 1, provider=Dummy Storage Provider> >>> obj = container.upload_object_via_stream( ... object_name='test object 5', ... iterator=DummyFileObject(5, 10), extra={}) >>> obj.name 'test object 5' >>> obj.get_cdn_url() 'http://www.test.com/object/test_object_5' @inherits: :class:`StorageDriver.get_object_cdn_url` rB Nr[ rY )rC rS rA r � meta_data)r rL rX r] r r r �get_object_cdn_url s z%DummyStorageDriver.get_object_cdn_urlc C sP || j v r t|d| d��ddi}t||| d�}|i d|�dd� d �| j |< |S ) aA >>> driver = DummyStorageDriver('key', 'secret') >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container <Container: name=test container 1, provider=Dummy Storage Provider> >>> container = driver.create_container( ... container_name='test container 1') ... #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ContainerAlreadyExistsError: @inherits: :class:`StorageDriver.create_container` N�rX r7 rW rF r )rS �extrarW z http://www.test.com/container/%s� rK )rC rB rY )rA r r �replace)r rX rb rC r r r �create_container* s � �z#DummyStorageDriver.create_containerc C sR |j }|| jvrt|d| d��| j| }t|d �dkr#t|d| d��| j|= dS )a� >>> driver = DummyStorageDriver('key', 'secret') >>> container = Container(name = 'test container', ... extra={'object_count': 0}, driver=driver) >>> driver.delete_container(container=container) ... #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ContainerDoesNotExistError: >>> container = driver.create_container( ... container_name='test container 1') ... #doctest: +IGNORE_EXCEPTION_DETAIL >>> len(driver._containers) 1 >>> driver.delete_container(container=container) True >>> len(driver._containers) 0 >>> container = driver.create_container( ... container_name='test container 1') ... #doctest: +IGNORE_EXCEPTION_DETAIL >>> obj = container.upload_object_via_stream( ... object_name='test object', iterator=DummyFileObject(5, 10), ... extra={}) >>> driver.delete_container(container=container) ... #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ContainerIsNotEmptyError: @inherits: :class:`StorageDriver.delete_container` Nra rB r T)rS rA r r4 r )r rC rX r r r �delete_containerJ s z#DummyStorageDriver.delete_containerFTc C s" |t � |||d�}| jdi |��S )N)rL �response�destination_path�overwrite_existing�delete_on_failurer )r Z _save_object)r rL rh ri rj Z kwargs_dictr r r �download_objectx s �z"DummyStorageDriver.download_objectc C s t � S )a >>> driver = DummyStorageDriver('key', 'secret') >>> container = driver.create_container( ... container_name='test container 1') ... #doctest: +IGNORE_EXCEPTION_DETAIL >>> obj = container.upload_object_via_stream(object_name='test object', ... iterator=DummyFileObject(5, 10), extra={}) >>> stream = container.download_object_as_stream(obj) >>> stream #doctest: +ELLIPSIS <...closed...> @inherits: :class:`StorageDriver.download_object_as_stream` )r )r rL Z chunk_sizer r r �download_object_as_stream� s z,DummyStorageDriver.download_object_as_streamc C s: t j�|�std| | d��t j�|�}| j||||d�S )a+ >>> driver = DummyStorageDriver('key', 'secret') >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container.upload_object(file_path='/tmp/inexistent.file', ... object_name='test') #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): LibcloudError: >>> file_path = path = os.path.abspath(__file__) >>> file_size = os.path.getsize(file_path) >>> obj = container.upload_object(file_path=file_path, ... object_name='test') >>> obj #doctest: +ELLIPSIS <Object: name=test, size=...> >>> obj.size == file_size True @inherits: :class:`StorageDriver.upload_object` zFile %s does not exist)r7 rW �rC r\ r rb )�os�path�existsr �getsize� _add_object)r Z file_pathrC r\ rb Z verify_hash�headersr r r r � upload_object� s �z DummyStorageDriver.upload_objectc C s t |�}| j||||d�S )a >>> driver = DummyStorageDriver('key', 'secret') >>> container = driver.create_container( ... container_name='test container 1') ... #doctest: +IGNORE_EXCEPTION_DETAIL >>> obj = container.upload_object_via_stream( ... object_name='test object', iterator=DummyFileObject(5, 10), ... extra={}) >>> obj #doctest: +ELLIPSIS <Object: name=test object, size=50, ...> @inherits: :class:`StorageDriver.upload_object_via_stream` rm )r4 rr )r �iteratorrC r\ rb rs r r r r �upload_object_via_stream� s �z+DummyStorageDriver.upload_object_via_streamc C s0 |j j}|j}| j||d�}| j| d |= dS )a` >>> driver = DummyStorageDriver('key', 'secret') >>> container = driver.create_container( ... container_name='test container 1') ... #doctest: +IGNORE_EXCEPTION_DETAIL >>> obj = container.upload_object_via_stream(object_name='test object', ... iterator=DummyFileObject(5, 10), extra={}) >>> obj #doctest: +ELLIPSIS <Object: name=test object, size=50, ...> >>> container.delete_object(obj=obj) True >>> obj = Object(name='test object 2', ... size=1000, hash=None, extra=None, ... meta_data=None, container=container,driver=None) >>> container.delete_object(obj=obj) #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ObjectDoesNotExistError: @inherits: :class:`StorageDriver.delete_object` )rX r\ rB T)rC rS r^ rA )r rL rX r\ r r r � delete_object� s z DummyStorageDriver.delete_objectc C sh | � |j�}|p i }|�di �}|�dd|�dd� i� t|||d ||| d�}|| j|j d |< |S )Nr_ rY z http://www.test.com/object/%src rK )rS r rb r. r_ rC rW rB )rR rS �getr6 rd r rA )r rC r\ r rb r_ rL r r r rr � s" �� z DummyStorageDriver._add_object)NN)FTr )NTN)r( r) r* �__doc__rS Zwebsiter rM rQ rU rR rZ r^ r` re rf rk rl rt rv rw rr r r r r r@ R s0 - ! / � � & r@ �__main__)r# r, Zos.pathrn �ior �fileZlibcloud.utils.py3r Zlibcloud.common.typesr Zlibcloud.storage.baser r r Zlibcloud.storage.typesr r r r r r+ r@ r( ZdoctestZtestmodr r r r �<module> s$ 7 �
| ver. 1.4 |
Github
|
.
| PHP 8.2.30 | Generation time: 0 |
proxy
|
phpinfo
|
Settings