YunoHost/yunohost
1
0
Fork 0
mirror of https://github.com/YunoHost/yunohost.git synced 2024-09-03 20:06:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

codequality: Apply black on python script in bin/

Browse source
This commit is contained in:
Alexandre Aubin 2021-11-15 04:00:47 +01:00
parent f824d9e0ea
commit 467c04f382
3 changed files with 85 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

51
bin/yunohost
View file

@ -12,37 +12,42 @@ import yunohost
def _parse_cli_args(): def _parse_cli_args():
"""Parse additional arguments for the cli""" """Parse additional arguments for the cli"""
parser = argparse.ArgumentParser(add_help=False) parser = argparse.ArgumentParser(add_help=False)
parser.add_argument('--output-as', parser.add_argument(
choices=['json', 'plain', 'none'], default=None, "--output-as",
help="Output result in another format" choices=["json", "plain", "none"],
default=None,
help="Output result in another format",
) )
parser.add_argument('--debug', parser.add_argument(
action='store_true', default=False, "--debug",
help="Log and print debug messages" action="store_true",
default=False,
help="Log and print debug messages",
) )
parser.add_argument('--quiet', parser.add_argument(
action='store_true', default=False, "--quiet", action="store_true", default=False, help="Don't produce any output"
help="Don't produce any output"
) )
parser.add_argument('--timeout', parser.add_argument(
type=int, default=None, "--timeout",
help="Number of seconds before this command will timeout because it can't acquire the lock (meaning that another command is currently running), by default there is no timeout and the command will wait until it can get the lock" type=int,
default=None,
help="Number of seconds before this command will timeout because it can't acquire the lock (meaning that another command is currently running), by default there is no timeout and the command will wait until it can get the lock",
) )
# deprecated arguments # deprecated arguments
parser.add_argument('--plain', parser.add_argument(
action='store_true', default=False, help=argparse.SUPPRESS "--plain", action="store_true", default=False, help=argparse.SUPPRESS
) )
parser.add_argument('--json', parser.add_argument(
action='store_true', default=False, help=argparse.SUPPRESS "--json", action="store_true", default=False, help=argparse.SUPPRESS
) )
opts, args = parser.parse_known_args() opts, args = parser.parse_known_args()
# output compatibility # output compatibility
if opts.plain: if opts.plain:
opts.output_as = 'plain' opts.output_as = "plain"
elif opts.json: elif opts.json:
opts.output_as = 'json' opts.output_as = "json"
return (parser, opts, args) return (parser, opts, args)
@ -54,10 +59,12 @@ if os.environ["PATH"] != default_path:
# Main action ---------------------------------------------------------- # Main action ----------------------------------------------------------
if __name__ == '__main__': if __name__ == "__main__":
if os.geteuid() != 0: if os.geteuid() != 0:
sys.stderr.write("\033[1;31mError:\033[0m yunohost command must be " sys.stderr.write(
"run as root or with sudo.\n") "\033[1;31mError:\033[0m yunohost command must be "
"run as root or with sudo.\n"
)
sys.exit(1) sys.exit(1)
parser, opts, args = _parse_cli_args() parser, opts, args = _parse_cli_args()
@ -69,5 +76,5 @@ if __name__ == '__main__':
output_as=opts.output_as, output_as=opts.output_as,
timeout=opts.timeout, timeout=opts.timeout,
args=args, args=args,
parser=parser parser=parser,
) )

38
bin/yunohost-api
View file

@ -8,37 +8,49 @@ sys.path.insert(0, "/usr/lib/moulinette/")
import yunohost import yunohost
# Default server configuration # Default server configuration
DEFAULT_HOST = 'localhost' DEFAULT_HOST = "localhost"
DEFAULT_PORT = 6787 DEFAULT_PORT = 6787
def _parse_api_args(): def _parse_api_args():
"""Parse main arguments for the api""" """Parse main arguments for the api"""
parser = argparse.ArgumentParser(add_help=False, parser = argparse.ArgumentParser(
add_help=False,
description="Run the YunoHost API to manage your server.", description="Run the YunoHost API to manage your server.",
) )
srv_group = parser.add_argument_group('server configuration') srv_group = parser.add_argument_group("server configuration")
srv_group.add_argument('-h', '--host', srv_group.add_argument(
action='store', default=DEFAULT_HOST, "-h",
"--host",
action="store",
default=DEFAULT_HOST,
help="Host to listen on (default: %s)" % DEFAULT_HOST, help="Host to listen on (default: %s)" % DEFAULT_HOST,
) )
srv_group.add_argument('-p', '--port', srv_group.add_argument(
action='store', default=DEFAULT_PORT, type=int, "-p",
"--port",
action="store",
default=DEFAULT_PORT,
type=int,
help="Port to listen on (default: %d)" % DEFAULT_PORT, help="Port to listen on (default: %d)" % DEFAULT_PORT,
) )
glob_group = parser.add_argument_group('global arguments') glob_group = parser.add_argument_group("global arguments")
glob_group.add_argument('--debug', glob_group.add_argument(
action='store_true', default=False, "--debug",
action="store_true",
default=False,
help="Set log level to DEBUG", help="Set log level to DEBUG",
) )
glob_group.add_argument('--help', glob_group.add_argument(
action='help', help="Show this help message and exit", "--help",
action="help",
help="Show this help message and exit",
) )
return parser.parse_args() return parser.parse_args()
if __name__ == '__main__': if __name__ == "__main__":
opts = _parse_api_args() opts = _parse_api_args()
# Run the server # Run the server
yunohost.api(debug=opts.debug, host=opts.host, port=opts.port) yunohost.api(debug=opts.debug, host=opts.host, port=opts.port)

