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

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

from click.testing import CliRunner

from vantage6.common.globals import APPNAME, InstanceType

from vantage6.cli.common.utils import attach_logs
from vantage6.cli.server.attach import cli_server_attach
from vantage6.cli.server.files import cli_server_files
from vantage6.cli.server.import_ import cli_server_import
from vantage6.cli.server.list import cli_server_configuration_list
from vantage6.cli.server.new import cli_server_new
from vantage6.cli.server.start import cli_server_start
from vantage6.cli.server.stop import cli_server_stop


class ServerCLITest(unittest.TestCase):
    @patch("vantage6.cli.server.start.NetworkManager")
    @patch("vantage6.cli.server.start.docker.types.Mount")
    @patch("os.makedirs")
    @patch("vantage6.cli.server.start.pull_infra_image")
    @patch("vantage6.cli.common.decorator.get_context")
    @patch("vantage6.cli.server.start.docker.from_env")
    @patch("vantage6.cli.common.start.check_docker_running", return_value=True)
    def test_start(
        self,
        docker_check,
        containers,
        context,
        pull,
        os_makedirs,
        mount,
        network_manager,
    ):
        """Start server without errors"""
        container1 = MagicMock()
        container1.containers.name = f"{APPNAME}-iknl-system"
        containers.containers.list.return_value = [container1]
        containers.containers.run.return_value = True

        # mount.types.Mount.return_value = MagicMock()

        ctx = MagicMock(
            config={"uri": "sqlite:///file.db", "port": 9999},
            config_file="/config.yaml",
            data_dir=Path("."),
        )
        ctx.config_exists.return_value = True
        ctx.name = "not-running"
        context.return_value = ctx

        runner = CliRunner()
        result = runner.invoke(cli_server_start, ["--name", "not-running"])

        self.assertEqual(result.exit_code, 0)

    @patch("vantage6.cli.server.common.ServerContext")
    @patch("docker.DockerClient.containers")
    @patch("vantage6.cli.server.list.check_docker_running", return_value=True)
    def test_configuration_list(self, docker_check, containers, context):
        """Configuration list without errors."""
        container1 = MagicMock()
        container1.name = f"{APPNAME}-iknl-system"
        containers.list.return_value = [container1]

        config = MagicMock()
        config.name = "iknl"
        context.available_configurations.return_value = ([config], [])

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

        self.assertEqual(result.exit_code, 0)
        self.assertIsNone(result.exception)

    @patch("vantage6.cli.common.decorator.get_context")
    def test_files(self, context):
        """Configuration files without errors."""

        ctx = context.return_value = MagicMock(
            log_file="/log_file.log", config_file="/iknl.yaml"
        )
        ctx.get_database_uri.return_value = "sqlite:///test.db"

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

        self.assertIsNone(result.exception)
        self.assertEqual(result.exit_code, 0)

    @patch("docker.DockerClient.containers")
    @patch("vantage6.cli.server.import_.print_log_worker")
    @patch("vantage6.cli.server.import_.click.Path")
    @patch("vantage6.cli.server.import_.pull_image")
    @patch("vantage6.cli.server.import_.check_docker_running", return_value=True)
    @patch("vantage6.cli.common.decorator.get_context")
    def test_import(self, context, docker_check, pull, click_path, log, containers):
        """Import entities without errors."""
        click_path.return_value = MagicMock()

        ctx = MagicMock()
        ctx.name = "some-name"
        context.return_value = ctx

        runner = CliRunner()
        with runner.isolated_filesystem():
            with open("some.yaml", "w") as fp:
                fp.write("does-not-matter")
            result = runner.invoke(cli_server_import, ["--name", "iknl", "some.yaml"])

        self.assertIsNone(result.exception)
        self.assertEqual(result.exit_code, 0)

    @patch("vantage6.cli.server.new.make_configuration")
    @patch("vantage6.cli.server.new.ensure_config_dir_writable")
    @patch("vantage6.cli.server.new.ServerContext")
    def test_new(self, context, permissions, make_configuration):
        """New configuration without errors."""

        context.config_exists.return_value = False
        permissions.return_value = True
        make_configuration.return_value = "/some/file.yaml"

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

        self.assertIsNone(result.exception)
        self.assertEqual(result.exit_code, 0)

    @patch("vantage6.cli.server.stop.docker.from_env")
    def test_stop(self, containers):
        """Stop server without errors."""

        container1 = MagicMock()
        container1.name = f"{APPNAME}-iknl-system-{InstanceType.SERVER.value}"
        containers.containers.list.return_value = [container1]

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

        self.assertIsNone(result.exception)
        self.assertEqual(result.exit_code, 0)

    @patch("vantage6.cli.server.attach.attach_logs")
    def test_attach(self, attach_logs):
        """Attach logs to the console without errors."""
        runner = CliRunner()
        result = runner.invoke(cli_server_attach)

        self.assertIsNone(result.exception)
        self.assertEqual(result.exit_code, 0)
        attach_logs.assert_called_once_with(
            "app=vantage6-server", "component=vantage6-server"
        )

    @patch("vantage6.cli.common.utils.Popen")
    def test_attach_logs(self, mock_popen):
        # Mock the Popen instance and its methods
        mock_process = mock_popen.return_value
        mock_process.wait.return_value = None

        # Call the function with a sample label
        attach_logs("app=node", "env=dev")

        # Construct the expected command
        expected_command = [
            "devspace",
            "logs",
            "--follow",
            "--label-selector",
            "app=node,env=dev",
        ]

        # Verify that Popen was called with the expected command
        mock_popen.assert_called_once_with(expected_command, stdout=None, stderr=None)

        # Verify that wait was called on the process
        mock_process.wait.assert_called_once()

