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

57.89

/vantage6/vantage6/cli/context/node.py

from __future__ import annotations

import hashlib
import os.path
from pathlib import Path

from vantage6.common.context import AppContext
from vantage6.common.globals import APPNAME, STRING_ENCODING, InstanceType

from vantage6.cli import __version__
from vantage6.cli.configuration_manager import NodeConfigurationManager
from vantage6.cli.globals import DEFAULT_NODE_SYSTEM_FOLDERS as N_FOL


class NodeContext(AppContext):
    """
    Node context object for the host system.

    Parameters
    ----------
    instance_name : str
        Name of the configuration instance, corresponds to the filename
        of the configuration file.
    system_folders : bool, optional
        _description_, by default N_FOL
    config_file : str, optional
        _description_, by default None
    in_container : bool, optional
        Whether the application is running inside a container, by default False
    is_sandbox : bool, optional
        Whether the configuration is a sandbox configuration, by default False
    """

    # The server configuration manager is aware of the structure of the server
    # configuration file and makes sure only valid configuration can be loaded.
    INST_CONFIG_MANAGER = NodeConfigurationManager

    def __init__(
        self,
        instance_name: str,
        system_folders: bool = N_FOL,
        config_file: str = None,
        print_log_header: bool = True,
        logger_prefix: str = "",
        in_container: bool = False,
        is_sandbox: bool = False,
    ):
        super().__init__(
            InstanceType.NODE,
            instance_name,
            system_folders=system_folders,
            config_file=config_file,
            print_log_header=print_log_header,
            logger_prefix=logger_prefix,
            in_container=in_container,
            is_sandbox=is_sandbox,
        )
        if print_log_header:
            self.log.info("vantage6 version '%s'", __version__)
        self.identifier = self.__create_node_identifier()

    @classmethod
    def from_external_config_file(
        cls, path: str, system_folders: bool = N_FOL, is_sandbox: bool = False
    ) -> NodeContext:
        """
        Create a node context from an external configuration file. External
        means that the configuration file is not located in the default folders
        but its location is specified by the user.

        Parameters
        ----------
        path : str
            Path of the configuration file
        system_folders : bool, optional
            System wide or user configuration, by default N_FOL

        Returns
        -------
        NodeContext
            Node context object
        """
        return super().from_external_config_file(
            Path(path).resolve(),
            InstanceType.NODE,
            system_folders,
            is_sandbox=is_sandbox,
        )

    @classmethod
    def config_exists(
        cls, instance_name: str, system_folders: bool = N_FOL, is_sandbox: bool = False
    ) -> bool:
        """
        Check if a configuration file exists.

        Parameters
        ----------
        instance_name : str
            Name of the configuration instance, corresponds to the filename
            of the configuration file.
        system_folders : bool, optional
            System wide or user configuration, by default N_FOL
        is_sandbox : bool, optional
            Whether the configuration is a sandbox configuration, by default False
        Returns
        -------
        bool
            Whether the configuration file exists or not
        """
        return super().config_exists(
            InstanceType.NODE,
            instance_name,
            system_folders=system_folders,
            is_sandbox=is_sandbox,
        )

    @classmethod
    def available_configurations(
        cls, system_folders: bool = N_FOL, is_sandbox: bool = False
    ) -> tuple[list, list]:
        """
        Find all available server configurations in the default folders.

        Parameters
        ----------
        system_folders : bool, optional
            System wide or user configuration, by default N_FOL
        is_sandbox : bool, optional
            Whether the configuration is a sandbox configuration, by default False

        Returns
        -------
        tuple[list, list]
            The first list contains validated configuration files, the second
            list contains invalid configuration files.
        """
        return super().available_configurations(
            InstanceType.NODE, system_folders, is_sandbox
        )

    @staticmethod
    def type_data_folder(system_folders: bool = N_FOL) -> Path:
        """
        Obtain OS specific data folder where to store node specific data.

        Parameters
        ----------
        system_folders : bool, optional
            System wide or user configuration

        Returns
        -------
        Path
            Path to the data folder
        """
        return AppContext.type_data_folder(InstanceType.NODE, system_folders)

    @property
    def databases(self) -> dict:
        """
        Dictionary of local databases that are available for this node.

        Returns
        -------
        dict
            dictionary with database names as keys and their corresponding
            paths as values.
        """
        if self.in_container:
            return self.config["databases"]
        else:
            return self.config["node"]["databases"]

    @property
    def docker_container_name(self) -> str:
        """
        Docker container name of the node.

        Returns
        -------
        str
            Node's Docker container name
        """
        return f"{APPNAME}-{self.name}-{self.scope}"

    @property
    def docker_network_name(self) -> str:
        """
        Private Docker network name which is unique for this node.

        Returns
        -------
        str
            Docker network name
        """
        return f"{APPNAME}-{self.name}-{self.scope}-net"

    @property
    def docker_volume_name(self) -> str:
        """
        Docker volume in which task data is stored. In case a file based
        database is used, this volume contains the database file as well.

        Returns
        -------
        str
            Docker volume name
        """
        return os.environ.get("DATA_VOLUME_NAME", f"{self.docker_container_name}-vol")

    @property
    def proxy_log_file(self):
        return self.log_file_name(type_="proxy_server")

    def get_database_uri(self, label: str = "default") -> str:
        """
        Obtain the database URI for a specific database.

        Parameters
        ----------
        label : str, optional
            Database label, by default "default"

        Returns
        -------
        str
            URI to the database
        """
        return self.config["databases"][label]

    def set_folders(
        self, instance_type: InstanceType, instance_name: str, system_folders: bool
    ) -> None:
        """
        Set the folders for the node.

        Parameters
        ----------
        instance_type : InstanceType
            Instance type
        instance_name : str
            Instance name
        system_folders : bool
            Whether to use system folders
        """
        dirs = self.instance_folders(instance_type, instance_name, system_folders)

        self.log_dir = dirs.get("log")
        self.data_dir = dirs.get("data")
        self.config_dir = dirs.get("config")

    def __create_node_identifier(self) -> str:
        """
        Create a unique identifier for the node.

        Returns
        -------
        str
            Unique identifier for the node
        """
        if self.in_container:
            api_key = os.environ.get("V6_API_KEY")
        else:
            api_key = self.config["node"]["apiKey"]

        return hashlib.sha256(api_key.encode(STRING_ENCODING)).hexdigest()[:16]