43
bin/yunomdns
View file

@ -21,8 +21,18 @@ def get_network_local_interfaces() -> Dict[str, Dict[str, List[str]]]:
interfaces = { interfaces = {
adapter.name: { adapter.name: {
"ipv4": [ip.ip for ip in adapter.ips if ip.is_IPv4 and ip_address(ip.ip).is_private], "ipv4": [
"ipv6": [ip.ip[0] for ip in adapter.ips if ip.is_IPv6 and ip_address(ip.ip[0]).is_private and not ip_address(ip.ip[0]).is_link_local], ip.ip
for ip in adapter.ips
if ip.is_IPv4 and ip_address(ip.ip).is_private
],
"ipv6": [
ip.ip[0]
for ip in adapter.ips
if ip.is_IPv6
and ip_address(ip.ip[0]).is_private
and not ip_address(ip.ip[0]).is_link_local
],
} }
for adapter in ifaddr.get_adapters() for adapter in ifaddr.get_adapters()
if adapter.name != "lo" if adapter.name != "lo"
@ -33,7 +43,6 @@ def get_network_local_interfaces() -> Dict[str, Dict[str, List[str]]]:
# Listener class, to detect duplicates on the network # Listener class, to detect duplicates on the network
# Stores the list of servers in its list property # Stores the list of servers in its list property
class Listener: class Listener:
def __init__(self): def __init__(self):
self.list = [] self.list = []
@ -66,14 +75,18 @@ def main() -> bool:
return False return False
if "interfaces" not in config: if "interfaces" not in config:
config["interfaces"] = [interface config["interfaces"] = [
for interface, local_ips in interfaces.items() interface
if local_ips["ipv4"]] for interface, local_ips in interfaces.items()
if local_ips["ipv4"]
]
if "ban_interfaces" in config: if "ban_interfaces" in config:
config["interfaces"] = [interface config["interfaces"] = [
for interface in config["interfaces"] interface
if interface not in config["ban_interfaces"]] for interface in config["interfaces"]
if interface not in config["ban_interfaces"]
]
# Let's discover currently published .local domains accross the network # Let's discover currently published .local domains accross the network
zc = Zeroconf() zc = Zeroconf()
@ -103,14 +116,18 @@ def main() -> bool:
return domain_i return domain_i
config['domains'] = [find_domain_not_already_published(domain) for domain in config['domains']] config["domains"] = [
find_domain_not_already_published(domain) for domain in config["domains"]
]
zcs: Dict[Zeroconf, List[ServiceInfo]] = {} zcs: Dict[Zeroconf, List[ServiceInfo]] = {}
for interface in config["interfaces"]: for interface in config["interfaces"]:
if interface not in interfaces: if interface not in interfaces:
print(f"Interface {interface} listed in config file is not present on system.") print(
f"Interface {interface} listed in config file is not present on system."
)
continue continue
# Only broadcast IPv4 because IPv6 is buggy ... because we ain't using python3-ifaddr >= 0.1.7 # Only broadcast IPv4 because IPv6 is buggy ... because we ain't using python3-ifaddr >= 0.1.7
@ -149,7 +166,9 @@ def main() -> bool:
print("Registering...") print("Registering...")
for zc, infos in zcs.items(): for zc, infos in zcs.items():
for info in infos: for info in infos:
zc.register_service(info, allow_name_change=True, cooperating_responders=True) zc.register_service(
info, allow_name_change=True, cooperating_responders=True
)
try: try:
print("Registered. Press Ctrl+C or stop service to stop.") print("Registered. Press Ctrl+C or stop service to stop.")