1	import unittest	×
2	from pathlib import Path	×
3	from unittest.mock import MagicMock, patch	×
4
5	from click.testing import CliRunner	×
6
7	from vantage6.common.globals import APPNAME, InstanceType	×
8
9	from vantage6.cli.common.utils import attach_logs	×
10	from vantage6.cli.server.attach import cli_server_attach	×
11	from vantage6.cli.server.files import cli_server_files	×
12	from vantage6.cli.server.import_ import cli_server_import	×
13	from vantage6.cli.server.list import cli_server_configuration_list	×
14	from vantage6.cli.server.new import cli_server_new	×
15	from vantage6.cli.server.start import cli_server_start	×
16	from vantage6.cli.server.stop import cli_server_stop	×
17
18
19	class ServerCLITest(unittest.TestCase):	×
20	@patch("vantage6.cli.server.start.NetworkManager")	×
21	@patch("vantage6.cli.server.start.docker.types.Mount")	×
22	@patch("os.makedirs")	×
23	@patch("vantage6.cli.server.start.pull_infra_image")	×
24	@patch("vantage6.cli.common.decorator.get_context")	×
25	@patch("vantage6.cli.server.start.docker.from_env")	×
26	@patch("vantage6.cli.common.start.check_docker_running", return_value=True)	×
27	def test_start(	×
28	self,
29	docker_check,
30	containers,
31	context,
32	pull,
33	os_makedirs,
34	mount,
35	network_manager,
36	):
37	"""Start server without errors"""
38	container1 = MagicMock()	×
39	container1.containers.name = f"{APPNAME}-iknl-system"	×
40	containers.containers.list.return_value = [container1]	×
41	containers.containers.run.return_value = True	×
42
43	# mount.types.Mount.return_value = MagicMock()
44
45	ctx = MagicMock(	×
46	config={"uri": "sqlite:///file.db", "port": 9999},
47	config_file="/config.yaml",
48	data_dir=Path("."),
49	)
50	ctx.config_exists.return_value = True	×
51	ctx.name = "not-running"	×
52	context.return_value = ctx	×
53
54	runner = CliRunner()	×
55	result = runner.invoke(cli_server_start, ["--name", "not-running"])	×
56
57	self.assertEqual(result.exit_code, 0)	×
58
59	@patch("vantage6.cli.server.common.ServerContext")	×
60	@patch("docker.DockerClient.containers")	×
61	@patch("vantage6.cli.server.list.check_docker_running", return_value=True)	×
62	def test_configuration_list(self, docker_check, containers, context):	×
63	"""Configuration list without errors."""
64	container1 = MagicMock()	×
65	container1.name = f"{APPNAME}-iknl-system"	×
66	containers.list.return_value = [container1]	×
67
68	config = MagicMock()	×
69	config.name = "iknl"	×
70	context.available_configurations.return_value = ([config], [])	×
71
72	runner = CliRunner()	×
73	result = runner.invoke(cli_server_configuration_list)	×
74
75	self.assertEqual(result.exit_code, 0)	×
76	self.assertIsNone(result.exception)	×
77
78	@patch("vantage6.cli.common.decorator.get_context")	×
79	def test_files(self, context):	×
80	"""Configuration files without errors."""
81
82	ctx = context.return_value = MagicMock(	×
83	log_file="/log_file.log", config_file="/iknl.yaml"
84	)
85	ctx.get_database_uri.return_value = "sqlite:///test.db"	×
86
87	runner = CliRunner()	×
88	result = runner.invoke(cli_server_files, ["--name", "iknl"])	×
89
90	self.assertIsNone(result.exception)	×
91	self.assertEqual(result.exit_code, 0)	×
92
93	@patch("docker.DockerClient.containers")	×
94	@patch("vantage6.cli.server.import_.print_log_worker")	×
95	@patch("vantage6.cli.server.import_.click.Path")	×
96	@patch("vantage6.cli.server.import_.pull_image")	×
97	@patch("vantage6.cli.server.import_.check_docker_running", return_value=True)	×
98	@patch("vantage6.cli.common.decorator.get_context")	×
99	def test_import(self, context, docker_check, pull, click_path, log, containers):	×
100	"""Import entities without errors."""
101	click_path.return_value = MagicMock()	×
102
103	ctx = MagicMock()	×
104	ctx.name = "some-name"	×
105	context.return_value = ctx	×
106
107	runner = CliRunner()	×
108	with runner.isolated_filesystem():	×
109	with open("some.yaml", "w") as fp:	×
110	fp.write("does-not-matter")	×
111	result = runner.invoke(cli_server_import, ["--name", "iknl", "some.yaml"])	×
112
113	self.assertIsNone(result.exception)	×
114	self.assertEqual(result.exit_code, 0)	×
115
NEW 116	@patch("vantage6.cli.server.new.make_configuration")	×
117	@patch("vantage6.cli.server.new.ensure_config_dir_writable")	×
118	@patch("vantage6.cli.server.new.ServerContext")	×
NEW 119	def test_new(self, context, permissions, make_configuration):	×
120	"""New configuration without errors."""
121
122	context.config_exists.return_value = False	×
123	permissions.return_value = True	×
NEW 124	make_configuration.return_value = "/some/file.yaml"	×
125
126	runner = CliRunner()	×
127	result = runner.invoke(cli_server_new, ["--name", "iknl"])	×
128
129	self.assertIsNone(result.exception)	×
130	self.assertEqual(result.exit_code, 0)	×
131
132	@patch("vantage6.cli.server.stop.docker.from_env")	×
133	def test_stop(self, containers):	×
134	"""Stop server without errors."""
135
136	container1 = MagicMock()	×
137	container1.name = f"{APPNAME}-iknl-system-{InstanceType.SERVER.value}"	×
138	containers.containers.list.return_value = [container1]	×
139
140	runner = CliRunner()	×
141	result = runner.invoke(cli_server_stop, ["--name", "iknl"])	×
142
143	self.assertIsNone(result.exception)	×
144	self.assertEqual(result.exit_code, 0)	×
145
146	@patch("vantage6.cli.server.attach.attach_logs")	×
147	def test_attach(self, attach_logs):	×
148	"""Attach logs to the console without errors."""
149	runner = CliRunner()	×
150	result = runner.invoke(cli_server_attach)	×
151
152	self.assertIsNone(result.exception)	×
153	self.assertEqual(result.exit_code, 0)	×
154	attach_logs.assert_called_once_with(	×
155	"app=vantage6-server", "component=vantage6-server"
156	)
157
158	@patch("vantage6.cli.common.utils.Popen")	×
159	def test_attach_logs(self, mock_popen):	×
160	# Mock the Popen instance and its methods
161	mock_process = mock_popen.return_value	×
162	mock_process.wait.return_value = None	×
163
164	# Call the function with a sample label
165	attach_logs("app=node", "env=dev")	×
166
167	# Construct the expected command
168	expected_command = [	×
169	"devspace",
170	"logs",
171	"--follow",
172	"--label-selector",
173	"app=node,env=dev",
174	]
175
176	# Verify that Popen was called with the expected command
177	mock_popen.assert_called_once_with(expected_command, stdout=None, stderr=None)	×
178
179	# Verify that wait was called on the process
180	mock_process.wait.assert_called_once()	×

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