14877653111

Committed 07 May 2025 07:27AM UTC coverage: 73.562% (-3.2%) from 76.779%

Build # 14877653111

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1814 from vantage6/253-feature-request-improved-node-diagnostic-tools

Feature/#253 Improved node diagnostic tools

Run Details

32 of 88 new or added lines in 5 files covered. (36.36%)

31 existing lines in 5 files now uncovered.

1522 of 2069 relevant lines covered (73.56%)

0.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.44

/vantage6/tests_cli/test_node_cli.py

import unittest
import logging
import os
import contextlib

from unittest.mock import MagicMock, patch
from pathlib import Path
from io import BytesIO, StringIO
from click.testing import CliRunner
from docker.errors import APIError

from vantage6.common.globals import Ports
from vantage6.cli.globals import APPNAME
from vantage6.common import STRING_ENCODING
from vantage6.cli.common.utils import print_log_worker
from vantage6.cli.node.list import cli_node_list
from vantage6.cli.node.new import cli_node_new_configuration
from vantage6.cli.node.files import cli_node_files
from vantage6.cli.node.start import cli_node_start
from vantage6.cli.node.restart import cli_node_restart
from vantage6.cli.node.stop import cli_node_stop
from vantage6.cli.node.attach import cli_node_attach
from vantage6.cli.node.create_private_key import cli_node_create_private_key
from vantage6.cli.node.clean import cli_node_clean
from vantage6.cli.node.common import create_client_and_authenticate