1	from __future__ import annotations	1✔
2
3	import hashlib	1✔
4	import os.path	1✔
5	from pathlib import Path	1✔
6
7	from vantage6.common.context import AppContext	1✔
8	from vantage6.common.globals import APPNAME, STRING_ENCODING, InstanceType	1✔
9
10	from vantage6.cli import __version__	1✔
11	from vantage6.cli.configuration_manager import NodeConfigurationManager	1✔
12	from vantage6.cli.globals import DEFAULT_NODE_SYSTEM_FOLDERS as N_FOL	1✔
13
14
15	class NodeContext(AppContext):	1✔
16	"""
17	Node context object for the host system.
18
19	Parameters
20	----------
21	instance_name : str
22	Name of the configuration instance, corresponds to the filename
23	of the configuration file.
24	system_folders : bool, optional
25	_description_, by default N_FOL
26	config_file : str, optional
27	_description_, by default None
28	in_container : bool, optional
29	Whether the application is running inside a container, by default False
30	is_sandbox : bool, optional
31	Whether the configuration is a sandbox configuration, by default False
32	"""
33
34	# The server configuration manager is aware of the structure of the server
35	# configuration file and makes sure only valid configuration can be loaded.
36	INST_CONFIG_MANAGER = NodeConfigurationManager	1✔
37
38	def __init__(	1✔
39	self,
40	instance_name: str,
41	system_folders: bool = N_FOL,
42	config_file: str = None,
43	print_log_header: bool = True,
44	logger_prefix: str = "",
45	in_container: bool = False,
46	is_sandbox: bool = False,
47	):
48	super().__init__(	×
49	InstanceType.NODE,
50	instance_name,
51	system_folders=system_folders,
52	config_file=config_file,
53	print_log_header=print_log_header,
54	logger_prefix=logger_prefix,
55	in_container=in_container,
56	is_sandbox=is_sandbox,
57	)
58	if print_log_header:	×
59	self.log.info("vantage6 version '%s'", __version__)	×
60	self.identifier = self.__create_node_identifier()	×
61
62	@classmethod	1✔
63	def from_external_config_file(	1✔
64	cls, path: str, system_folders: bool = N_FOL, is_sandbox: bool = False
65	) -> NodeContext:
66	"""
67	Create a node context from an external configuration file. External
68	means that the configuration file is not located in the default folders
69	but its location is specified by the user.
70
71	Parameters
72	----------
73	path : str
74	Path of the configuration file
75	system_folders : bool, optional
76	System wide or user configuration, by default N_FOL
77
78	Returns
79	-------
80	NodeContext
81	Node context object
82	"""
83	return super().from_external_config_file(	×
84	Path(path).resolve(),
85	InstanceType.NODE,
86	system_folders,
87	is_sandbox=is_sandbox,
88	)
89
90	@classmethod	1✔
91	def config_exists(	1✔
92	cls, instance_name: str, system_folders: bool = N_FOL, is_sandbox: bool = False
93	) -> bool:
94	"""
95	Check if a configuration file exists.
96
97	Parameters
98	----------
99	instance_name : str
100	Name of the configuration instance, corresponds to the filename
101	of the configuration file.
102	system_folders : bool, optional
103	System wide or user configuration, by default N_FOL
104	is_sandbox : bool, optional
105	Whether the configuration is a sandbox configuration, by default False
106	Returns
107	-------
108	bool
109	Whether the configuration file exists or not
110	"""
111	return super().config_exists(	×
112	InstanceType.NODE,
113	instance_name,
114	system_folders=system_folders,
115	is_sandbox=is_sandbox,
116	)
117
118	@classmethod	1✔
119	def available_configurations(	1✔
120	cls, system_folders: bool = N_FOL, is_sandbox: bool = False
121	) -> tuple[list, list]:
122	"""
123	Find all available server configurations in the default folders.
124
125	Parameters
126	----------
127	system_folders : bool, optional
128	System wide or user configuration, by default N_FOL
129	is_sandbox : bool, optional
130	Whether the configuration is a sandbox configuration, by default False
131
132	Returns
133	-------
134	tuple[list, list]
135	The first list contains validated configuration files, the second
136	list contains invalid configuration files.
137	"""
NEW 138	return super().available_configurations(	×
139	InstanceType.NODE, system_folders, is_sandbox
140	)
141
142	@staticmethod	1✔
143	def type_data_folder(system_folders: bool = N_FOL) -> Path:	1✔
144	"""
145	Obtain OS specific data folder where to store node specific data.
146
147	Parameters
148	----------
149	system_folders : bool, optional
150	System wide or user configuration
151
152	Returns
153	-------
154	Path
155	Path to the data folder
156	"""
157	return AppContext.type_data_folder(InstanceType.NODE, system_folders)	×
158
159	@property	1✔
160	def databases(self) -> dict:	1✔
161	"""
162	Dictionary of local databases that are available for this node.
163
164	Returns
165	-------
166	dict
167	dictionary with database names as keys and their corresponding
168	paths as values.
169	"""
170	if self.in_container:	×
171	return self.config["databases"]	×
172	else:
173	return self.config["node"]["databases"]	×
174
175	@property	1✔
176	def docker_container_name(self) -> str:	1✔
177	"""
178	Docker container name of the node.
179
180	Returns
181	-------
182	str
183	Node's Docker container name
184	"""
185	return f"{APPNAME}-{self.name}-{self.scope}"	×
186
187	@property	1✔
188	def docker_network_name(self) -> str:	1✔
189	"""
190	Private Docker network name which is unique for this node.
191
192	Returns
193	-------
194	str
195	Docker network name
196	"""
197	return f"{APPNAME}-{self.name}-{self.scope}-net"	×
198
199	@property	1✔
200	def docker_volume_name(self) -> str:	1✔
201	"""
202	Docker volume in which task data is stored. In case a file based
203	database is used, this volume contains the database file as well.
204
205	Returns
206	-------
207	str
208	Docker volume name
209	"""
210	return os.environ.get("DATA_VOLUME_NAME", f"{self.docker_container_name}-vol")	×
211
212	@property	1✔
213	def proxy_log_file(self):	1✔
214	return self.log_file_name(type_="proxy_server")	×
215
216	def get_database_uri(self, label: str = "default") -> str:	1✔
217	"""
218	Obtain the database URI for a specific database.
219
220	Parameters
221	----------
222	label : str, optional
223	Database label, by default "default"
224
225	Returns
226	-------
227	str
228	URI to the database
229	"""
230	return self.config["databases"][label]	×
231
232	def set_folders(	1✔
233	self, instance_type: InstanceType, instance_name: str, system_folders: bool
234	) -> None:
235	"""
236	Set the folders for the node.
237
238	Parameters
239	----------
240	instance_type : InstanceType
241	Instance type
242	instance_name : str
243	Instance name
244	system_folders : bool
245	Whether to use system folders
246	"""
247	dirs = self.instance_folders(instance_type, instance_name, system_folders)	×
248
249	self.log_dir = dirs.get("log")	×
250	self.data_dir = dirs.get("data")	×
251	self.config_dir = dirs.get("config")	×
252
253	def __create_node_identifier(self) -> str:	1✔
254	"""
255	Create a unique identifier for the node.
256
257	Returns
258	-------
259	str
260	Unique identifier for the node
261	"""
262	if self.in_container:	×
263	api_key = os.environ.get("V6_API_KEY")	×
264	else:
265	api_key = self.config["node"]["apiKey"]	×
266
267	return hashlib.sha256(api_key.encode(STRING_ENCODING)).hexdigest()[:16]	×

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