Mercurial > libervia-backend

comparison libervia/cli/cmd_gateway.py @ 4301:9deb3ddb2921

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
CLI: add commands to handle gateways: rel 449
author Goffi <goffi@goffi.org>
date Fri, 06 Sep 2024 18:03:01 +0200
parents
children
comparison
equal deleted inserted replaced
4300:7ded09452875 4301:9deb3ddb2921
1 #!/usr/bin/env python3
2
3 # Libervia CLI
4 # Copyright (C) 2009-2024 Jérôme Poisson (goffi@goffi.org)
5
6 # This program is free software: you can redistribute it and/or modify
7 # it under the terms of the GNU Affero General Public License as published by
8 # the Free Software Foundation, either version 3 of the License, or
9 # (at your option) any later version.
10
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU Affero General Public License for more details.
15
16 # You should have received a copy of the GNU Affero General Public License
17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19
20 from libervia.backend.core.i18n import _
21 from libervia.backend.tools.common import data_format
22 from libervia.cli import xmlui_manager
23 from libervia.cli.constants import Const as C
24 from . import base
25 from rich.table import Table
26
27 __commands__ = ["Gateway"]
28
29
30 class List(base.CommandBase):
31 def __init__(self, host):
32 super().__init__(
33 host,
34 "list",
35 use_output=C.OUTPUT_COMPLEX,
36 extra_outputs={"default": self.default_output},
37 use_verbose=True,
38 help=_("list gateways of a server"),
39 )
40
41 def add_parser_options(self):
42 self.parser.add_argument(
43 "server_jid",
44 nargs="?",
45 metavar="JID",
46 default="",
47 help=_("JID of the server to check, use own server if not specified"),
48 )
49
50 def default_output(self, data: dict) -> None:
51 available_table = Table(
52 "JID",
53 _("Name"),
54 _("Category"),
55 _("Type"),
56 title=":white_check_mark: " + _("Available Gateways"),
57 title_justify="left",
58 )
59 for gateway in data["available"]:
60 identities = gateway["identities"]
61 names = "\n".join([identity["name"] for identity in identities])
62 categories = "\n".join([identity["category"] for identity in identities])
63 types = "\n".join([identity["type"] for identity in identities])
64 available_table.add_row(gateway["entity"], names, categories, types)
65
66 self.console.print(available_table)
67
68 if self.verbosity:
69 # In verbose mode, we show unavailable gateways too.
70 self.console.print()
71
72 if self.verbosity > 1:
73 self.console.print(
74 _("Following services are registered but don't answer:"),
75 "\n",
76 style="italic",
77 )
78
79 unavailable_table = Table(
80 "JID",
81 title=":x: " + _("Unavailable Services"),
82 title_justify="left",
83 )
84 for gateway_jid in data["unavailable"]:
85 unavailable_table.add_row(gateway_jid)
86
87 self.console.print(unavailable_table)
88
89 async def start(self):
90 try:
91 found_gateways = data_format.deserialise(
92 await self.host.bridge.gateways_find(self.args.server_jid, self.profile)
93 )
94 except Exception as e:
95 self.disp(f"can't find gateways for {self.args.server_jid}: {e}", error=True)
96 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
97 else:
98 await self.output(found_gateways)
99 self.host.quit(C.EXIT_OK)
100
101
102 class Register(base.CommandBase):
103 def __init__(self, host):
104 super().__init__(
105 host,
106 "register",
107 help=_("register to a gateway"),
108 )
109
110 def add_parser_options(self):
111 self.parser.add_argument(
112 "gateway_jid",
113 metavar="JID",
114 help=_("JID of gateway to register to"),
115 )
116
117 async def start(self):
118 try:
119 xmlui_raw = await self.host.bridge.gateway_register(
120 self.args.gateway_jid, self.profile
121 )
122 except Exception as e:
123 self.disp(f"can't register to gateway: {e}", error=True)
124 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
125 else:
126 xmlui = xmlui_manager.create(self.host, xmlui_raw)
127 await xmlui.show()
128 # FIXME: Registration errors are not correctly managed.
129 ret = await xmlui.submit_form()
130 self.host.quit(C.EXIT_OK)
131
132
133 class Unregister(base.CommandBase):
134 def __init__(self, host):
135 super().__init__(
136 host,
137 "unregister",
138 help=_("unregister from a gateway"),
139 )
140
141 def add_parser_options(self):
142 self.parser.add_argument(
143 "-f",
144 "--force",
145 action="store_true",
146 help=_("unregister without confirmation"),
147 )
148 self.parser.add_argument(
149 "gateway_jid",
150 metavar="JID",
151 help=_("JID of gateway to unregister from"),
152 )
153
154 async def start(self):
155 if not self.args.force:
156 await self.host.confirm_or_quit(
157 _(
158 "Are you sure that you want to unregister from {gateway_jid}?"
159 ).format(gateway_jid=self.args.gateway_jid),
160 _("Gateway unregistration cancelled.")
161 )
162
163 try:
164 await self.host.bridge.in_band_unregister(
165 self.args.gateway_jid, self.profile
166 )
167 except Exception as e:
168 self.disp(f"can't unregister from gateway: {e}", error=True)
169 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
170 else:
171 self.host.quit(C.EXIT_OK)
172
173
174 class Gateway(base.CommandBase):
175 subcommands = (List, Register, Unregister)
176
177 def __init__(self, host):
178 super().__init__(host, "gateway", use_profile=False, help=_("gateway handling"))