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_wizard.py

import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch

from vantage6.common.globals import InstanceType, NodePolicy

from vantage6.cli.configuration_create import (
    make_configuration,
    node_configuration_questionaire,
    select_configuration_questionnaire,
    server_configuration_questionaire,
)

module_path = "vantage6.cli.configuration_create"


class WizardTest(unittest.TestCase):
    @staticmethod
    def prompts(*args, **kwargs):
        result = {}
        for arg in args[0]:
            name = arg["name"]
            if name == "default":  # default db path
                result[name] = "/some/path/db.sqlite"
            else:
                if "default" in arg:
                    result[name] = arg["default"]
                else:
                    result[name] = None
        return result

    @patch("vantage6.cli.configuration_create.NodeClient.authenticate")
    def test_node_wizard(self, authenticate):
        """An error is printed when docker is not running"""
        authenticate.return_value = None

        with patch(f"{module_path}.q") as q:
            q.unsafe_prompt.side_effect = self.prompts
            q.confirm.return_value.unsafe_ask.side_effect = [
                True,  # add a database
                False,  # don't enable two-factor authentication
                True,  # add VPN server
                True,  # add algorithm policies
                True,  # add single algorithms to allowed_algorithms
                "some-image",  # algorithm image to whitelist
                False,  # don't add another algorithm image
                True,  # add algorithm stores to allowed_algorithm_stores
                "some-store",  # algorithm store to whitelist
                False,  # don't add another algorithm store
                False,  # answer question on combining policies on store level and
                # single algorithm level
                False,  # don't abort if no server connection is made to pull
                # collaboration settings
                True,  # Enable encryption
            ]
            dirs = MagicMock(data="/")
            config = node_configuration_questionaire(dirs, "iknl")

        keys = [
            "api_key",
            "server_url",
            "port",
            "api_path",
            "task_dir",
            "databases",
            "logging",
            "encryption",
            "vpn_subnet",
        ]
        for key in keys:
            self.assertIn(key, config)
        nested_keys = [
            ["policies", NodePolicy.ALLOWED_ALGORITHMS.value],
            ["policies", NodePolicy.ALLOWED_ALGORITHM_STORES.value],
        ]
        for nesting in nested_keys:
            current_config = config
            for key in nesting:
                self.assertIn(key, current_config)
                current_config = current_config[key]

    def test_server_wizard(self):
        with patch(f"{module_path}.q") as q:
            q.unsafe_prompt.side_effect = self.prompts
            q.confirm.return_value.unsafe_ask.side_effect = [
                True,
                True,
                True,
                True,
                True,
                True,
                False,
            ]

            config = server_configuration_questionaire()

            keys = [
                "description",
                "ip",
                "port",
                "api_path",
                "uri",
                "allow_drop_all",
                "logging",
                "vpn_server",
                "rabbitmq",
                "two_factor_auth",
                "algorithm_stores",
            ]

            for key in keys:
                self.assertIn(key, config)

    @patch(f"{module_path}.node_configuration_questionaire")
    @patch(f"{module_path}.server_configuration_questionaire")
    @patch(f"{module_path}.ServerConfigurationManager")
    @patch(f"{module_path}.NodeConfigurationManager")
    @patch("vantage6.cli.configuration_create.AppContext")
    def test_configuration_create_interface(
        self, context, node_m, server_m, server_q, node_q
    ):
        context.instance_folders.return_value = {"config": "/some/path/"}

        file_ = make_configuration(InstanceType.NODE, "vtg6", False)
        self.assertEqual(Path("/some/path/vtg6.yaml"), file_)

        file_ = make_configuration(InstanceType.SERVER, "vtg6", True)
        self.assertEqual(Path("/some/path/vtg6.yaml"), file_)

    @patch("vantage6.cli.configuration_create.AppContext.available_configurations")
    def test_select_configuration(self, available_configurations):
        config = MagicMock()
        config.name = "vtg6"

        available_configurations.return_value = [[config], []]

        with patch(f"{module_path}.q") as q:
            q.select.return_value.unsafe_ask.return_value = "vtg6"
            name = select_configuration_questionnaire(InstanceType.NODE, True)

        self.assertEqual(name, "vtg6")

