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

"""
Tests for the grid manager CLI.
"""

import os
from io import (
    BytesIO,
)
from unittest import (
    skipIf,
)

from twisted.trial.unittest import (
    TestCase,
)
from allmydata.cli.grid_manager import (
    grid_manager,
)

import click.testing

# these imports support the tests for `tahoe *` subcommands
from ..common_util import (
    run_cli,
)
from ..common import (
    superuser,
)
from twisted.internet.defer import (
    inlineCallbacks,
)
from twisted.python.filepath import (
    FilePath,
)
from twisted.python.runtime import (
    platform,
)
from allmydata.util import jsonbytes as json

class GridManagerCommandLine(TestCase):
    """
    Test the mechanics of the `grid-manager` command
    """

    def setUp(self):
        self.runner = click.testing.CliRunner()
        super(GridManagerCommandLine, self).setUp()

    def invoke_and_check(self, *args, **kwargs):
        """Invoke a command with the runner and ensure it succeeded."""
        result = self.runner.invoke(*args, **kwargs)
        if result.exception is not None:
            raise result.exc_info[1].with_traceback(result.exc_info[2])
        self.assertEqual(result.exit_code, 0, result)
        return result

    def test_create(self):
        """
        Create a new grid-manager
        """
        with self.runner.isolated_filesystem():
            result = self.invoke_and_check(grid_manager, ["--config", "foo", "create"])
            self.assertEqual(["foo"], os.listdir("."))
            self.assertEqual(["config.json"], os.listdir("./foo"))
            result = self.invoke_and_check(grid_manager, ["--config", "foo", "public-identity"])
            self.assertTrue(result.output.startswith("pub-v0-"))

    def test_load_invalid(self):
        """
        An invalid config is reported to the user
        """
        with self.runner.isolated_filesystem():
            with open("config.json", "wb") as f:
                f.write(json.dumps_bytes({"not": "valid"}))
            result = self.runner.invoke(grid_manager, ["--config", ".", "public-identity"])
            self.assertNotEqual(result.exit_code, 0)
            self.assertIn(
                "Error loading Grid Manager",
                result.output,
            )

    def test_create_already(self):
        """
        It's an error to create a new grid-manager in an existing
        directory.
        """
        with self.runner.isolated_filesystem():
            result = self.invoke_and_check(grid_manager, ["--config", "foo", "create"])
            result = self.runner.invoke(grid_manager, ["--config", "foo", "create"])
            self.assertEqual(1, result.exit_code)
            self.assertIn(
                "Can't create",
                result.output,
            )

    def test_create_stdout(self):
        """
        Create a new grid-manager with no files
        """
        with self.runner.isolated_filesystem():
            result = self.invoke_and_check(grid_manager, ["--config", "-", "create"])
            self.assertEqual([], os.listdir("."))
            config = json.loads(result.output)
            self.assertEqual(
                {"private_key", "grid_manager_config_version"},
                set(config.keys()),
            )

    def test_list_stdout(self):
        """
        Load Grid Manager without files (using 'list' subcommand, but any will do)
        """
        config = {
            "storage_servers": {
                "storage0": {
                    "public_key": "pub-v0-cbq6hcf3pxcz6ouoafrbktmkixkeuywpcpbcomzd3lqbkq4nmfga"
                }
            },
            "private_key": "priv-v0-6uinzyaxy3zvscwgsps5pxcfezhrkfb43kvnrbrhhfzyduyqnniq",
            "grid_manager_config_version": 0
        }
        result = self.invoke_and_check(
            grid_manager, ["--config", "-", "list"],
            input=BytesIO(json.dumps_bytes(config)),
        )
        self.assertEqual(result.exit_code, 0)
        self.assertEqual(
            "storage0: pub-v0-cbq6hcf3pxcz6ouoafrbktmkixkeuywpcpbcomzd3lqbkq4nmfga\n",
            result.output,
        )

    def test_add_and_sign(self):
        """
        Add a new storage-server and sign a certificate for it
        """
        pubkey = "pub-v0-cbq6hcf3pxcz6ouoafrbktmkixkeuywpcpbcomzd3lqbkq4nmfga"
        with self.runner.isolated_filesystem():
            self.invoke_and_check(grid_manager, ["--config", "foo", "create"])
            self.invoke_and_check(grid_manager, ["--config", "foo", "add", "storage0", pubkey])
            result = self.invoke_and_check(grid_manager, ["--config", "foo", "sign", "storage0", "10"])
            sigcert = json.loads(result.output)
            self.assertEqual({"certificate", "signature"}, set(sigcert.keys()))
            cert = json.loads(sigcert['certificate'])
            self.assertEqual(cert["public_key"], pubkey)

    def test_add_and_sign_second_cert(self):
        """
        Add a new storage-server and sign two certificates.
        """
        pubkey = "pub-v0-cbq6hcf3pxcz6ouoafrbktmkixkeuywpcpbcomzd3lqbkq4nmfga"
        with self.runner.isolated_filesystem():
            self.invoke_and_check(grid_manager, ["--config", "foo", "create"])
            self.invoke_and_check(grid_manager, ["--config", "foo", "add", "storage0", pubkey])
            self.invoke_and_check(grid_manager, ["--config", "foo", "sign", "storage0", "10"])
            self.invoke_and_check(grid_manager, ["--config", "foo", "sign", "storage0", "10"])
            # we should now have two certificates stored
            self.assertEqual(
                set(FilePath("foo").listdir()),
                {'storage0.cert.1', 'storage0.cert.0', 'config.json'},
            )

    def test_add_twice(self):
        """
        An error is reported trying to add an existing server
        """
        pubkey0 = "pub-v0-cbq6hcf3pxcz6ouoafrbktmkixkeuywpcpbcomzd3lqbkq4nmfga"
        pubkey1 = "pub-v0-5ysc55trfvfvg466v46j4zmfyltgus3y2gdejifctv7h4zkuyveq"
        with self.runner.isolated_filesystem():
            self.invoke_and_check(grid_manager, ["--config", "foo", "create"])
            self.invoke_and_check(grid_manager, ["--config", "foo", "add", "storage0", pubkey0])
            result = self.runner.invoke(grid_manager, ["--config", "foo", "add", "storage0", pubkey1])
            self.assertNotEqual(result.exit_code, 0)
            self.assertIn(
                "A storage-server called 'storage0' already exists",
                result.output,
            )

    def test_add_list_remove(self):
        """
        Add a storage server, list it, remove it.
        """
        pubkey = "pub-v0-cbq6hcf3pxcz6ouoafrbktmkixkeuywpcpbcomzd3lqbkq4nmfga"
        with self.runner.isolated_filesystem():
            self.invoke_and_check(grid_manager, ["--config", "foo", "create"])
            self.invoke_and_check(grid_manager, ["--config", "foo", "add", "storage0", pubkey])
            self.invoke_and_check(grid_manager, ["--config", "foo", "sign", "storage0", "1"])

            result = self.invoke_and_check(grid_manager, ["--config", "foo", "list"])
            names = [
                line.split(':')[0]
                for line in result.output.strip().split('\n')
                if not line.startswith("  ")  # "cert" lines start with whitespace
            ]
            self.assertEqual(names, ["storage0"])

            self.invoke_and_check(grid_manager, ["--config", "foo", "remove", "storage0"])

            result = self.invoke_and_check(grid_manager, ["--config", "foo", "list"])
            self.assertEqual(result.output.strip(), "")

    def test_remove_missing(self):
        """
        Error reported when removing non-existent server
        """
        with self.runner.isolated_filesystem():
            self.invoke_and_check(grid_manager, ["--config", "foo", "create"])
            result = self.runner.invoke(grid_manager, ["--config", "foo", "remove", "storage0"])
            self.assertNotEqual(result.exit_code, 0)
            self.assertIn(
                "No storage-server called 'storage0' exists",
                result.output,
            )

    def test_sign_missing(self):
        """
        Error reported when signing non-existent server
        """
        with self.runner.isolated_filesystem():
            self.invoke_and_check(grid_manager, ["--config", "foo", "create"])
            result = self.runner.invoke(grid_manager, ["--config", "foo", "sign", "storage0", "42"])
            self.assertNotEqual(result.exit_code, 0)
            self.assertIn(
                "No storage-server called 'storage0' exists",
                result.output,
            )

    @skipIf(platform.isWindows(), "We don't know how to set permissions on Windows.")
    @skipIf(superuser, "cannot test as superuser with all permissions")
    def test_sign_bad_perms(self):
        """
        Error reported if we can't create certificate file
        """
        pubkey = "pub-v0-cbq6hcf3pxcz6ouoafrbktmkixkeuywpcpbcomzd3lqbkq4nmfga"
        with self.runner.isolated_filesystem():
            self.invoke_and_check(grid_manager, ["--config", "foo", "create"])
            self.invoke_and_check(grid_manager, ["--config", "foo", "add", "storage0", pubkey])
            # make the directory un-writable (so we can't create a new cert)
            os.chmod("foo", 0o550)
            result = self.runner.invoke(grid_manager, ["--config", "foo", "sign", "storage0", "42"])
            self.assertEqual(result.exit_code, 1)
            self.assertIn(
                "Permission denied",
                result.output,
            )


