18191710511

Committed 02 Oct 2025 11:30AM UTC coverage: 29.08%. First build

Build # 18191710511

Build Type

Pull #2287

github

Committed by

web-flow

Commit Message

Merge 3e2880106 into d88a9c354

Pull Request Pull Request #2287: Change/#1510 and reinstate the dev network commands

Run Details

17 of 48 new or added lines in 10 files covered. (35.42%)

196 of 674 relevant lines covered (29.08%)

0.29 hits per line

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

0.0

/vantage6/tests_cli/test_node_cli.py

import contextlib
import logging
import os
import unittest
from io import BytesIO, StringIO
from pathlib import Path
from unittest.mock import MagicMock, patch

from click.testing import CliRunner

from vantage6.common import STRING_ENCODING
from vantage6.common.globals import Ports

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


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.make_configuration")
    @patch("vantage6.cli.node.new.ensure_config_dir_writable")
    @patch("vantage6.cli.node.common.NodeContext")
    def test_new_config(self, context, permissions, make_configuration):
        """No error produced when creating new configuration."""
        context.config_exists.return_value = False
        permissions.return_value = True
        make_configuration.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.make_configuration")
    def test_new_config_replace_whitespace_in_name(self, make_configuration):
        """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_questionnaire")
    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.attach_logs")
    def test_attach(self, attach_logs):
        """Attach docker logs without errors."""
        runner = CliRunner()
        runner.invoke(cli_node_attach)
        attach_logs.assert_called_once_with("app=node")

    @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)

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

vantage6 / vantage6 / 18191710511

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