Merge "Add verbose output to all stat commands"
This commit is contained in:
Jenkins
Gerrit Code Review
commit
04e0cb2783
126
bin/swift
126
bin/swift
@ -33,7 +33,8 @@ except ImportError:
|
||||
import json
|
||||
|
||||
from swiftclient import Connection, HTTPException
|
||||
from swiftclient.utils import config_true_value
|
||||
from swiftclient import command_helpers
|
||||
from swiftclient.utils import config_true_value, prt_bytes
|
||||
from swiftclient.multithreading import MultiThreadingManager
|
||||
from swiftclient.exceptions import ClientException
|
||||
from swiftclient import __version__ as client_version
|
||||
@ -457,34 +458,6 @@ def st_download(parser, args, thread_manager):
|
||||
for obj in args[1:]:
|
||||
object_queue.put((args[0], obj))
|
||||
|
||||
|
||||
def prt_bytes(bytes, human_flag):
|
||||
"""
|
||||
convert a number > 1024 to printable format, either in 4 char -h format as
|
||||
with ls -lh or return as 12 char right justified string
|
||||
"""
|
||||
|
||||
if human_flag:
|
||||
suffix = ''
|
||||
mods = 'KMGTPEZY'
|
||||
temp = float(bytes)
|
||||
if temp > 0:
|
||||
while (temp > 1023):
|
||||
temp /= 1024.0
|
||||
suffix = mods[0]
|
||||
mods = mods[1:]
|
||||
if suffix != '':
|
||||
if temp >= 10:
|
||||
bytes = '%3d%s' % (temp, suffix)
|
||||
else:
|
||||
bytes = '%.1f%s' % (temp, suffix)
|
||||
if suffix == '': # must be < 1024
|
||||
bytes = '%4s' % bytes
|
||||
else:
|
||||
bytes = '%12s' % bytes
|
||||
|
||||
return(bytes)
|
||||
|
||||
st_list_options = '''[--long] [--lh] [--totals]
|
||||
[--container-threads <threads>]
|
||||
'''
|
||||
@ -630,34 +603,7 @@ def st_stat(parser, args, thread_manager):
|
||||
conn = get_conn(options)
|
||||
if not args:
|
||||
try:
|
||||
headers = conn.head_account()
|
||||
if options.verbose > 1:
|
||||
thread_manager.print_msg('''
|
||||
StorageURL: %s
|
||||
Auth Token: %s
|
||||
'''.strip('\n'), conn.url, conn.token)
|
||||
container_count = int(headers.get('x-account-container-count', 0))
|
||||
object_count = prt_bytes(headers.get('x-account-object-count', 0),
|
||||
options.human).lstrip()
|
||||
bytes_used = prt_bytes(headers.get('x-account-bytes-used', 0),
|
||||
options.human).lstrip()
|
||||
thread_manager.print_msg('''
|
||||
Account: %s
|
||||
Containers: %d
|
||||
Objects: %s
|
||||
Bytes: %s'''.strip('\n'), conn.url.rsplit('/', 1)[-1], container_count,
|
||||
object_count, bytes_used)
|
||||
for key, value in headers.items():
|
||||
if key.startswith('x-account-meta-'):
|
||||
thread_manager.print_msg(
|
||||
'%10s: %s',
|
||||
'Meta %s' % key[len('x-account-meta-'):].title(),
|
||||
value)
|
||||
for key, value in headers.items():
|
||||
if not key.startswith('x-account-meta-') and key not in (
|
||||
'content-length', 'date', 'x-account-container-count',
|
||||
'x-account-object-count', 'x-account-bytes-used'):
|
||||
thread_manager.print_msg('%10s: %s', key.title(), value)
|
||||
command_helpers.stat_account(conn, options, thread_manager)
|
||||
except ClientException as err:
|
||||
if err.http_status != 404:
|
||||
raise
|
||||
@ -668,75 +614,15 @@ Containers: %d
|
||||
'meant %r instead of %r.' % \
|
||||
(args[0].replace('/', ' ', 1), args[0])
|
||||
try:
|
||||
headers = conn.head_container(args[0])
|
||||
object_count = prt_bytes(
|
||||
headers.get('x-container-object-count', 0),
|
||||
options.human).lstrip()
|
||||
bytes_used = prt_bytes(headers.get('x-container-bytes-used', 0),
|
||||
options.human).lstrip()
|
||||
thread_manager.print_msg('''
|
||||
Account: %s
|
||||
Container: %s
|
||||
Objects: %s
|
||||
Bytes: %s
|
||||
Read ACL: %s
|
||||
Write ACL: %s
|
||||
Sync To: %s
|
||||
Sync Key: %s'''.strip('\n'), conn.url.rsplit('/', 1)[-1], args[0],
|
||||
object_count, bytes_used,
|
||||
headers.get('x-container-read', ''),
|
||||
headers.get('x-container-write', ''),
|
||||
headers.get('x-container-sync-to', ''),
|
||||
headers.get('x-container-sync-key', ''))
|
||||
for key, value in headers.items():
|
||||
if key.startswith('x-container-meta-'):
|
||||
thread_manager.print_msg(
|
||||
'%9s: %s',
|
||||
'Meta %s' % key[len('x-container-meta-'):].title(),
|
||||
value)
|
||||
for key, value in headers.items():
|
||||
if not key.startswith('x-container-meta-') and key not in (
|
||||
'content-length', 'date', 'x-container-object-count',
|
||||
'x-container-bytes-used', 'x-container-read',
|
||||
'x-container-write', 'x-container-sync-to',
|
||||
'x-container-sync-key'):
|
||||
thread_manager.print_msg('%9s: %s', key.title(), value)
|
||||
command_helpers.stat_container(conn, options, args,
|
||||
thread_manager)
|
||||
except ClientException as err:
|
||||
if err.http_status != 404:
|
||||
raise
|
||||
thread_manager.error('Container %r not found', args[0])
|
||||
elif len(args) == 2:
|
||||
try:
|
||||
headers = conn.head_object(args[0], args[1])
|
||||
thread_manager.print_msg('''
|
||||
Account: %s
|
||||
Container: %s
|
||||
Object: %s
|
||||
Content Type: %s'''.strip('\n'), conn.url.rsplit('/', 1)[-1], args[0],
|
||||
args[1], headers.get('content-type'))
|
||||
if 'content-length' in headers:
|
||||
thread_manager.print_msg('Content Length: %s',
|
||||
prt_bytes(headers['content-length'],
|
||||
options.human).lstrip())
|
||||
if 'last-modified' in headers:
|
||||
thread_manager.print_msg(' Last Modified: %s',
|
||||
headers['last-modified'])
|
||||
if 'etag' in headers:
|
||||
thread_manager.print_msg(' ETag: %s', headers['etag'])
|
||||
if 'x-object-manifest' in headers:
|
||||
thread_manager.print_msg(' Manifest: %s',
|
||||
headers['x-object-manifest'])
|
||||
for key, value in headers.items():
|
||||
if key.startswith('x-object-meta-'):
|
||||
thread_manager.print_msg(
|
||||
'%14s: %s',
|
||||
'Meta %s' % key[len('x-object-meta-'):].title(),
|
||||
value)
|
||||
for key, value in headers.items():
|
||||
if not key.startswith('x-object-meta-') and key not in (
|
||||
'content-type', 'content-length', 'last-modified',
|
||||
'etag', 'date', 'x-object-manifest'):
|
||||
thread_manager.print_msg('%14s: %s', key.title(), value)
|
||||
command_helpers.stat_object(conn, options, args, thread_manager)
|
||||
except ClientException as err:
|
||||
if err.http_status != 404:
|
||||
raise
|
||||
|
||||
91
swiftclient/command_helpers.py
Normal file
91
swiftclient/command_helpers.py
Normal file
@ -0,0 +1,91 @@
|
||||
from swiftclient.utils import prt_bytes
|
||||
|
||||
|
||||
def stat_account(conn, options, thread_manager):
|
||||
headers = conn.head_account()
|
||||
if options.verbose > 1:
|
||||
thread_manager.print_items((
|
||||
('StorageURL', conn.url),
|
||||
('Auth Token', conn.token),
|
||||
))
|
||||
container_count = int(headers.get('x-account-container-count', 0))
|
||||
object_count = prt_bytes(headers.get('x-account-object-count', 0),
|
||||
options.human).lstrip()
|
||||
bytes_used = prt_bytes(headers.get('x-account-bytes-used', 0),
|
||||
options.human).lstrip()
|
||||
thread_manager.print_items((
|
||||
('Account', conn.url.rsplit('/', 1)[-1]),
|
||||
('Containers', container_count),
|
||||
('Objects', object_count),
|
||||
('Bytes', bytes_used),
|
||||
))
|
||||
thread_manager.print_headers(headers,
|
||||
meta_prefix='x-account-meta-',
|
||||
exclude_headers=(
|
||||
'content-length', 'date',
|
||||
'x-account-container-count',
|
||||
'x-account-object-count',
|
||||
'x-account-bytes-used'))
|
||||
|
||||
|
||||
def stat_container(conn, options, args, thread_manager):
|
||||
headers = conn.head_container(args[0])
|
||||
if options.verbose > 1:
|
||||
path = '%s/%s' % (conn.url, args[0])
|
||||
thread_manager.print_items((
|
||||
('URL', path),
|
||||
('Auth Token', conn.token),
|
||||
))
|
||||
object_count = prt_bytes(
|
||||
headers.get('x-container-object-count', 0),
|
||||
options.human).lstrip()
|
||||
bytes_used = prt_bytes(headers.get('x-container-bytes-used', 0),
|
||||
options.human).lstrip()
|
||||
thread_manager.print_items((
|
||||
('Account', conn.url.rsplit('/', 1)[-1]),
|
||||
('Container', args[0]),
|
||||
('Objects', object_count),
|
||||
('Bytes', bytes_used),
|
||||
('Read ACL', headers.get('x-container-read', '')),
|
||||
('Write ACL', headers.get('x-container-write', '')),
|
||||
('Sync To', headers.get('x-container-sync-to', '')),
|
||||
('Sync Key', headers.get('x-container-sync-key', '')),
|
||||
))
|
||||
thread_manager.print_headers(headers,
|
||||
meta_prefix='x-container-meta-',
|
||||
exclude_headers=(
|
||||
'content-length', 'date',
|
||||
'x-container-object-count',
|
||||
'x-container-bytes-used',
|
||||
'x-container-read',
|
||||
'x-container-write',
|
||||
'x-container-sync-to',
|
||||
'x-container-sync-key'))
|
||||
|
||||
|
||||
def stat_object(conn, options, args, thread_manager):
|
||||
headers = conn.head_object(args[0], args[1])
|
||||
if options.verbose > 1:
|
||||
path = '%s/%s/%s' % (conn.url, args[0], args[1])
|
||||
thread_manager.print_items((
|
||||
('URL', path),
|
||||
('Auth Token', conn.token),
|
||||
))
|
||||
content_length = prt_bytes(headers.get('content-length', 0),
|
||||
options.human).lstrip()
|
||||
thread_manager.print_items((
|
||||
('Account', conn.url.rsplit('/', 1)[-1]),
|
||||
('Container', args[0]),
|
||||
('Object', args[1]),
|
||||
('Content Type', headers.get('content-type')),
|
||||
('Content Length', content_length),
|
||||
('Last Modified', headers.get('last-modified')),
|
||||
('ETag', headers.get('etag')),
|
||||
('Manifest', headers.get('x-object-manifest')),
|
||||
), skip_missing=True)
|
||||
thread_manager.print_headers(headers,
|
||||
meta_prefix='x-object-meta-',
|
||||
exclude_headers=(
|
||||
'content-type', 'content-length',
|
||||
'last-modified', 'etag', 'date',
|
||||
'x-object-manifest'))
|
||||
@ -12,6 +12,7 @@
|
||||
# implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
from itertools import chain
|
||||
import sys
|
||||
from time import sleep
|
||||
from Queue import Queue
|
||||
@ -224,6 +225,29 @@ class MultiThreadingManager(object):
|
||||
msg = msg % fmt_args
|
||||
self.printer.queue.put(msg)
|
||||
|
||||
def print_items(self, items, offset=14, skip_missing=False):
|
||||
lines = []
|
||||
template = '%%%ds: %%s' % offset
|
||||
for k, v in items:
|
||||
if skip_missing and not v:
|
||||
continue
|
||||
lines.append((template % (k, v)).rstrip())
|
||||
self.print_msg('\n'.join(lines))
|
||||
|
||||
def print_headers(self, headers, meta_prefix='', exclude_headers=None,
|
||||
offset=14):
|
||||
exclude_headers = exclude_headers or []
|
||||
meta_headers = []
|
||||
other_headers = []
|
||||
template = '%%%ds: %%s' % offset
|
||||
for key, value in headers.items():
|
||||
if key.startswith(meta_prefix):
|
||||
meta_key = 'Meta %s' % key[len(meta_prefix):].title()
|
||||
meta_headers.append(template % (meta_key, value))
|
||||
elif key not in exclude_headers:
|
||||
other_headers.append(template % (key.title(), value))
|
||||
self.print_msg('\n'.join(chain(meta_headers, other_headers)))
|
||||
|
||||
def error(self, msg, *fmt_args):
|
||||
if fmt_args:
|
||||
msg = msg % fmt_args
|
||||
|
||||
@ -25,3 +25,33 @@ def config_true_value(value):
|
||||
"""
|
||||
return value is True or \
|
||||
(isinstance(value, basestring) and value.lower() in TRUE_VALUES)
|
||||
|
||||
|
||||
def prt_bytes(bytes, human_flag):
|
||||
"""
|
||||
convert a number > 1024 to printable format, either in 4 char -h format as
|
||||
with ls -lh or return as 12 char right justified string
|
||||
"""
|
||||
|
||||
if human_flag:
|
||||
suffix = ''
|
||||
mods = list('KMGTPEZY')
|
||||
temp = float(bytes)
|
||||
if temp > 0:
|
||||
while (temp > 1023):
|
||||
try:
|
||||
suffix = mods.pop(0)
|
||||
except IndexError:
|
||||
break
|
||||
temp /= 1024.0
|
||||
if suffix != '':
|
||||
if temp >= 10:
|
||||
bytes = '%3d%s' % (temp, suffix)
|
||||
else:
|
||||
bytes = '%.1f%s' % (temp, suffix)
|
||||
if suffix == '': # must be < 1024
|
||||
bytes = '%4s' % bytes
|
||||
else:
|
||||
bytes = '%12s' % bytes
|
||||
|
||||
return(bytes)
|
||||
|
||||
193
tests/test_command_helpers.py
Normal file
193
tests/test_command_helpers.py
Normal file
@ -0,0 +1,193 @@
|
||||
# Copyright (c) 2010-2013 OpenStack, LLC.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
# implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from StringIO import StringIO
|
||||
import mock
|
||||
import testtools
|
||||
|
||||
from swiftclient import command_helpers as h
|
||||
from swiftclient.multithreading import MultiThreadingManager
|
||||
|
||||
|
||||
class TestStatHelpers(testtools.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(TestStatHelpers, self).setUp()
|
||||
conn_attrs = {
|
||||
'url': 'http://storage/v1/a',
|
||||
'token': 'tk12345',
|
||||
}
|
||||
self.conn = mock.MagicMock(**conn_attrs)
|
||||
self.options = mock.MagicMock(human=False, verbose=1)
|
||||
self.stdout = StringIO()
|
||||
self.stderr = StringIO()
|
||||
self.thread_manager = MultiThreadingManager(self.stdout, self.stderr)
|
||||
|
||||
def assertOut(self, expected):
|
||||
real = self.stdout.getvalue()
|
||||
# commonly if we strip of blank lines we have a match
|
||||
try:
|
||||
self.assertEqual(expected.strip('\n'),
|
||||
real.strip('\n'))
|
||||
except AssertionError:
|
||||
# could be anything, try to find typos line by line
|
||||
expected_lines = [line.lstrip() for line in
|
||||
expected.splitlines() if line.strip()]
|
||||
real_lines = [line.lstrip() for line in
|
||||
real.splitlines() if line.strip()]
|
||||
for expected, real in zip(expected_lines, real_lines):
|
||||
self.assertEqual(expected, real)
|
||||
# not a typo, might be an indent thing, hopefully you can spot it
|
||||
raise
|
||||
|
||||
def test_stat_account_human(self):
|
||||
self.options.human = True
|
||||
# stub head_account
|
||||
stub_headers = {
|
||||
'x-account-container-count': 42,
|
||||
'x-account-object-count': 1000000,
|
||||
'x-account-bytes-used': 2 ** 30,
|
||||
}
|
||||
self.conn.head_account.return_value = stub_headers
|
||||
|
||||
with self.thread_manager as thread_manager:
|
||||
h.stat_account(self.conn, self.options, thread_manager)
|
||||
expected = """
|
||||
Account: a
|
||||
Containers: 42
|
||||
Objects: 976K
|
||||
Bytes: 1.0G
|
||||
"""
|
||||
self.assertOut(expected)
|
||||
|
||||
def test_stat_account_verbose(self):
|
||||
self.options.verbose += 1
|
||||
# stub head_account
|
||||
stub_headers = {
|
||||
'x-account-container-count': 42,
|
||||
'x-account-object-count': 1000000,
|
||||
'x-account-bytes-used': 2 ** 30,
|
||||
}
|
||||
self.conn.head_account.return_value = stub_headers
|
||||
|
||||
with self.thread_manager as thread_manager:
|
||||
h.stat_account(self.conn, self.options, thread_manager)
|
||||
expected = """
|
||||
StorageURL: http://storage/v1/a
|
||||
Auth Token: tk12345
|
||||
Account: a
|
||||
Containers: 42
|
||||
Objects: 1000000
|
||||
Bytes: 1073741824
|
||||
"""
|
||||
self.assertOut(expected)
|
||||
|
||||
def test_stat_container_human(self):
|
||||
self.options.human = True
|
||||
# stub head container request
|
||||
stub_headers = {
|
||||
'x-container-object-count': 10 ** 6,
|
||||
'x-container-bytes-used': 2 ** 30,
|
||||
}
|
||||
self.conn.head_container.return_value = stub_headers
|
||||
args = ('c',)
|
||||
with self.thread_manager as thread_manager:
|
||||
h.stat_container(self.conn, self.options, args, thread_manager)
|
||||
expected = """
|
||||
Account: a
|
||||
Container: c
|
||||
Objects: 976K
|
||||
Bytes: 1.0G
|
||||
Read ACL:
|
||||
Write ACL:
|
||||
Sync To:
|
||||
Sync Key:
|
||||
"""
|
||||
self.assertOut(expected)
|
||||
|
||||
def test_stat_container_verbose(self):
|
||||
self.options.verbose += 1
|
||||
# stub head container request
|
||||
stub_headers = {
|
||||
'x-container-object-count': 10 ** 6,
|
||||
'x-container-bytes-used': 2 ** 30,
|
||||
}
|
||||
self.conn.head_container.return_value = stub_headers
|
||||
args = ('c',)
|
||||
with self.thread_manager as thread_manager:
|
||||
h.stat_container(self.conn, self.options, args, thread_manager)
|
||||
expected = """
|
||||
URL: http://storage/v1/a/c
|
||||
Auth Token: tk12345
|
||||
Account: a
|
||||
Container: c
|
||||
Objects: 1000000
|
||||
Bytes: 1073741824
|
||||
Read ACL:
|
||||
Write ACL:
|
||||
Sync To:
|
||||
Sync Key:
|
||||
"""
|
||||
self.assertOut(expected)
|
||||
|
||||
def test_stat_object_human(self):
|
||||
self.options.human = True
|
||||
# stub head object request
|
||||
stub_headers = {
|
||||
'content-length': 2 ** 20,
|
||||
'x-object-meta-color': 'blue',
|
||||
'etag': '68b329da9893e34099c7d8ad5cb9c940',
|
||||
'content-encoding': 'gzip',
|
||||
}
|
||||
self.conn.head_object.return_value = stub_headers
|
||||
args = ('c', 'o')
|
||||
with self.thread_manager as thread_manager:
|
||||
h.stat_object(self.conn, self.options, args, thread_manager)
|
||||
expected = """
|
||||
Account: a
|
||||
Container: c
|
||||
Object: o
|
||||
Content Length: 1.0M
|
||||
ETag: 68b329da9893e34099c7d8ad5cb9c940
|
||||
Meta Color: blue
|
||||
Content-Encoding: gzip
|
||||
"""
|
||||
self.assertOut(expected)
|
||||
|
||||
def test_stat_object_verbose(self):
|
||||
self.options.verbose += 1
|
||||
# stub head object request
|
||||
stub_headers = {
|
||||
'content-length': 2 ** 20,
|
||||
'x-object-meta-color': 'blue',
|
||||
'etag': '68b329da9893e34099c7d8ad5cb9c940',
|
||||
'content-encoding': 'gzip',
|
||||
}
|
||||
self.conn.head_object.return_value = stub_headers
|
||||
args = ('c', 'o')
|
||||
with self.thread_manager as thread_manager:
|
||||
h.stat_object(self.conn, self.options, args, thread_manager)
|
||||
expected = """
|
||||
URL: http://storage/v1/a/c/o
|
||||
Auth Token: tk12345
|
||||
Account: a
|
||||
Container: c
|
||||
Object: o
|
||||
Content Length: 1048576
|
||||
ETag: 68b329da9893e34099c7d8ad5cb9c940
|
||||
Meta Color: blue
|
||||
Content-Encoding: gzip
|
||||
"""
|
||||
self.assertOut(expected)
|
||||
@ -27,7 +27,6 @@ from urlparse import urlparse
|
||||
from .utils import fake_http_connect, fake_get_keystoneclient_2_0
|
||||
|
||||
from swiftclient import client as c
|
||||
from swiftclient import utils as u
|
||||
|
||||
|
||||
class TestClientException(testtools.TestCase):
|
||||
@ -97,25 +96,6 @@ class TestJsonImport(testtools.TestCase):
|
||||
self.assertEqual(loads, c.json_loads)
|
||||
|
||||
|
||||
class TestConfigTrueValue(testtools.TestCase):
|
||||
|
||||
def test_TRUE_VALUES(self):
|
||||
for v in u.TRUE_VALUES:
|
||||
self.assertEqual(v, v.lower())
|
||||
|
||||
def test_config_true_value(self):
|
||||
orig_trues = u.TRUE_VALUES
|
||||
try:
|
||||
u.TRUE_VALUES = 'hello world'.split()
|
||||
for val in 'hello world HELLO WORLD'.split():
|
||||
self.assertTrue(u.config_true_value(val) is True)
|
||||
self.assertTrue(u.config_true_value(True) is True)
|
||||
self.assertTrue(u.config_true_value('foo') is False)
|
||||
self.assertTrue(u.config_true_value(False) is False)
|
||||
finally:
|
||||
u.TRUE_VALUES = orig_trues
|
||||
|
||||
|
||||
class MockHttpTest(testtools.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
|
||||
119
tests/test_utils.py
Normal file
119
tests/test_utils.py
Normal file
@ -0,0 +1,119 @@
|
||||
# Copyright (c) 2010-2013 OpenStack, LLC.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
# implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import testtools
|
||||
|
||||
from swiftclient import utils as u
|
||||
|
||||
|
||||
class TestConfigTrueValue(testtools.TestCase):
|
||||
|
||||
def test_TRUE_VALUES(self):
|
||||
for v in u.TRUE_VALUES:
|
||||
self.assertEqual(v, v.lower())
|
||||
|
||||
def test_config_true_value(self):
|
||||
orig_trues = u.TRUE_VALUES
|
||||
try:
|
||||
u.TRUE_VALUES = 'hello world'.split()
|
||||
for val in 'hello world HELLO WORLD'.split():
|
||||
self.assertTrue(u.config_true_value(val) is True)
|
||||
self.assertTrue(u.config_true_value(True) is True)
|
||||
self.assertTrue(u.config_true_value('foo') is False)
|
||||
self.assertTrue(u.config_true_value(False) is False)
|
||||
finally:
|
||||
u.TRUE_VALUES = orig_trues
|
||||
|
||||
|
||||
class TestPrtBytes(testtools.TestCase):
|
||||
|
||||
def test_zero_bytes(self):
|
||||
bytes_ = 0
|
||||
raw = '0'
|
||||
human = '0'
|
||||
self.assertEquals(raw, u.prt_bytes(bytes_, False).lstrip())
|
||||
self.assertEquals(human, u.prt_bytes(bytes_, True).lstrip())
|
||||
|
||||
def test_one_byte(self):
|
||||
bytes_ = 1
|
||||
raw = '1'
|
||||
human = '1'
|
||||
self.assertEquals(raw, u.prt_bytes(bytes_, False).lstrip())
|
||||
self.assertEquals(human, u.prt_bytes(bytes_, True).lstrip())
|
||||
|
||||
def test_less_than_one_k(self):
|
||||
bytes_ = (2 ** 10) - 1
|
||||
raw = '1023'
|
||||
human = '1023'
|
||||
self.assertEquals(raw, u.prt_bytes(bytes_, False).lstrip())
|
||||
self.assertEquals(human, u.prt_bytes(bytes_, True).lstrip())
|
||||
|
||||
def test_one_k(self):
|
||||
bytes_ = 2 ** 10
|
||||
raw = '1024'
|
||||
human = '1.0K'
|
||||
self.assertEquals(raw, u.prt_bytes(bytes_, False).lstrip())
|
||||
self.assertEquals(human, u.prt_bytes(bytes_, True).lstrip())
|
||||
|
||||
def test_a_decimal_k(self):
|
||||
bytes_ = (3 * 2 ** 10) + 512
|
||||
raw = '3584'
|
||||
human = '3.5K'
|
||||
self.assertEquals(raw, u.prt_bytes(bytes_, False).lstrip())
|
||||
self.assertEquals(human, u.prt_bytes(bytes_, True).lstrip())
|
||||
|
||||
def test_a_bit_less_than_one_meg(self):
|
||||
bytes_ = (2 ** 20) - (2 ** 10)
|
||||
raw = '1047552'
|
||||
human = '1023K'
|
||||
self.assertEquals(raw, u.prt_bytes(bytes_, False).lstrip())
|
||||
self.assertEquals(human, u.prt_bytes(bytes_, True).lstrip())
|
||||
|
||||
def test_just_a_hair_less_than_one_meg(self):
|
||||
bytes_ = (2 ** 20) - (2 ** 10) + 1
|
||||
raw = '1047553'
|
||||
human = '1.0M'
|
||||
self.assertEquals(raw, u.prt_bytes(bytes_, False).lstrip())
|
||||
self.assertEquals(human, u.prt_bytes(bytes_, True).lstrip())
|
||||
|
||||
def test_one_meg(self):
|
||||
bytes_ = 2 ** 20
|
||||
raw = '1048576'
|
||||
human = '1.0M'
|
||||
self.assertEquals(raw, u.prt_bytes(bytes_, False).lstrip())
|
||||
self.assertEquals(human, u.prt_bytes(bytes_, True).lstrip())
|
||||
|
||||
def test_ten_meg(self):
|
||||
bytes_ = 10 * 2 ** 20
|
||||
human = '10M'
|
||||
self.assertEquals(human, u.prt_bytes(bytes_, True).lstrip())
|
||||
|
||||
def test_bit_less_than_ten_meg(self):
|
||||
bytes_ = (10 * 2 ** 20) - (100 * 2 ** 10)
|
||||
human = '9.9M'
|
||||
self.assertEquals(human, u.prt_bytes(bytes_, True).lstrip())
|
||||
|
||||
def test_just_a_hair_less_than_ten_meg(self):
|
||||
bytes_ = (10 * 2 ** 20) - 1
|
||||
human = '10.0M'
|
||||
self.assertEquals(human, u.prt_bytes(bytes_, True).lstrip())
|
||||
|
||||
def test_a_yotta(self):
|
||||
bytes_ = 42 * 2 ** 80
|
||||
self.assertEquals('42Y', u.prt_bytes(bytes_, True).lstrip())
|
||||
|
||||
def test_overflow(self):
|
||||
bytes_ = 2 ** 90
|
||||
self.assertEquals('1024Y', u.prt_bytes(bytes_, True).lstrip())
|
||||
Loading…
Reference in New Issue
Block a user