class NodeCLITest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        logging.getLogger("docker.utils.config").setLevel(logging.WARNING)
        return super().setUpClass()

    @patch("docker.DockerClient.ping")
    def test_list_docker_not_running(self, docker_ping):
        """An error is printed when docker is not running"""
        docker_ping.side_effect = Exception("Boom!")

        runner = CliRunner()
        result = runner.invoke(cli_node_list, [])

        # check exit code
        self.assertEqual(result.exit_code, 1)

    @patch("vantage6.cli.context.node.NodeContext.available_configurations")
    @patch("docker.DockerClient.ping")
    @patch("docker.DockerClient.containers")
    def test_list(self, containers, docker_ping, available_configurations):
        """A container list and their current status."""
        # https://docs.python.org/3/library/unittest.mock.html#mock-names-and-the-name-attribute

        # mock that docker-deamon is running
        docker_ping.return_value = True

        # docker deamon returns a list of running node-containers
        container1 = MagicMock()
        container1.name = f"{APPNAME}-iknl-user"
        containers.list.return_value = [container1]

        # returns a list of configurations and failed inports
        def side_effect(system_folders):
            config = MagicMock()
            config.name = "iknl"
            if not system_folders:
                return [[config], []]
            else:
                return [[config], []]

        available_configurations.side_effect = side_effect

        # invoke CLI method
        runner = CliRunner()
        result = runner.invoke(cli_node_list, [])

        # validate exit code
        self.assertEqual(result.exit_code, 0)

        # check printed lines
        self.assertEqual(
            result.output,
            "\nName                     Status          System/User\n"
            "-----------------------------------------------------\n"
            "iknl                     Not running     System \n"
            "iknl                     Running         User   \n"
            "-----------------------------------------------------\n",
        )

    @patch("vantage6.cli.node.new.configuration_wizard")
    @patch("vantage6.cli.node.new.ensure_config_dir_writable")
    @patch("vantage6.cli.node.common.NodeContext")
    def test_new_config(self, context, permissions, wizard):
        """No error produced when creating new configuration."""
        context.config_exists.return_value = False
        permissions.return_value = True
        wizard.return_value = "/some/file/path"

        runner = CliRunner()
        result = runner.invoke(
            cli_node_new_configuration,
            [
                "--name",
                "some-name",
            ],
        )

        # check that info message is produced
        self.assertEqual(result.output[:7], "[info ]")

        # check OK exit code
        self.assertEqual(result.exit_code, 0)

    @patch("vantage6.cli.node.new.configuration_wizard")
    def test_new_config_replace_whitespace_in_name(self, _):
        """Whitespaces are replaced in the name."""

        runner = CliRunner()
        result = runner.invoke(
            cli_node_new_configuration,
            [
                "--name",
                "some name",
            ],
        )

        self.assertEqual(
            result.output[:60],
            "[info ] - Replaced spaces from configuration name: some-name",
        )

    @patch("vantage6.cli.node.new.NodeContext")
    def test_new_config_already_exists(self, context):
        """No duplicate configurations are allowed."""

        context.config_exists.return_value = True

        runner = CliRunner()
        result = runner.invoke(
            cli_node_new_configuration,
            [
                "--name",
                "some-name",
            ],
        )

        # check that error is produced
        self.assertEqual(result.output[:7], "[error]")

        # check non-zero exit code
        self.assertEqual(result.exit_code, 1)

    @patch("vantage6.cli.node.new.ensure_config_dir_writable")
    @patch("vantage6.cli.node.common.NodeContext")
    def test_new_write_permissions(self, context, permissions):
        """User needs write permissions."""

        context.config_exists.return_value = False
        permissions.return_value = False

        runner = CliRunner()
        result = runner.invoke(
            cli_node_new_configuration,
            [
                "--name",
                "some-name",
            ],
        )

        # check that error is produced
        self.assertEqual(result.output[:7], "[error]")

        # check non-zero exit code
        self.assertEqual(result.exit_code, 1)

    @patch("vantage6.cli.node.common.NodeContext")
    @patch("vantage6.cli.node.files.NodeContext")
    @patch("vantage6.cli.node.common.select_configuration_questionaire")
    def test_files(self, select_config, context, common_context):
        """No errors produced when retrieving filepaths."""

        common_context.config_exists.return_value = True
        context.return_value = MagicMock(
            config_file="/file.yaml", log_file="/log.log", data_dir="/dir"
        )
        context.return_value.databases.items.return_value = [["label", "/file.db"]]
        select_config.return_value = "iknl"

        runner = CliRunner()
        result = runner.invoke(cli_node_files, [])

        # we check that no warnings have been produced
        self.assertEqual(result.output[:7], "[info ]")

        # check status code is OK
        self.assertEqual(result.exit_code, 0)

    @patch("vantage6.cli.node.common.NodeContext")
    def test_files_non_existing_config(self, context):
        """An error is produced when a non existing config is used."""

        context.config_exists.return_value = False

        runner = CliRunner()
        result = runner.invoke(cli_node_files, ["--name", "non-existing"])

        # Check that error is produced
        self.assertEqual(result.output[:7], "[error]")

        # check for non zero exit-code
        self.assertNotEqual(result.exit_code, 0)

    @patch("docker.DockerClient.volumes")
    @patch("vantage6.cli.node.start.pull_infra_image")
    @patch("vantage6.cli.common.decorator.get_context")
    @patch("docker.DockerClient.containers")
    @patch("vantage6.cli.node.start.check_docker_running", return_value=True)
    def test_start(self, check_docker, client, context, pull, volumes):
        # client.containers = MagicMock(name="docker.DockerClient.containers")
        client.list.return_value = []
        volume = MagicMock()
        volume.name = "data-vol-name"
        volumes.create.return_value = volume
        context.config_exists.return_value = True

        ctx = MagicMock(
            data_dir=Path("data"),
            log_dir=Path("logs"),
            config_dir=Path("configs"),
            databases=[{"label": "some_label", "uri": "data.csv", "type": "csv"}],
        )

        # cli_node_start() tests for truth value of a set-like object derived
        # from ctx.config.get('node_extra_env', {}). Default MagicMock() will
        # evaluate to True, empty dict to False. False signifies no overwritten
        # env vars, hence no error.
        def config_get_side_effect(key, default=None):
            if key == "node_extra_env":
                return {}
            return MagicMock()

        ctx.config.get.side_effect = config_get_side_effect
        ctx.get_data_file.return_value = "data.csv"
        ctx.name = "some-name"
        context.return_value = ctx

        runner = CliRunner()

        # Should fail when starting node with non-existing database CSV file
        with runner.isolated_filesystem():
            result = runner.invoke(cli_node_start, ["--name", "some-name"])
        self.assertEqual(result.exit_code, 1)

        # now do it with a SQL database which doesn't have to be an existing file
        ctx.databases = [{"label": "some_label", "uri": "data.db", "type": "sql"}]
        with runner.isolated_filesystem():
            result = runner.invoke(cli_node_start, ["--name", "some-name"])
        self.assertEqual(result.exit_code, 0)

    def _setup_stop_test(self, containers):
        container1 = MagicMock()
        container1.name = f"{APPNAME}-iknl-user"
        containers.list.return_value = [container1]

    @patch("docker.DockerClient.containers")
    @patch("vantage6.cli.node.stop.check_docker_running", return_value=True)
    @patch("vantage6.cli.node.stop.NodeContext")
    @patch("vantage6.cli.node.stop.delete_volume_if_exists")
    def test_stop(self, delete_volume, node_context, check_docker, containers):
        self._setup_stop_test(containers)

        runner = CliRunner()

        result = runner.invoke(cli_node_stop, ["--name", "iknl"])

        self.assertEqual(
            result.output, "[info ] - Stopped the vantage6-iknl-user Node.\n"
        )

        self.assertEqual(result.exit_code, 0)

    @patch("docker.DockerClient.containers")
    @patch("vantage6.cli.node.stop.check_docker_running", return_value=True)
    @patch("vantage6.cli.node.stop.NodeContext")
    @patch("vantage6.cli.node.stop.delete_volume_if_exists")
    @patch("vantage6.cli.node.restart.subprocess.run")
    def test_restart(
        self, subprocess_run, delete_volume, node_context, check_docker, containers
    ):
        """Restart a node without errors."""
        self._setup_stop_test(containers)
        # The subprocess.run() function is called with the command to start the node.
        # Unfortunately it is hard to test this, so we just return a successful run
        subprocess_run.return_value = MagicMock(returncode=0)
        runner = CliRunner()
        with runner.isolated_filesystem():
            result = runner.invoke(cli_node_restart, ["--name", "iknl"])
        self.assertEqual(result.exit_code, 0)

    @patch("vantage6.cli.node.attach.time")
    @patch("vantage6.cli.node.attach.print_log_worker")
    @patch("docker.DockerClient.containers")
    @patch("vantage6.cli.node.attach.check_docker_running", return_value=True)
    def test_attach(self, check_docker, containers, log_worker, time_):
        """Attach docker logs without errors."""
        container1 = MagicMock()
        container1.name = f"{APPNAME}-iknl-user"
        containers.list.return_value = [container1]

        log_worker.return_value = ""
        time_.sleep.side_effect = KeyboardInterrupt()

        runner = CliRunner()
        result = runner.invoke(cli_node_attach, ["--name", "iknl"])

        self.assertEqual(
            result.output,
            "[info ] - Closing log file. Keyboard Interrupt.\n"
            "[info ] - Note that your node is still running! Shut it down "
            "with 'v6 node stop'\n",
        )
        self.assertEqual(result.exit_code, 0)

    @patch("vantage6.cli.node.clean.q")
    @patch("docker.DockerClient.volumes")
    @patch("vantage6.cli.node.clean.check_docker_running", return_value=True)
    def test_clean(self, check_docker, volumes, q):
        """Clean Docker volumes without errors."""
        volume1 = MagicMock()
        volume1.name = "some-name-tmpvol"
        volumes.list.return_value = [volume1]

        question = MagicMock(name="pop-the-question")
        question.ask.return_value = True
        q.confirm.return_value = question

        runner = CliRunner()
        result = runner.invoke(cli_node_clean)

        # check exit code
        self.assertEqual(result.exit_code, 0)

    @patch("vantage6.cli.node.create_private_key.create_client_and_authenticate")
    @patch("vantage6.cli.node.common.NodeContext")
    @patch("vantage6.cli.node.create_private_key.NodeContext")
    def test_create_private_key(self, context, common_context, client):
        common_context.config_exists.return_value = True
        context.return_value.type_data_folder.return_value = Path(".")
        client.return_value = MagicMock(whoami=MagicMock(organization_name="Test"))
        # client.whoami.organization_name = "Test"
        runner = CliRunner()

        result = runner.invoke(cli_node_create_private_key, ["--name", "application"])

        self.assertEqual(result.exit_code, 0)

        # remove the private key file again
        os.remove("privkey_Test.pem")

    @patch("vantage6.cli.node.create_private_key.RSACryptor")
    @patch("vantage6.cli.node.create_private_key.create_client_and_authenticate")
    @patch("vantage6.cli.node.common.NodeContext")
    @patch("vantage6.cli.node.create_private_key.NodeContext")
    def test_create_private_key_overwite(
        self, context, common_context, client, cryptor
    ):
        common_context.config_exists.return_value = True
        context.return_value.type_data_folder.return_value = Path(".")
        client.return_value = MagicMock(whoami=MagicMock(organization_name="Test"))
        cryptor.create_public_key_bytes.return_value = b""
        # client.whoami.organization_name = "Test"

        runner = CliRunner()

        # overwrite
        with runner.isolated_filesystem():
            with open("privkey_iknl.pem", "w") as f:
                f.write("does-not-matter")

            result = runner.invoke(
                cli_node_create_private_key,
                ["--name", "application", "--overwrite", "--organization-name", "iknl"],
            )
        self.assertEqual(result.exit_code, 0)

        # do not overwrite
        with runner.isolated_filesystem():
            with open("privkey_iknl.pem", "w") as f:
                f.write("does-not-matter")

            result = runner.invoke(
                cli_node_create_private_key,
                ["--name", "application", "--organization-name", "iknl"],
            )

            # print(result.output)

        self.assertEqual(result.exit_code, 0)

    @patch("vantage6.cli.node.common.NodeContext")
    def test_create_private_key_config_not_found(self, context):
        context.config_exists.return_value = False

        runner = CliRunner()
        result = runner.invoke(cli_node_create_private_key, ["--name", "application"])

        self.assertEqual(result.exit_code, 1)

    @patch("vantage6.cli.node.clean.q")
    @patch("docker.DockerClient.volumes")
    @patch("vantage6.common.docker.addons.check_docker_running")
    def test_clean_docker_error(self, check_docker, volumes, q):
        volume1 = MagicMock()
        volume1.name = "some-name-tmpvol"
        volume1.remove.side_effect = APIError("Testing")
        volumes.list.return_value = [volume1]
        question = MagicMock(name="pop-the-question")
        question.ask.return_value = True
        q.confirm.return_value = question

        runner = CliRunner()
        result = runner.invoke(cli_node_clean)

        # check exit code
        self.assertEqual(result.exit_code, 1)

    def test_print_log_worker(self):
        stream = BytesIO("Hello!".encode(STRING_ENCODING))
        temp_stdout = StringIO()
        with contextlib.redirect_stdout(temp_stdout):
            print_log_worker(stream)
        output = temp_stdout.getvalue().strip()
        self.assertEqual(output, "Hello!")

    @patch("vantage6.cli.node.common.info")
    @patch("vantage6.cli.node.common.debug")
    @patch("vantage6.cli.node.common.error")
    @patch("vantage6.cli.node.common.UserClient")
    @patch("vantage6.cli.node.common.q")
    def test_client(self, q, client, error, debug, info):
        ctx = MagicMock(
            config={
                "server_url": "localhost",
                "port": Ports.DEV_SERVER.value,
                "api_path": "",
            }
        )

        # should not trigger an exception
        try:
            create_client_and_authenticate(ctx)
        except Exception:
            self.fail("Raised an exception!")

        # client raises exception
        client.side_effect = Exception("Boom!")
        with self.assertRaises(Exception):
            create_client_and_authenticate(ctx)

    # TODO this function has been moved to the common package. A test should
    # be added there instead of here
    # @patch("vantage6.cli.node.error")
    # def test_check_docker(self, error):
    #     docker = MagicMock()
    #     try:
    #         check_docker_running()
    #     except Exception:
    #         self.fail("Exception raised!")

    #     docker.ping.side_effect = Exception("Boom!")
    #     with self.assertRaises(SystemExit):
    #         check_docker_running()

