Merge pull request #1103 from LeastAuthority/3525.test_status-no-mock

Remove usage of mock from allmydata/test/cli/test_status.py

Fixes: ticket:3525

src/allmydata/scripts/tahoe_status.py

@@ -11,21 +11,48 @@ if PY2:
     from future.builtins import filter, map, zip, ascii, chr, hex, input, next, oct, open, pow, round, super, bytes, dict, list, object, range, str, max, min  # noqa: F401
 
 import os
-from urllib.parse import urlencode, quote as url_quote
+from sys import stdout as _sys_stdout
+from urllib.parse import urlencode
 
 import json
 
 from .common import BaseOptions
 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.encodingutil import argv_to_abspath
 
-
+_print = print
-def _get_json_for_fragment(options, fragment, method='GET', post_args=None):
+def print(*args, **kwargs):
     """
-    returns the JSON for a particular URI-fragment (to which is
+    Builtin ``print``-alike that will even write unicode which cannot be
-    pre-pended the node's URL)
+    encoded using the specified output file's encoding.
+
+    This differs from the builtin print in that it will use the "replace"
+    encoding error handler and then write the result whereas builtin print
+    uses the "strict" encoding error handler.
+    """
+    from past.builtins import unicode
+    out = kwargs.pop("file", None)
+    if out is None:
+        out = _sys_stdout
+    encoding = out.encoding or "ascii"
+    def ensafe(o):
+        if isinstance(o, unicode):
+            return o.encode(encoding, errors="replace").decode(encoding)
+        return o
+    return _print(
+        *(ensafe(a) for a in args),
+        file=out,
+        **kwargs
+    )
+
+def _get_request_parameters_for_fragment(options, fragment, method, post_args):
+    """
+    Get parameters for ``do_http`` for requesting the given fragment.
+
+    :return dict: A dictionary suitable for use as keyword arguments to
+        ``do_http``.
     """
     nodeurl = options['node-url']
     if nodeurl.endswith('/'):
@@ -40,7 +67,17 @@ def _get_json_for_fragment(options, fragment, method='GET', post_args=None):
         body = ''
         if post_args is not None:
             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):
         # specifically NOT using format_http_error() here because the
         # URL is pretty sensitive (we're doing /uri/<key>).
@@ -55,12 +92,6 @@ def _get_json_for_fragment(options, fragment, method='GET', post_args=None):
     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):
     """
     Displays a unicode or ascii based progress bar of a certain
@@ -251,7 +282,10 @@ def render_recent(verbose, stdout, status_data):
         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"]
     with open(os.path.join(nodedir, u'private', u'api_auth_token'), 'r') as f:
         token = f.read().strip()
@@ -260,25 +294,30 @@ def do_status(options):
 
     # do *all* our data-retrievals first in case there's an error
     try:
-        status_data = _get_json_for_fragment(
+        status_data = _handle_response_for_fragment(
+            do_http(**_get_request_parameters_for_fragment(
                 options,
                 'status?t=json',
                 method='POST',
                 post_args=dict(
                     t='json',
                     token=token,
+                ),
+            )),
+            options['node-url'],
         )
-        )
+        statistics_data = _handle_response_for_fragment(
-        statistics_data = _get_json_for_fragment(
+            do_http(**_get_request_parameters_for_fragment(
                 options,
                 'statistics?t=json',
                 method='POST',
                 post_args=dict(
                     t='json',
                     token=token,
+                ),
+            )),
+            options['node-url'],
         )
-        )
-
     except Exception as e:
         print(u"failed to retrieve data: %s" % str(e), file=options.stderr)
         return 2

src/allmydata/test/cli/test_status.py

@@ -12,7 +12,6 @@ if PY2:
 from six import ensure_text
 
 import os
-import mock
 import tempfile
 from io import BytesIO, StringIO
 from os.path import join
@@ -22,8 +21,8 @@ from twisted.internet import defer
 
 from allmydata.mutable.publish import MutableData
 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 do_status
 from allmydata.web.status import marshal_json
@@ -140,17 +139,12 @@ class CommandStatus(unittest.TestCase):
     """
     These tests just exercise the renderers and ensure they don't
     catastrophically fail.
-
-    They could be enhanced to look for "some" magic strings in the
-    results and assert they're in the output.
     """
 
     def setUp(self):
         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 = [
             StringIO(ensure_text(json.dumps({
                 "active": [],
@@ -165,12 +159,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 = [
             UploadStatus(),
             DownloadStatus(b"abcd", 12345),
@@ -201,80 +194,72 @@ class CommandStatus(unittest.TestCase):
                 }
             }).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")
-        http.side_effect = boom
+        do_status(self.options, do_http)
-        do_status(self.options)
 
 
 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:
-            _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'))
 
-    @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'))
 
-    @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:
-            _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):
         with self.assertRaises(ValueError) as ctx:
-            _get_json_for_fragment(
+            _get_request_parameters_for_fragment(
                 {'node-url': 'http://localhost:1234'},
                 '/fragment',
                 method='POST',
                 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):
         with self.assertRaises(ValueError) as ctx:
-            _get_json_for_fragment(
+            _get_request_parameters_for_fragment(
                 {'node-url': 'http://localhost:1234'},
                 '/fragment',
                 method='GET',
                 post_args={'foo': 'bar'}
             )
-        self.assertTrue(
+        self.assertIn(
-            "only valid for POST" in str(ctx.exception)
+            "only valid for POST",
+            str(ctx.exception),
         )

tox.ini

@@ -74,6 +74,8 @@ commands =
 
          tahoe --version
 
+         python -c "import sys; print('sys.stdout.encoding:', sys.stdout.encoding)"
+
          # Run tests with -b to catch bugs like `"%s" % (some_bytes,)`. -b makes
          # Python emit BytesWarnings, and warnings configuration in
          # src/allmydata/tests/__init__.py turns allmydata's BytesWarnings into