Mercurial > libervia-backend
changeset 4301:9deb3ddb2921
CLI: add commands to handle gateways:
rel 449
author | Goffi <goffi@goffi.org> |
---|---|
date | Fri, 06 Sep 2024 18:03:01 +0200 (3 months ago) |
parents | 7ded09452875 |
children | 9e7ea54b93ee |
files | libervia/cli/cmd_gateway.py |
diffstat | 1 files changed, 178 insertions(+), 0 deletions(-) [+] |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/libervia/cli/cmd_gateway.py Fri Sep 06 18:03:01 2024 +0200 @@ -0,0 +1,178 @@ +#!/usr/bin/env python3 + +# Libervia CLI +# Copyright (C) 2009-2024 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + + +from libervia.backend.core.i18n import _ +from libervia.backend.tools.common import data_format +from libervia.cli import xmlui_manager +from libervia.cli.constants import Const as C +from . import base +from rich.table import Table + +__commands__ = ["Gateway"] + + +class List(base.CommandBase): + def __init__(self, host): + super().__init__( + host, + "list", + use_output=C.OUTPUT_COMPLEX, + extra_outputs={"default": self.default_output}, + use_verbose=True, + help=_("list gateways of a server"), + ) + + def add_parser_options(self): + self.parser.add_argument( + "server_jid", + nargs="?", + metavar="JID", + default="", + help=_("JID of the server to check, use own server if not specified"), + ) + + def default_output(self, data: dict) -> None: + available_table = Table( + "JID", + _("Name"), + _("Category"), + _("Type"), + title=":white_check_mark: " + _("Available Gateways"), + title_justify="left", + ) + for gateway in data["available"]: + identities = gateway["identities"] + names = "\n".join([identity["name"] for identity in identities]) + categories = "\n".join([identity["category"] for identity in identities]) + types = "\n".join([identity["type"] for identity in identities]) + available_table.add_row(gateway["entity"], names, categories, types) + + self.console.print(available_table) + + if self.verbosity: + # In verbose mode, we show unavailable gateways too. + self.console.print() + + if self.verbosity > 1: + self.console.print( + _("Following services are registered but don't answer:"), + "\n", + style="italic", + ) + + unavailable_table = Table( + "JID", + title=":x: " + _("Unavailable Services"), + title_justify="left", + ) + for gateway_jid in data["unavailable"]: + unavailable_table.add_row(gateway_jid) + + self.console.print(unavailable_table) + + async def start(self): + try: + found_gateways = data_format.deserialise( + await self.host.bridge.gateways_find(self.args.server_jid, self.profile) + ) + except Exception as e: + self.disp(f"can't find gateways for {self.args.server_jid}: {e}", error=True) + self.host.quit(C.EXIT_BRIDGE_ERRBACK) + else: + await self.output(found_gateways) + self.host.quit(C.EXIT_OK) + + +class Register(base.CommandBase): + def __init__(self, host): + super().__init__( + host, + "register", + help=_("register to a gateway"), + ) + + def add_parser_options(self): + self.parser.add_argument( + "gateway_jid", + metavar="JID", + help=_("JID of gateway to register to"), + ) + + async def start(self): + try: + xmlui_raw = await self.host.bridge.gateway_register( + self.args.gateway_jid, self.profile + ) + except Exception as e: + self.disp(f"can't register to gateway: {e}", error=True) + self.host.quit(C.EXIT_BRIDGE_ERRBACK) + else: + xmlui = xmlui_manager.create(self.host, xmlui_raw) + await xmlui.show() + # FIXME: Registration errors are not correctly managed. + ret = await xmlui.submit_form() + self.host.quit(C.EXIT_OK) + + +class Unregister(base.CommandBase): + def __init__(self, host): + super().__init__( + host, + "unregister", + help=_("unregister from a gateway"), + ) + + def add_parser_options(self): + self.parser.add_argument( + "-f", + "--force", + action="store_true", + help=_("unregister without confirmation"), + ) + self.parser.add_argument( + "gateway_jid", + metavar="JID", + help=_("JID of gateway to unregister from"), + ) + + async def start(self): + if not self.args.force: + await self.host.confirm_or_quit( + _( + "Are you sure that you want to unregister from {gateway_jid}?" + ).format(gateway_jid=self.args.gateway_jid), + _("Gateway unregistration cancelled.") + ) + + try: + await self.host.bridge.in_band_unregister( + self.args.gateway_jid, self.profile + ) + except Exception as e: + self.disp(f"can't unregister from gateway: {e}", error=True) + self.host.quit(C.EXIT_BRIDGE_ERRBACK) + else: + self.host.quit(C.EXIT_OK) + + +class Gateway(base.CommandBase): + subcommands = (List, Register, Unregister) + + def __init__(self, host): + super().__init__(host, "gateway", use_profile=False, help=_("gateway handling"))