1	import unittest	1✔
2	import logging	1✔
3	import os	1✔
4	import contextlib	1✔
5
6	from unittest.mock import MagicMock, patch	1✔
7	from pathlib import Path	1✔
8	from io import BytesIO, StringIO	1✔
9	from click.testing import CliRunner	1✔
10	from docker.errors import APIError	1✔
11
12	from vantage6.common.globals import Ports	1✔
13	from vantage6.cli.globals import APPNAME	1✔
14	from vantage6.common import STRING_ENCODING	1✔
15	from vantage6.cli.common.utils import print_log_worker	1✔
16	from vantage6.cli.node.list import cli_node_list	1✔
17	from vantage6.cli.node.new import cli_node_new_configuration	1✔
18	from vantage6.cli.node.files import cli_node_files	1✔
19	from vantage6.cli.node.start import cli_node_start	1✔
20	from vantage6.cli.node.restart import cli_node_restart	1✔
21	from vantage6.cli.node.stop import cli_node_stop	1✔
22	from vantage6.cli.node.attach import cli_node_attach	1✔
23	from vantage6.cli.node.create_private_key import cli_node_create_private_key	1✔
24	from vantage6.cli.node.clean import cli_node_clean	1✔
25	from vantage6.cli.node.common import create_client_and_authenticate	1✔
26
27
28	class NodeCLITest(unittest.TestCase):	1✔
29	@classmethod	1✔
30	def setUpClass(cls):	1✔
31	logging.getLogger("docker.utils.config").setLevel(logging.WARNING)	1✔
32	return super().setUpClass()	1✔
33
34	@patch("docker.DockerClient.ping")	1✔
35	def test_list_docker_not_running(self, docker_ping):	1✔
36	"""An error is printed when docker is not running"""
37	docker_ping.side_effect = Exception("Boom!")	1✔
38
39	runner = CliRunner()	1✔
40	result = runner.invoke(cli_node_list, [])	1✔
41
42	# check exit code
43	self.assertEqual(result.exit_code, 1)	1✔
44
45	@patch("vantage6.cli.context.node.NodeContext.available_configurations")	1✔
46	@patch("docker.DockerClient.ping")	1✔
47	@patch("docker.DockerClient.containers")	1✔
48	def test_list(self, containers, docker_ping, available_configurations):	1✔
49	"""A container list and their current status."""
50	# https://docs.python.org/3/library/unittest.mock.html#mock-names-and-the-name-attribute
51
52	# mock that docker-deamon is running
53	docker_ping.return_value = True	1✔
54
55	# docker deamon returns a list of running node-containers
56	container1 = MagicMock()	1✔
57	container1.name = f"{APPNAME}-iknl-user"	1✔
58	containers.list.return_value = [container1]	1✔
59
60	# returns a list of configurations and failed inports
61	def side_effect(system_folders):	1✔
62	config = MagicMock()	1✔
63	config.name = "iknl"	1✔
64	if not system_folders:	1✔
65	return [[config], []]	1✔
66	else:
67	return [[config], []]	1✔
68
69	available_configurations.side_effect = side_effect	1✔
70
71	# invoke CLI method
72	runner = CliRunner()	1✔
73	result = runner.invoke(cli_node_list, [])	1✔
74
75	# validate exit code
76	self.assertEqual(result.exit_code, 0)	1✔
77
78	# check printed lines
79	self.assertEqual(	1✔
80	result.output,
81	"\nName Status System/User\n"
82	"-----------------------------------------------------\n"
83	"iknl Not running System \n"
84	"iknl Running User \n"
85	"-----------------------------------------------------\n",
86	)
87
88	@patch("vantage6.cli.node.new.configuration_wizard")	1✔
89	@patch("vantage6.cli.node.new.ensure_config_dir_writable")	1✔
90	@patch("vantage6.cli.node.common.NodeContext")	1✔
91	def test_new_config(self, context, permissions, wizard):	1✔
92	"""No error produced when creating new configuration."""
93	context.config_exists.return_value = False	1✔
94	permissions.return_value = True	1✔
95	wizard.return_value = "/some/file/path"	1✔
96
97	runner = CliRunner()	1✔
98	result = runner.invoke(	1✔
99	cli_node_new_configuration,
100	[
101	"--name",
102	"some-name",
103	],
104	)
105
106	# check that info message is produced
107	self.assertEqual(result.output[:7], "[info ]")	1✔
108
109	# check OK exit code
110	self.assertEqual(result.exit_code, 0)	1✔
111
112	@patch("vantage6.cli.node.new.configuration_wizard")	1✔
113	def test_new_config_replace_whitespace_in_name(self, _):	1✔
114	"""Whitespaces are replaced in the name."""
115
116	runner = CliRunner()	1✔
117	result = runner.invoke(	1✔
118	cli_node_new_configuration,
119	[
120	"--name",
121	"some name",
122	],
123	)
124
125	self.assertEqual(	1✔
126	result.output[:60],
127	"[info ] - Replaced spaces from configuration name: some-name",
128	)
129
130	@patch("vantage6.cli.node.new.NodeContext")	1✔
131	def test_new_config_already_exists(self, context):	1✔
132	"""No duplicate configurations are allowed."""
133
134	context.config_exists.return_value = True	1✔
135
136	runner = CliRunner()	1✔
137	result = runner.invoke(	1✔
138	cli_node_new_configuration,
139	[
140	"--name",
141	"some-name",
142	],
143	)
144
145	# check that error is produced
146	self.assertEqual(result.output[:7], "[error]")	1✔
147
148	# check non-zero exit code
149	self.assertEqual(result.exit_code, 1)	1✔
150
151	@patch("vantage6.cli.node.new.ensure_config_dir_writable")	1✔
152	@patch("vantage6.cli.node.common.NodeContext")	1✔
153	def test_new_write_permissions(self, context, permissions):	1✔
154	"""User needs write permissions."""
155
156	context.config_exists.return_value = False	1✔
157	permissions.return_value = False	1✔
158
159	runner = CliRunner()	1✔
160	result = runner.invoke(	1✔
161	cli_node_new_configuration,
162	[
163	"--name",
164	"some-name",
165	],
166	)
167
168	# check that error is produced
169	self.assertEqual(result.output[:7], "[error]")	1✔
170
171	# check non-zero exit code
172	self.assertEqual(result.exit_code, 1)	1✔
173
174	@patch("vantage6.cli.node.common.NodeContext")	1✔
175	@patch("vantage6.cli.node.files.NodeContext")	1✔
176	@patch("vantage6.cli.node.common.select_configuration_questionaire")	1✔
177	def test_files(self, select_config, context, common_context):	1✔
178	"""No errors produced when retrieving filepaths."""
179
180	common_context.config_exists.return_value = True	1✔
181	context.return_value = MagicMock(	1✔
182	config_file="/file.yaml", log_file="/log.log", data_dir="/dir"
183	)
184	context.return_value.databases.items.return_value = [["label", "/file.db"]]	1✔
185	select_config.return_value = "iknl"	1✔
186
187	runner = CliRunner()	1✔
188	result = runner.invoke(cli_node_files, [])	1✔
189
190	# we check that no warnings have been produced
191	self.assertEqual(result.output[:7], "[info ]")	1✔
192
193	# check status code is OK
194	self.assertEqual(result.exit_code, 0)	1✔
195
196	@patch("vantage6.cli.node.common.NodeContext")	1✔
197	def test_files_non_existing_config(self, context):	1✔
198	"""An error is produced when a non existing config is used."""
199
200	context.config_exists.return_value = False	1✔
201
202	runner = CliRunner()	1✔
203	result = runner.invoke(cli_node_files, ["--name", "non-existing"])	1✔
204
205	# Check that error is produced
206	self.assertEqual(result.output[:7], "[error]")	1✔
207
208	# check for non zero exit-code
209	self.assertNotEqual(result.exit_code, 0)	1✔
210
211	@patch("docker.DockerClient.volumes")	1✔
212	@patch("vantage6.cli.node.start.pull_infra_image")	1✔
213	@patch("vantage6.cli.common.decorator.get_context")	1✔
214	@patch("docker.DockerClient.containers")	1✔
215	@patch("vantage6.cli.node.start.check_docker_running", return_value=True)	1✔
216	def test_start(self, check_docker, client, context, pull, volumes):	1✔
217	# client.containers = MagicMock(name="docker.DockerClient.containers")
218	client.list.return_value = []	1✔
219	volume = MagicMock()	1✔
220	volume.name = "data-vol-name"	1✔
221	volumes.create.return_value = volume	1✔
222	context.config_exists.return_value = True	1✔
223
224	ctx = MagicMock(	1✔
225	data_dir=Path("data"),
226	log_dir=Path("logs"),
227	config_dir=Path("configs"),
228	databases=[{"label": "some_label", "uri": "data.csv", "type": "csv"}],
229	)
230
231	# cli_node_start() tests for truth value of a set-like object derived
232	# from ctx.config.get('node_extra_env', {}). Default MagicMock() will
233	# evaluate to True, empty dict to False. False signifies no overwritten
234	# env vars, hence no error.
235	def config_get_side_effect(key, default=None):	1✔
236	if key == "node_extra_env":	1✔
237	return {}	1✔
238	return MagicMock()	1✔
239
240	ctx.config.get.side_effect = config_get_side_effect	1✔
241	ctx.get_data_file.return_value = "data.csv"	1✔
242	ctx.name = "some-name"	1✔
243	context.return_value = ctx	1✔
244
245	runner = CliRunner()	1✔
246
247	# Should fail when starting node with non-existing database CSV file
248	with runner.isolated_filesystem():	1✔
249	result = runner.invoke(cli_node_start, ["--name", "some-name"])	1✔
250	self.assertEqual(result.exit_code, 1)	1✔
251
252	# now do it with a SQL database which doesn't have to be an existing file
253	ctx.databases = [{"label": "some_label", "uri": "data.db", "type": "sql"}]	1✔
254	with runner.isolated_filesystem():	1✔
255	result = runner.invoke(cli_node_start, ["--name", "some-name"])	1✔
256	self.assertEqual(result.exit_code, 0)	1✔
257
258	def _setup_stop_test(self, containers):	1✔
259	container1 = MagicMock()	1✔
260	container1.name = f"{APPNAME}-iknl-user"	1✔
261	containers.list.return_value = [container1]	1✔
262
263	@patch("docker.DockerClient.containers")	1✔
264	@patch("vantage6.cli.node.stop.check_docker_running", return_value=True)	1✔
265	@patch("vantage6.cli.node.stop.NodeContext")	1✔
266	@patch("vantage6.cli.node.stop.delete_volume_if_exists")	1✔
267	def test_stop(self, delete_volume, node_context, check_docker, containers):	1✔
268	self._setup_stop_test(containers)	1✔
269
270	runner = CliRunner()	1✔
271
272	result = runner.invoke(cli_node_stop, ["--name", "iknl"])	1✔
273
274	self.assertEqual(	1✔
275	result.output, "[info ] - Stopped the vantage6-iknl-user Node.\n"
276	)
277
278	self.assertEqual(result.exit_code, 0)	1✔
279
280	@patch("docker.DockerClient.containers")	1✔
281	@patch("vantage6.cli.node.stop.check_docker_running", return_value=True)	1✔
282	@patch("vantage6.cli.node.stop.NodeContext")	1✔
283	@patch("vantage6.cli.node.stop.delete_volume_if_exists")	1✔
284	@patch("vantage6.cli.node.restart.subprocess.run")	1✔
285	def test_restart(	1✔
286	self, subprocess_run, delete_volume, node_context, check_docker, containers
287	):
288	"""Restart a node without errors."""
289	self._setup_stop_test(containers)	1✔
290	# The subprocess.run() function is called with the command to start the node.
291	# Unfortunately it is hard to test this, so we just return a successful run
292	subprocess_run.return_value = MagicMock(returncode=0)	1✔
293	runner = CliRunner()	1✔
294	with runner.isolated_filesystem():	1✔
295	result = runner.invoke(cli_node_restart, ["--name", "iknl"])	1✔
296	self.assertEqual(result.exit_code, 0)	1✔
297
298	@patch("vantage6.cli.node.attach.time")	1✔
299	@patch("vantage6.cli.node.attach.print_log_worker")	1✔
300	@patch("docker.DockerClient.containers")	1✔
301	@patch("vantage6.cli.node.attach.check_docker_running", return_value=True)	1✔
302	def test_attach(self, check_docker, containers, log_worker, time_):	1✔
303	"""Attach docker logs without errors."""
304	container1 = MagicMock()	1✔
305	container1.name = f"{APPNAME}-iknl-user"	1✔
306	containers.list.return_value = [container1]	1✔
307
308	log_worker.return_value = ""	1✔
309	time_.sleep.side_effect = KeyboardInterrupt()	1✔
310
311	runner = CliRunner()	1✔
312	result = runner.invoke(cli_node_attach, ["--name", "iknl"])	1✔
313
UNCOV 314	self.assertEqual(	×
315	result.output,
316	"[info ] - Closing log file. Keyboard Interrupt.\n"
317	"[info ] - Note that your node is still running! Shut it down "
318	"with 'v6 node stop'\n",
319	)
UNCOV 320	self.assertEqual(result.exit_code, 0)	×
321
322	@patch("vantage6.cli.node.clean.q")	1✔
323	@patch("docker.DockerClient.volumes")	1✔
324	@patch("vantage6.cli.node.clean.check_docker_running", return_value=True)	1✔
325	def test_clean(self, check_docker, volumes, q):	1✔
326	"""Clean Docker volumes without errors."""
327	volume1 = MagicMock()	1✔
328	volume1.name = "some-name-tmpvol"	1✔
329	volumes.list.return_value = [volume1]	1✔
330
331	question = MagicMock(name="pop-the-question")	1✔
332	question.ask.return_value = True	1✔
333	q.confirm.return_value = question	1✔
334
335	runner = CliRunner()	1✔
336	result = runner.invoke(cli_node_clean)	1✔
337
338	# check exit code
339	self.assertEqual(result.exit_code, 0)	1✔
340
341	@patch("vantage6.cli.node.create_private_key.create_client_and_authenticate")	1✔
342	@patch("vantage6.cli.node.common.NodeContext")	1✔
343	@patch("vantage6.cli.node.create_private_key.NodeContext")	1✔
344	def test_create_private_key(self, context, common_context, client):	1✔
345	common_context.config_exists.return_value = True	1✔
346	context.return_value.type_data_folder.return_value = Path(".")	1✔
347	client.return_value = MagicMock(whoami=MagicMock(organization_name="Test"))	1✔
348	# client.whoami.organization_name = "Test"
349	runner = CliRunner()	1✔
350
351	result = runner.invoke(cli_node_create_private_key, ["--name", "application"])	1✔
352
353	self.assertEqual(result.exit_code, 0)	1✔
354
355	# remove the private key file again
356	os.remove("privkey_Test.pem")	1✔
357
358	@patch("vantage6.cli.node.create_private_key.RSACryptor")	1✔
359	@patch("vantage6.cli.node.create_private_key.create_client_and_authenticate")	1✔
360	@patch("vantage6.cli.node.common.NodeContext")	1✔
361	@patch("vantage6.cli.node.create_private_key.NodeContext")	1✔
362	def test_create_private_key_overwite(	1✔
363	self, context, common_context, client, cryptor
364	):
365	common_context.config_exists.return_value = True	1✔
366	context.return_value.type_data_folder.return_value = Path(".")	1✔
367	client.return_value = MagicMock(whoami=MagicMock(organization_name="Test"))	1✔
368	cryptor.create_public_key_bytes.return_value = b""	1✔
369	# client.whoami.organization_name = "Test"
370
371	runner = CliRunner()	1✔
372
373	# overwrite
374	with runner.isolated_filesystem():	1✔
375	with open("privkey_iknl.pem", "w") as f:	1✔
376	f.write("does-not-matter")	1✔
377
378	result = runner.invoke(	1✔
379	cli_node_create_private_key,
380	["--name", "application", "--overwrite", "--organization-name", "iknl"],
381	)
382	self.assertEqual(result.exit_code, 0)	1✔
383
384	# do not overwrite
385	with runner.isolated_filesystem():	1✔
386	with open("privkey_iknl.pem", "w") as f:	1✔
387	f.write("does-not-matter")	1✔
388
389	result = runner.invoke(	1✔
390	cli_node_create_private_key,
391	["--name", "application", "--organization-name", "iknl"],
392	)
393
394	# print(result.output)
395
396	self.assertEqual(result.exit_code, 0)	1✔
397
398	@patch("vantage6.cli.node.common.NodeContext")	1✔
399	def test_create_private_key_config_not_found(self, context):	1✔
400	context.config_exists.return_value = False	1✔
401
402	runner = CliRunner()	1✔
403	result = runner.invoke(cli_node_create_private_key, ["--name", "application"])	1✔
404
405	self.assertEqual(result.exit_code, 1)	1✔
406
407	@patch("vantage6.cli.node.clean.q")	1✔
408	@patch("docker.DockerClient.volumes")	1✔
409	@patch("vantage6.common.docker.addons.check_docker_running")	1✔
410	def test_clean_docker_error(self, check_docker, volumes, q):	1✔
411	volume1 = MagicMock()	1✔
412	volume1.name = "some-name-tmpvol"	1✔
413	volume1.remove.side_effect = APIError("Testing")	1✔
414	volumes.list.return_value = [volume1]	1✔
415	question = MagicMock(name="pop-the-question")	1✔
416	question.ask.return_value = True	1✔
417	q.confirm.return_value = question	1✔
418
419	runner = CliRunner()	1✔
420	result = runner.invoke(cli_node_clean)	1✔
421
422	# check exit code
423	self.assertEqual(result.exit_code, 1)	1✔
424
425	def test_print_log_worker(self):	1✔
426	stream = BytesIO("Hello!".encode(STRING_ENCODING))	1✔
427	temp_stdout = StringIO()	1✔
428	with contextlib.redirect_stdout(temp_stdout):	1✔
429	print_log_worker(stream)	1✔
430	output = temp_stdout.getvalue().strip()	1✔
431	self.assertEqual(output, "Hello!")	1✔
432
433	@patch("vantage6.cli.node.common.info")	1✔
434	@patch("vantage6.cli.node.common.debug")	1✔
435	@patch("vantage6.cli.node.common.error")	1✔
436	@patch("vantage6.cli.node.common.UserClient")	1✔
437	@patch("vantage6.cli.node.common.q")	1✔
438	def test_client(self, q, client, error, debug, info):	1✔
439	ctx = MagicMock(	1✔
440	config={
441	"server_url": "localhost",
442	"port": Ports.DEV_SERVER.value,
443	"api_path": "",
444	}
445	)
446
447	# should not trigger an exception
448	try:	1✔
449	create_client_and_authenticate(ctx)	1✔
450	except Exception:	×
451	self.fail("Raised an exception!")	×
452
453	# client raises exception
454	client.side_effect = Exception("Boom!")	1✔
455	with self.assertRaises(Exception):	1✔
456	create_client_and_authenticate(ctx)	1✔
457
458	# TODO this function has been moved to the common package. A test should
459	# be added there instead of here
460	# @patch("vantage6.cli.node.error")
461	# def test_check_docker(self, error):
462	# docker = MagicMock()
463	# try:
464	# check_docker_running()
465	# except Exception:
466	# self.fail("Exception raised!")
467
468	# docker.ping.side_effect = Exception("Boom!")
469	# with self.assertRaises(SystemExit):
470	# check_docker_running()

vantage6 / vantage6 / 14877653111

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous