master.py

#  Copyright 2019-2021  Dom Sekotill <dom.sekotill@kodo.org.uk>
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.

"""
Master control client class
"""

import pathlib
from typing import Set

from ..types import PathLike
from . import consts
from .base import BaseClient
from .interfaces import InterfaceClient


class MasterClient(BaseClient):
	"""
	A client for listing, adding and removing interfaces, and getting per-client interfaces
	"""

	ctrl_dir = None

	async def connect(self, path: PathLike) -> None:
		if not isinstance(path, pathlib.Path):
			path = pathlib.Path(path)
		await super().connect(path)
		self.ctrl_dir = path.parent

	async def list_interfaces(self) -> Set[str]:
		"""
		Return a set of the interfaces currently managed by the daemon
		"""
		return await self.send_command(
			consts.COMMAND_INTERFACES, convert=lambda x: set(x.splitlines()),
		)

	async def add_interface(self, ifname: str, driver: str = "", driver_param: str = "") -> None:
		"""
		Add a network interface to the daemon's control interfaces
		"""
		if self.ctrl_dir:
			ctrl_iface = f"DIR={self.ctrl_dir} GROUP={self.ctrl_dir.group()}"
		else:
			# RuntimeError should be raised by send_command() as connect() does not appear
			# to have been called; set ctrl_iface to any string
			ctrl_iface = ""
		await self.send_command(
			consts.COMMAND_INTERFACE_ADD, ifname, "", driver, ctrl_iface, driver_param,
		)
		assert self.ctrl_dir is not None, \
			"RuntimeError should be raised for sends on unconnected clients; " \
			"or connect() may not have set ctrl_dir"

	async def remove_interface(self, ifname: str) -> None:  # pragma: no cover
		"""
		Remove a network interface from the daemon's control
		"""
		await self.send_command(consts.COMMAND_INTERFACE_REMOVE, ifname)

	async def connect_interface(self, ifname: str) -> InterfaceClient:
		"""
		Return an InterfaceClient controlling the named interface

		If the daemon is not currently managing the interface, it is added to the daemon's
		interfaces.
		"""
		if ifname not in await self.list_interfaces():
			await self.add_interface(ifname)
		assert self.ctrl_dir is not None, \
			"RuntimeError should be raised for sends on unconnected clients; " \
			"or connect() may not have set ctrl_dir"
		client = InterfaceClient(logger=self.logger)
		await client.connect(self.ctrl_dir.joinpath(ifname).as_posix())
		return client