ExternalVendorCode/tahoe-lafs
1
0
Fork 0
mirror of https://github.com/tahoe-lafs/tahoe-lafs.git synced 2025-04-25 05:19:53 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Remove use of the mock module from test_status

Browse Source
This commit is contained in:
Jean-Paul Calderone 2021-08-12 16:50:01 -04:00
parent 85ba6567ba
commit acc8cbd28b
2 changed files with 86 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
src/allmydata/scripts/tahoe_status.py
View File

@ -17,15 +17,17 @@ import json
from .common import BaseOptions from .common import BaseOptions
from allmydata.scripts.common import get_default_nodedir from allmydata.scripts.common import get_default_nodedir
from allmydata.scripts.common_http import do_http, BadResponse from allmydata.scripts.common_http import BadResponse
from allmydata.util.abbreviate import abbreviate_space, abbreviate_time from allmydata.util.abbreviate import abbreviate_space, abbreviate_time
from allmydata.util.encodingutil import argv_to_abspath from allmydata.util.encodingutil import argv_to_abspath
def _get_json_for_fragment(options, fragment, method='GET', post_args=None): def _get_request_parameters_for_fragment(options, fragment, method, post_args):
""" """
returns the JSON for a particular URI-fragment (to which is Get parameters for ``do_http`` for requesting the given fragment.
pre-pended the node's URL)
:return dict: A dictionary suitable for use as keyword arguments to
``do_http``.
""" """
nodeurl = options['node-url'] nodeurl = options['node-url']
if nodeurl.endswith('/'): if nodeurl.endswith('/'):
@ -40,7 +42,17 @@ def _get_json_for_fragment(options, fragment, method='GET', post_args=None):
body = '' body = ''
if post_args is not None: if post_args is not None:
raise ValueError("post_args= only valid for POST method") raise ValueError("post_args= only valid for POST method")
resp = do_http(method, url, body=body.encode("utf-8")) return dict(
method=method,
url=url,
body=body.encode("utf-8"),
)
def _handle_response_for_fragment(resp, nodeurl):
"""
Inspect an HTTP response and return the parsed payload, if possible.
"""
if isinstance(resp, BadResponse): if isinstance(resp, BadResponse):
# specifically NOT using format_http_error() here because the # specifically NOT using format_http_error() here because the
# URL is pretty sensitive (we're doing /uri/<key>). # URL is pretty sensitive (we're doing /uri/<key>).
@ -55,12 +67,6 @@ def _get_json_for_fragment(options, fragment, method='GET', post_args=None):
return parsed return parsed
def _get_json_for_cap(options, cap):
return _get_json_for_fragment(
options,
'uri/%s?t=json' % url_quote(cap),
)
def pretty_progress(percent, size=10, output_ascii=False): def pretty_progress(percent, size=10, output_ascii=False):
""" """
Displays a unicode or ascii based progress bar of a certain Displays a unicode or ascii based progress bar of a certain
@ -251,7 +257,10 @@ def render_recent(verbose, stdout, status_data):
print(u" Skipped {} non-upload/download operations; use --verbose to see".format(skipped), file=stdout) print(u" Skipped {} non-upload/download operations; use --verbose to see".format(skipped), file=stdout)
def do_status(options): def do_status(options, do_http=None):
if do_http is None:
from allmydata.scripts.common_http import do_http
nodedir = options["node-directory"] nodedir = options["node-directory"]
with open(os.path.join(nodedir, u'private', u'api_auth_token'), 'r') as f: with open(os.path.join(nodedir, u'private', u'api_auth_token'), 'r') as f:
token = f.read().strip() token = f.read().strip()
@ -260,25 +269,30 @@ def do_status(options):
# do *all* our data-retrievals first in case there's an error # do *all* our data-retrievals first in case there's an error
try: try:
status_data = _get_json_for_fragment( status_data = _handle_response_for_fragment(
options, do_http(**_get_request_parameters_for_fragment(
'status?t=json', options,
method='POST', 'status?t=json',
post_args=dict( method='POST',
t='json', post_args=dict(
token=token, t='json',
) token=token,
),
)),
options['node-url'],
) )
statistics_data = _get_json_for_fragment( statistics_data = _handle_response_for_fragment(
options, do_http(**_get_request_parameters_for_fragment(
'statistics?t=json', options,
method='POST', 'statistics?t=json',
post_args=dict( method='POST',
t='json', post_args=dict(
token=token, t='json',
) token=token,
),
)),
options['node-url'],
) )
except Exception as e: except Exception as e:
print(u"failed to retrieve data: %s" % str(e), file=options.stderr) print(u"failed to retrieve data: %s" % str(e), file=options.stderr)
return 2 return 2

98
src/allmydata/test/cli/test_status.py
View File

@ -12,7 +12,6 @@ if PY2:
from six import ensure_text from six import ensure_text
import os import os
import mock
import tempfile import tempfile
from io import BytesIO, StringIO from io import BytesIO, StringIO
from os.path import join from os.path import join
@ -22,8 +21,8 @@ from twisted.internet import defer
from allmydata.mutable.publish import MutableData from allmydata.mutable.publish import MutableData
from allmydata.scripts.common_http import BadResponse from allmydata.scripts.common_http import BadResponse
from allmydata.scripts.tahoe_status import _get_json_for_fragment from allmydata.scripts.tahoe_status import _handle_response_for_fragment
from allmydata.scripts.tahoe_status import _get_json_for_cap from allmydata.scripts.tahoe_status import _get_request_parameters_for_fragment
from allmydata.scripts.tahoe_status import pretty_progress from allmydata.scripts.tahoe_status import pretty_progress
from allmydata.scripts.tahoe_status import do_status from allmydata.scripts.tahoe_status import do_status
from allmydata.web.status import marshal_json from allmydata.web.status import marshal_json
@ -148,9 +147,7 @@ class CommandStatus(unittest.TestCase):
def setUp(self): def setUp(self):
self.options = _FakeOptions() self.options = _FakeOptions()
@mock.patch('allmydata.scripts.tahoe_status.do_http') def test_no_operations(self):
@mock.patch('sys.stdout', StringIO())
def test_no_operations(self, http):
values = [ values = [
StringIO(ensure_text(json.dumps({ StringIO(ensure_text(json.dumps({
"active": [], "active": [],
@ -165,12 +162,11 @@ class CommandStatus(unittest.TestCase):
} }
}))), }))),
] ]
http.side_effect = lambda *args, **kw: values.pop(0) def do_http(*args, **kw):
do_status(self.options) return values.pop(0)
do_status(self.options, do_http)
@mock.patch('allmydata.scripts.tahoe_status.do_http') def test_simple(self):
@mock.patch('sys.stdout', StringIO())
def test_simple(self, http):
recent_items = active_items = [ recent_items = active_items = [
UploadStatus(), UploadStatus(),
DownloadStatus(b"abcd", 12345), DownloadStatus(b"abcd", 12345),
@ -201,80 +197,72 @@ class CommandStatus(unittest.TestCase):
} }
}).encode("utf-8")), }).encode("utf-8")),
] ]
http.side_effect = lambda *args, **kw: values.pop(0) def do_http(*args, **kw):
do_status(self.options) return values.pop(0)
do_status(self.options, do_http)
@mock.patch('allmydata.scripts.tahoe_status.do_http') def test_fetch_error(self):
def test_fetch_error(self, http): def do_http(*args, **kw):
def boom(*args, **kw):
raise RuntimeError("boom") raise RuntimeError("boom")
http.side_effect = boom do_status(self.options, do_http)
do_status(self.options)
class JsonHelpers(unittest.TestCase): class JsonHelpers(unittest.TestCase):
@mock.patch('allmydata.scripts.tahoe_status.do_http') def test_bad_response(self):
def test_bad_response(self, http): def do_http(*args, **kw):
http.return_value = BadResponse('the url', 'some err') return
with self.assertRaises(RuntimeError) as ctx: with self.assertRaises(RuntimeError) as ctx:
_get_json_for_fragment({'node-url': 'http://localhost:1234'}, '/fragment') _handle_response_for_fragment(
self.assertTrue( BadResponse('the url', 'some err'),
"Failed to get" in str(ctx.exception) 'http://localhost:1234',
)
self.assertIn(
"Failed to get",
str(ctx.exception),
) )
@mock.patch('allmydata.scripts.tahoe_status.do_http') def test_happy_path(self):
def test_happy_path(self, http): resp = _handle_response_for_fragment(
http.return_value = StringIO('{"some": "json"}') StringIO('{"some": "json"}'),
resp = _get_json_for_fragment({'node-url': 'http://localhost:1234/'}, '/fragment/') 'http://localhost:1234/',
self.assertEqual(resp, dict(some='json'))
@mock.patch('allmydata.scripts.tahoe_status.do_http')
def test_happy_path_post(self, http):
http.return_value = StringIO('{"some": "json"}')
resp = _get_json_for_fragment(
{'node-url': 'http://localhost:1234/'},
'/fragment/',
method='POST',
post_args={'foo': 'bar'}
) )
self.assertEqual(resp, dict(some='json')) self.assertEqual(resp, dict(some='json'))
@mock.patch('allmydata.scripts.tahoe_status.do_http') def test_happy_path_post(self):
def test_happy_path_for_cap(self, http): resp = _handle_response_for_fragment(
http.return_value = StringIO('{"some": "json"}') StringIO('{"some": "json"}'),
resp = _get_json_for_cap({'node-url': 'http://localhost:1234'}, 'fake cap') 'http://localhost:1234/',
)
self.assertEqual(resp, dict(some='json')) self.assertEqual(resp, dict(some='json'))
@mock.patch('allmydata.scripts.tahoe_status.do_http') def test_no_data_returned(self):
def test_no_data_returned(self, http):
http.return_value = StringIO('null')
with self.assertRaises(RuntimeError) as ctx: with self.assertRaises(RuntimeError) as ctx:
_get_json_for_cap({'node-url': 'http://localhost:1234'}, 'fake cap') _handle_response_for_fragment(StringIO('null'), 'http://localhost:1234')
self.assertTrue('No data from' in str(ctx.exception)) self.assertIn('No data from', str(ctx.exception))
def test_no_post_args(self): def test_no_post_args(self):
with self.assertRaises(ValueError) as ctx: with self.assertRaises(ValueError) as ctx:
_get_json_for_fragment( _get_request_parameters_for_fragment(
{'node-url': 'http://localhost:1234'}, {'node-url': 'http://localhost:1234'},
'/fragment', '/fragment',
method='POST', method='POST',
post_args=None, post_args=None,
) )
self.assertTrue( self.assertIn(
"Must pass post_args" in str(ctx.exception) "Must pass post_args",
str(ctx.exception),
) )
def test_post_args_for_get(self): def test_post_args_for_get(self):
with self.assertRaises(ValueError) as ctx: with self.assertRaises(ValueError) as ctx:
_get_json_for_fragment( _get_request_parameters_for_fragment(
{'node-url': 'http://localhost:1234'}, {'node-url': 'http://localhost:1234'},
'/fragment', '/fragment',
method='GET', method='GET',
post_args={'foo': 'bar'} post_args={'foo': 'bar'}
) )
self.assertTrue( self.assertIn(
"only valid for POST" in str(ctx.exception) "only valid for POST",
str(ctx.exception),
) )