1	import unittest	×
2	from pathlib import Path	×
3	from unittest.mock import MagicMock, patch	×
4
5	from vantage6.common.globals import InstanceType, NodePolicy	×
6
NEW 7	from vantage6.cli.configuration_create import (	×
8	make_configuration,
9	node_configuration_questionaire,
10	select_configuration_questionnaire,
11	server_configuration_questionaire,
12	)
13
NEW 14	module_path = "vantage6.cli.configuration_create"	×
15
16
17	class WizardTest(unittest.TestCase):	×
18	@staticmethod	×
19	def prompts(args, *kwargs):	×
20	result = {}	×
21	for arg in args[0]:	×
22	name = arg["name"]	×
23	if name == "default": # default db path	×
24	result[name] = "/some/path/db.sqlite"	×
25	else:
26	if "default" in arg:	×
27	result[name] = arg["default"]	×
28	else:
29	result[name] = None	×
30	return result	×
31
NEW 32	@patch("vantage6.cli.configuration_create.NodeClient.authenticate")	×
33	def test_node_wizard(self, authenticate):	×
34	"""An error is printed when docker is not running"""
35	authenticate.return_value = None	×
36
37	with patch(f"{module_path}.q") as q:	×
38	q.unsafe_prompt.side_effect = self.prompts	×
39	q.confirm.return_value.unsafe_ask.side_effect = [	×
40	True, # add a database
41	False, # don't enable two-factor authentication
42	True, # add VPN server
43	True, # add algorithm policies
44	True, # add single algorithms to allowed_algorithms
45	"some-image", # algorithm image to whitelist
46	False, # don't add another algorithm image
47	True, # add algorithm stores to allowed_algorithm_stores
48	"some-store", # algorithm store to whitelist
49	False, # don't add another algorithm store
50	False, # answer question on combining policies on store level and
51	# single algorithm level
52	False, # don't abort if no server connection is made to pull
53	# collaboration settings
54	True, # Enable encryption
55	]
56	dirs = MagicMock(data="/")	×
57	config = node_configuration_questionaire(dirs, "iknl")	×
58
59	keys = [	×
60	"api_key",
61	"server_url",
62	"port",
63	"api_path",
64	"task_dir",
65	"databases",
66	"logging",
67	"encryption",
68	"vpn_subnet",
69	]
70	for key in keys:	×
71	self.assertIn(key, config)	×
72	nested_keys = [	×
73	["policies", NodePolicy.ALLOWED_ALGORITHMS.value],
74	["policies", NodePolicy.ALLOWED_ALGORITHM_STORES.value],
75	]
76	for nesting in nested_keys:	×
77	current_config = config	×
78	for key in nesting:	×
79	self.assertIn(key, current_config)	×
80	current_config = current_config[key]	×
81
82	def test_server_wizard(self):	×
83	with patch(f"{module_path}.q") as q:	×
84	q.unsafe_prompt.side_effect = self.prompts	×
85	q.confirm.return_value.unsafe_ask.side_effect = [	×
86	True,
87	True,
88	True,
89	True,
90	True,
91	True,
92	False,
93	]
94
95	config = server_configuration_questionaire()	×
96
97	keys = [	×
98	"description",
99	"ip",
100	"port",
101	"api_path",
102	"uri",
103	"allow_drop_all",
104	"logging",
105	"vpn_server",
106	"rabbitmq",
107	"two_factor_auth",
108	"algorithm_stores",
109	]
110
111	for key in keys:	×
112	self.assertIn(key, config)	×
113
114	@patch(f"{module_path}.node_configuration_questionaire")	×
115	@patch(f"{module_path}.server_configuration_questionaire")	×
116	@patch(f"{module_path}.ServerConfigurationManager")	×
117	@patch(f"{module_path}.NodeConfigurationManager")	×
NEW 118	@patch("vantage6.cli.configuration_create.AppContext")	×
NEW 119	def test_configuration_create_interface(	×
120	self, context, node_m, server_m, server_q, node_q
121	):
122	context.instance_folders.return_value = {"config": "/some/path/"}	×
123
NEW 124	file_ = make_configuration(InstanceType.NODE, "vtg6", False)	×
125	self.assertEqual(Path("/some/path/vtg6.yaml"), file_)	×
126
NEW 127	file_ = make_configuration(InstanceType.SERVER, "vtg6", True)	×
128	self.assertEqual(Path("/some/path/vtg6.yaml"), file_)	×
129
NEW 130	@patch("vantage6.cli.configuration_create.AppContext.available_configurations")	×
131	def test_select_configuration(self, available_configurations):	×
132	config = MagicMock()	×
133	config.name = "vtg6"	×
134
135	available_configurations.return_value = [[config], []]	×
136
137	with patch(f"{module_path}.q") as q:	×
138	q.select.return_value.unsafe_ask.return_value = "vtg6"	×
NEW 139	name = select_configuration_questionnaire(InstanceType.NODE, True)	×
140
141	self.assertEqual(name, "vtg6")	×

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