class TahoeAddGridManagerCert(TestCase):
    """
    Test `tahoe admin add-grid-manager-cert` subcommand
    """

    @inlineCallbacks
    def test_help(self):
        """
        some kind of help is printed
        """
        code, out, err = yield run_cli("admin", "add-grid-manager-cert")
        self.assertEqual(err, "")
        self.assertNotEqual(0, code)

    @inlineCallbacks
    def test_no_name(self):
        """
        error to miss --name option
        """
        code, out, err = yield run_cli(
            "admin", "add-grid-manager-cert", "--filename", "-",
            stdin=b"the cert",
        )
        self.assertIn(
            "Must provide --name",
            out
        )

    @inlineCallbacks
    def test_no_filename(self):
        """
        error to miss --name option
        """
        code, out, err = yield run_cli(
            "admin", "add-grid-manager-cert", "--name", "foo",
            stdin=b"the cert",
        )
        self.assertIn(
            "Must provide --filename",
            out
        )

    @inlineCallbacks
    def test_add_one(self):
        """
        we can add a certificate
        """
        nodedir = self.mktemp()
        fake_cert = b"""{"certificate": "", "signature": ""}"""

        code, out, err = yield run_cli(
            "--node-directory", nodedir,
            "admin", "add-grid-manager-cert", "-f", "-", "--name", "foo",
            stdin=fake_cert,
            ignore_stderr=True,
        )
        nodepath = FilePath(nodedir)
        with nodepath.child("tahoe.cfg").open("r") as f:
            config_data = f.read()

        self.assertIn("tahoe.cfg", nodepath.listdir())
        self.assertIn(
            b"foo = foo.cert",
            config_data,
        )
        self.assertIn("foo.cert", nodepath.listdir())
        with nodepath.child("foo.cert").open("r") as f:
            self.assertEqual(
                json.load(f),
                json.loads(fake_cert)
            )