# Path: glanceclient/common/progressbar.py
# class _ProgressBarBase(object):
# class VerboseFileWrapper(_ProgressBarBase):
# class VerboseIteratorWrapper(_ProgressBarBase):
#     def __init__(self, wrapped, totalsize):
#     def _display_progress_bar(self, size_read):
#     def __getattr__(self, attr):
#     def read(self, *args, **kwargs):
#     def __iter__(self):
#     def next(self):
# 
# Path: glanceclient/common/utils.py
# SENSITIVE_HEADERS = ('X-Auth-Token', )
# REQUIRED_FIELDS_ON_DATA = ('disk_format', 'container_format')
# def arg(*args, **kwargs):
#     def _decorator(func):
# def on_data_require_fields(data_fields, required=REQUIRED_FIELDS_ON_DATA):
#     def args_decorator(func):
#         def prepare_fields(fields):
#         def func_wrapper(gc, args):
# def schema_args(schema_getter, omit=None):
#     def _decorator(func):
# def pretty_choice_list(l):
# def has_version(client, version):
# def print_cached_images(cached_images):
# def print_dict_list(objects, fields):
# def print_list(objs, fields, formatters=None, field_settings=None):
# def _encode(src):
# def unicode_key_value_to_string(src):
# def print_dict(d, max_column_width=80):
# def find_resource(manager, name_or_id):
# def env(*vars, **kwargs):
# def exit(msg='', exit_code=1):
# def print_err(msg):
# def save_image(data, path):
# def make_size_human_readable(size):
# def get_file_size(file_obj):
# def get_data_file(args):
# def strip_version(endpoint):
# def print_image(image_obj, human_readable=False, max_col_width=None):
# def integrity_iter(iter, checksum):
# def serious_integrity_iter(iter, hasher, hash_value):
# def memoized_property(fn):
#     def _memoized_property(self):
# def safe_header(name, value):
# def endpoint_version_from_url(endpoint, default_version=None):
# def debug_enabled(argv):
#     def __init__(self, iterable, length):
#     def __iter__(self):
#     def next(self):
#     def __len__(self):
#     def __init__(self, wrapped):
#     def request_ids(self):
#     def wrapped(self):
#     def next(self):
#     def __init__(self, wrapped):
#     def _set_request_ids(self, resp):
#     def _next(self):
#     def next(self):
#     def __next__(self):
#     def __iter__(self):
#     def request_ids(self):
#     def wrapped(self):
# def add_req_id_to_object():
#     def inner(wrapped, instance, args, kwargs):
# def add_req_id_to_generator():
#     def inner(wrapped, instance, args, kwargs):
# def _extract_request_id(resp):
# class IterableWithLength(object):
# class RequestIdProxy(wrapt.ObjectProxy):
# class GeneratorProxy(wrapt.ObjectProxy):
# 
# Path: glanceclient/tests/utils.py
# class FakeAPI(object):
# class FakeSchemaAPI(FakeAPI):
# class RawRequest(object):
# class FakeResponse(object):
# class TestCase(testtools.TestCase):
# class FakeTTYStdout(io.StringIO):
# class FakeNoTTYStdout(FakeTTYStdout):
#     def __init__(self, fixtures):
#     def _request(self, method, url, headers=None, data=None,
#                  content_length=None):
#     def get(self, *args, **kwargs):
#     def post(self, *args, **kwargs):
#     def put(self, *args, **kwargs):
#     def patch(self, *args, **kwargs):
#     def delete(self, *args, **kwargs):
#     def head(self, *args, **kwargs):
#     def get(self, *args, **kwargs):
#     def __init__(self, headers, body=None,
#                  version=1.0, status=200, reason="Ok"):
#     def getheaders(self):
#     def getheader(self, key, default):
#     def read(self, amt):
#     def __init__(self, headers=None, body=None,
#                  version=1.0, status_code=200, reason="Ok"):
#     def status(self):
#     def ok(self):
#     def read(self, amt):
#     def close(self):
#     def content(self):
#     def text(self):
#     def json(self, **kwargs):
#     def iter_content(self, chunk_size=1, decode_unicode=False):
#     def release_conn(self, **kwargs):
#     def isatty(self):
#     def write(self, data):
#     def isatty(self):
# def sort_url_by_query_keys(url):
# def build_call_record(method, url, headers, data):
#     TEST_REQUEST_BASE = {
#         'config': {'danger_mode': False},
#         'verify': True}


import io
import sys
import requests
import testtools
from glanceclient.common import progressbar
from glanceclient.common import utils
from glanceclient.tests import utils as test_utils

class TestProgressBarWrapper(testtools.TestCase):

    def test_iter_iterator_display_progress_bar(self):
        size = 100
        # create fake response object to return request-id with iterator
        resp = requests.Response()
        resp.headers['x-openstack-request-id'] = 'req-1234'
        iterator_with_len = utils.IterableWithLength(iter('X' * 100), size)
        requestid_proxy = utils.RequestIdProxy((iterator_with_len, resp))
        saved_stdout = sys.stdout
        try:
            sys.stdout = output = test_utils.FakeTTYStdout()
            # Consume iterator.
            data = list(progressbar.VerboseIteratorWrapper(requestid_proxy,
                                                           size))
            self.assertEqual(['X'] * 100, data)
            self.assertEqual(
                '[%s>] 100%%\n' % ('=' * 29),
                output.getvalue()
            )
        finally:
            sys.stdout = saved_stdout

    def test_iter_file_display_progress_bar(self):
        size = 98304
        file_obj = io.StringIO('X' * size)
        saved_stdout = sys.stdout
        try:
            sys.stdout = output = test_utils.FakeTTYStdout()

