source: trunk/src/allmydata/test/cli/test_status.py

"""
Ported to Python 3.
"""

import os
import tempfile
from io import BytesIO, StringIO
from os.path import join

from twisted.trial import unittest
from twisted.internet import defer

from allmydata.mutable.publish import MutableData
from allmydata.scripts.common_http import BadResponse
from allmydata.scripts.tahoe_status import _handle_response_for_fragment
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

from allmydata.immutable.upload import UploadStatus
from allmydata.immutable.downloader.status import DownloadStatus
from allmydata.mutable.publish import PublishStatus
from allmydata.mutable.retrieve import RetrieveStatus
from allmydata.mutable.servermap import UpdateStatus
from allmydata.util import jsonbytes as json

from ..no_network import GridTestMixin
from ..common_web import do_http
from .common import CLITestMixin


class FakeStatus:
    def __init__(self):
        self.status = []

    def setServiceParent(self, p):
        pass

    def get_status(self):
        return self.status

    def get_storage_index(self):
        return None

    def get_size(self):
        return None


class ProgressBar(unittest.TestCase):

    def test_ascii0(self):
        prog = pretty_progress(80.0, size=10, output_ascii=True)
        self.assertEqual('########. ', prog)

    def test_ascii1(self):
        prog = pretty_progress(10.0, size=10, output_ascii=True)
        self.assertEqual('#.        ', prog)

    def test_ascii2(self):
        prog = pretty_progress(13.0, size=10, output_ascii=True)
        self.assertEqual('#o        ', prog)

    def test_ascii3(self):
        prog = pretty_progress(90.0, size=10, output_ascii=True)
        self.assertEqual('#########.', prog)

    def test_unicode0(self):
        self.assertEqual(
            pretty_progress(82.0, size=10, output_ascii=False),
            u'\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u258e ',
        )

    def test_unicode1(self):
        self.assertEqual(
            pretty_progress(100.0, size=10, output_ascii=False),
            u'\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588',
        )


class _FakeOptions(dict):
    def __init__(self):
        self._tmp = tempfile.mkdtemp()
        os.mkdir(join(self._tmp, 'private'), 0o777)
        with open(join(self._tmp, 'private', 'api_auth_token'), 'w') as f:
            f.write('a' * 32)
        with open(join(self._tmp, 'node.url'), 'w') as f:
            f.write('localhost:9000')

        self['node-directory'] = self._tmp
        self['verbose'] = True
        self.stdout = StringIO()
        self.stderr = StringIO()


class Integration(GridTestMixin, CLITestMixin, unittest.TestCase):

    @defer.inlineCallbacks
    def setUp(self):
        yield super(Integration, self).setUp()
        self.basedir = "cli/status"
        self.set_up_grid()

        # upload something
        c0 = self.g.clients[0]
        data = MutableData(b"data" * 100)
        filenode = yield c0.create_mutable_file(data)
        self.uri = filenode.get_uri()

        # make sure our web-port is actually answering
        yield do_http("get", 'http://127.0.0.1:{}/status?t=json'.format(self.client_webports[0]))

    def test_simple(self):
        d = self.do_cli('status')# '--verbose')

        def _check(ign):
            code, stdout, stderr = ign
            self.assertEqual(code, 0, stderr)
            self.assertTrue('Skipped 1' in stdout)
        d.addCallback(_check)
        return d

    @defer.inlineCallbacks
    def test_help(self):
        rc, _, _ = yield self.do_cli('status', '--help')
        self.assertEqual(rc, 0)


class CommandStatus(unittest.TestCase):
    """
    These tests just exercise the renderers and ensure they don't
    catastrophically fail.
    """

    def setUp(self):
        self.options = _FakeOptions()

    def test_no_operations(self):
        values = [
            StringIO(json.dumps({
                "active": [],
                "recent": [],
            })),
            StringIO(json.dumps({
                "counters": {
                    "bytes_downloaded": 0,
                },
                "stats": {
                    "node.uptime": 0,
                }
            })),
        ]
        def do_http(*args, **kw):
            return values.pop(0)
        do_status(self.options, do_http)

    def test_simple(self):
        recent_items = active_items = [
            UploadStatus(),
            DownloadStatus(b"abcd", 12345),
            PublishStatus(),
            RetrieveStatus(),
            UpdateStatus(),
            FakeStatus(),
        ]
        values = [
            BytesIO(json.dumps({
                "active": list(
                    marshal_json(item)
                    for item
                    in active_items
                ),
                "recent": list(
                    marshal_json(item)
                    for item
                    in recent_items
                ),
            }).encode("utf-8")),
            BytesIO(json.dumps({
                "counters": {
                    "bytes_downloaded": 0,
                },
                "stats": {
                    "node.uptime": 0,
                }
            }).encode("utf-8")),
        ]
        def do_http(*args, **kw):
            return values.pop(0)
        do_status(self.options, do_http)

    def test_fetch_error(self):
        def do_http(*args, **kw):
            raise RuntimeError("boom")
        do_status(self.options, do_http)


class JsonHelpers(unittest.TestCase):

    def test_bad_response(self):
        def do_http(*args, **kw):
            return
        with self.assertRaises(RuntimeError) as ctx:
            _handle_response_for_fragment(
                BadResponse('the url', 'some err'),
                'http://localhost:1234',
            )
        self.assertIn(
            "Failed to get",
            str(ctx.exception),
        )

    def test_happy_path(self):
        resp = _handle_response_for_fragment(
            StringIO('{"some": "json"}'),
            'http://localhost:1234/',
        )
        self.assertEqual(resp, dict(some='json'))

    def test_happy_path_post(self):
        resp = _handle_response_for_fragment(
            StringIO('{"some": "json"}'),
            'http://localhost:1234/',
        )
        self.assertEqual(resp, dict(some='json'))

    def test_no_data_returned(self):
        with self.assertRaises(RuntimeError) as ctx:
            _handle_response_for_fragment(StringIO('null'), 'http://localhost:1234')
        self.assertIn('No data from', str(ctx.exception))

    def test_no_post_args(self):
        with self.assertRaises(ValueError) as ctx:
            _get_request_parameters_for_fragment(
                {'node-url': 'http://localhost:1234'},
                '/fragment',
                method='POST',
                post_args=None,
            )
        self.assertIn(
            "Must pass post_args",
            str(ctx.exception),
        )

    def test_post_args_for_get(self):
        with self.assertRaises(ValueError) as ctx:
            _get_request_parameters_for_fragment(
                {'node-url': 'http://localhost:1234'},
                '/fragment',
                method='GET',
                post_args={'foo': 'bar'}
            )
        self.assertIn(
            "only valid for POST",
            str(ctx.exception),
        )