# Path: glanceclient/common/progressbar.py
# class _ProgressBarBase(object):
# class VerboseFileWrapper(_ProgressBarBase):
# class VerboseIteratorWrapper(_ProgressBarBase):
#     def __init__(self, wrapped, totalsize):
#     def _display_progress_bar(self, size_read):
#     def __getattr__(self, attr):
#     def read(self, *args, **kwargs):
#     def __iter__(self):
#     def next(self):
# 
# Path: glanceclient/common/utils.py
# SENSITIVE_HEADERS = ('X-Auth-Token', )
# REQUIRED_FIELDS_ON_DATA = ('disk_format', 'container_format')
# def arg(*args, **kwargs):
#     def _decorator(func):
# def on_data_require_fields(data_fields, required=REQUIRED_FIELDS_ON_DATA):
#     def args_decorator(func):
#         def prepare_fields(fields):
#         def func_wrapper(gc, args):
# def schema_args(schema_getter, omit=None):
#     def _decorator(func):
# def pretty_choice_list(l):
# def has_version(client, version):
# def print_cached_images(cached_images):
# def print_dict_list(objects, fields):
# def print_list(objs, fields, formatters=None, field_settings=None):
# def _encode(src):
# def unicode_key_value_to_string(src):
# def print_dict(d, max_column_width=80):
# def find_resource(manager, name_or_id):
# def env(*vars, **kwargs):
# def exit(msg='', exit_code=1):
# def print_err(msg):
# def save_image(data, path):
# def make_size_human_readable(size):
# def get_file_size(file_obj):
# def get_data_file(args):
# def strip_version(endpoint):
# def print_image(image_obj, human_readable=False, max_col_width=None):
# def integrity_iter(iter, checksum):
# def serious_integrity_iter(iter, hasher, hash_value):
# def memoized_property(fn):
#     def _memoized_property(self):
# def safe_header(name, value):
# def endpoint_version_from_url(endpoint, default_version=None):
# def debug_enabled(argv):
#     def __init__(self, iterable, length):
#     def __iter__(self):
#     def next(self):
#     def __len__(self):
#     def __init__(self, wrapped):
#     def request_ids(self):
#     def wrapped(self):
#     def next(self):
#     def __init__(self, wrapped):
#     def _set_request_ids(self, resp):
#     def _next(self):
#     def next(self):
#     def __next__(self):
#     def __iter__(self):
#     def request_ids(self):
#     def wrapped(self):
# def add_req_id_to_object():
#     def inner(wrapped, instance, args, kwargs):
# def add_req_id_to_generator():
#     def inner(wrapped, instance, args, kwargs):
# def _extract_request_id(resp):
# class IterableWithLength(object):
# class RequestIdProxy(wrapt.ObjectProxy):
# class GeneratorProxy(wrapt.ObjectProxy):
# 
# Path: glanceclient/tests/utils.py
# class FakeAPI(object):
# class FakeSchemaAPI(FakeAPI):
# class RawRequest(object):
# class FakeResponse(object):
# class TestCase(testtools.TestCase):
# class FakeTTYStdout(io.StringIO):
# class FakeNoTTYStdout(FakeTTYStdout):
#     def __init__(self, fixtures):
#     def _request(self, method, url, headers=None, data=None,
#                  content_length=None):
#     def get(self, *args, **kwargs):
#     def post(self, *args, **kwargs):
#     def put(self, *args, **kwargs):
#     def patch(self, *args, **kwargs):
#     def delete(self, *args, **kwargs):
#     def head(self, *args, **kwargs):
#     def get(self, *args, **kwargs):
#     def __init__(self, headers, body=None,
#                  version=1.0, status=200, reason="Ok"):
#     def getheaders(self):
#     def getheader(self, key, default):
#     def read(self, amt):
#     def __init__(self, headers=None, body=None,
#                  version=1.0, status_code=200, reason="Ok"):
#     def status(self):
#     def ok(self):
#     def read(self, amt):
#     def close(self):
#     def content(self):
#     def text(self):
#     def json(self, **kwargs):
#     def iter_content(self, chunk_size=1, decode_unicode=False):
#     def release_conn(self, **kwargs):
#     def isatty(self):
#     def write(self, data):
#     def isatty(self):
# def sort_url_by_query_keys(url):
# def build_call_record(method, url, headers, data):
#     TEST_REQUEST_BASE = {
#         'config': {'danger_mode': False},
#         'verify': True}


# Path: glanceclient/tests/unit/test_progressbar.py
import io
import sys
import requests
import testtools
from glanceclient.common import progressbar
from glanceclient.common import utils
from glanceclient.tests import utils as test_utils


class TestProgressBarWrapper(testtools.TestCase):

    def test_iter_iterator_display_progress_bar(self):
        size = 100
        # create fake response object to return request-id with iterator
        resp = requests.Response()
        resp.headers['x-openstack-request-id'] = 'req-1234'
        iterator_with_len = utils.IterableWithLength(iter('X' * 100), size)
        requestid_proxy = utils.RequestIdProxy((iterator_with_len, resp))
        saved_stdout = sys.stdout
        try:
            sys.stdout = output = test_utils.FakeTTYStdout()
            # Consume iterator.
            data = list(progressbar.VerboseIteratorWrapper(requestid_proxy,
                                                           size))
            self.assertEqual(['X'] * 100, data)
            self.assertEqual(
                '[%s>] 100%%\n' % ('=' * 29),
                output.getvalue()
            )
        finally:
            sys.stdout = saved_stdout

    def test_iter_file_display_progress_bar(self):
        size = 98304
        file_obj = io.StringIO('X' * size)
        saved_stdout = sys.stdout
        try:
            sys.stdout = output = test_utils.FakeTTYStdout()

            file_obj = progressbar.VerboseFileWrapper(file_obj, size)