ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2026-04-13 23:45:04 +00:00
Code Activity

Cleanup: use super() instead of super(__class__, self) (#11016)

Browse source 
* Address UP008: Use super() instead of super(__class__, self).

* Linting.
This commit is contained in:
Felix Fontein 2025-10-30 20:17:26 +01:00 • committed by GitHub
parent 0c5466de47
commit 74c2c804e5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
204 changed files with 390 additions and 401 deletions
Download patch file Download diff file Expand all files Collapse all files

12
plugins/connection/chroot.py
View file

@ -99,7 +99,7 @@ class Connection(ConnectionBase):
default_user = 'root'
def __init__(self, play_context, new_stdin, *args, **kwargs):
super(Connection, self).__init__(play_context, new_stdin, *args, **kwargs)
super().__init__(play_context, new_stdin, *args, **kwargs)
self.chroot = self._play_context.remote_addr
@ -130,7 +130,7 @@ class Connection(ConnectionBase):
except ValueError as e:
raise AnsibleError(str(e))
super(Connection, self)._connect()
super()._connect()
if not self._connected:
display.vvv("THIS IS A LOCAL CHROOT DIR", host=self.chroot)
self._connected = True
@ -155,7 +155,7 @@ class Connection(ConnectionBase):
def exec_command(self, cmd, in_data=None, sudoable=False):
""" run a command on the chroot """
super(Connection, self).exec_command(cmd, in_data=in_data, sudoable=sudoable)
super().exec_command(cmd, in_data=in_data, sudoable=sudoable)
p = self._buffered_exec_command(cmd)
@ -179,7 +179,7 @@ class Connection(ConnectionBase):
def put_file(self, in_path, out_path):
""" transfer a file from local to chroot """
super(Connection, self).put_file(in_path, out_path)
super().put_file(in_path, out_path)
display.vvv(f"PUT {in_path} TO {out_path}", host=self.chroot)
out_path = shlex_quote(self._prefix_login_path(out_path))
@ -205,7 +205,7 @@ class Connection(ConnectionBase):
def fetch_file(self, in_path, out_path):
""" fetch a file from chroot to local """
super(Connection, self).fetch_file(in_path, out_path)
super().fetch_file(in_path, out_path)
display.vvv(f"FETCH {in_path} TO {out_path}", host=self.chroot)
in_path = shlex_quote(self._prefix_login_path(in_path))
@ -229,5 +229,5 @@ class Connection(ConnectionBase):
def close(self):
""" terminate the connection; nothing to do here """
super(Connection, self).close()
super().close()
self._connected = False

12
plugins/connection/incus.py
View file

@ -90,7 +90,7 @@ class Connection(ConnectionBase):
has_pipelining = True
def __init__(self, play_context, new_stdin, *args, **kwargs):
super(Connection, self).__init__(play_context, new_stdin, *args, **kwargs)
super().__init__(play_context, new_stdin, *args, **kwargs)
self._incus_cmd = get_bin_path("incus")
@ -99,7 +99,7 @@ class Connection(ConnectionBase):
def _connect(self):
"""connect to Incus (nothing to do here) """
super(Connection, self)._connect()
super()._connect()
if not self._connected:
self._display.vvv(f"ESTABLISH Incus CONNECTION FOR USER: {self.get_option('remote_user')}",
@ -137,7 +137,7 @@ class Connection(ConnectionBase):
def exec_command(self, cmd, in_data=None, sudoable=True):
""" execute a command on the Incus host """
super(Connection, self).exec_command(cmd, in_data=in_data, sudoable=sudoable)
super().exec_command(cmd, in_data=in_data, sudoable=sudoable)
self._display.vvv(f"EXEC {cmd}",
host=self._instance())
@ -207,7 +207,7 @@ class Connection(ConnectionBase):
def put_file(self, in_path, out_path):
""" put a file from local to Incus """
super(Connection, self).put_file(in_path, out_path)
super().put_file(in_path, out_path)
self._display.vvv(f"PUT {in_path} TO {out_path}",
host=self._instance())
@ -251,7 +251,7 @@ class Connection(ConnectionBase):
def fetch_file(self, in_path, out_path):
""" fetch a file from Incus to local """
super(Connection, self).fetch_file(in_path, out_path)
super().fetch_file(in_path, out_path)
self._display.vvv(f"FETCH {in_path} TO {out_path}",
host=self._instance())
@ -269,6 +269,6 @@ class Connection(ConnectionBase):
def close(self):
""" close the connection (nothing to do here) """
super(Connection, self).close()
super().close()
self._connected = False

2
plugins/connection/iocage.py
View file

@ -60,7 +60,7 @@ class Connection(Jail):
host=kwargs[Jail.modified_jailname_key]
)
super(Connection, self).__init__(play_context, new_stdin, *args, **kwargs)
super().__init__(play_context, new_stdin, *args, **kwargs)
def get_jail_uuid(self):
p = subprocess.Popen([self.iocage_cmd, 'get', 'host_hostuuid', self.ioc_jail],

12
plugins/connection/jail.py
View file

@ -60,7 +60,7 @@ class Connection(ConnectionBase):
has_tty = False
def __init__(self, play_context, new_stdin, *args, **kwargs):
super(Connection, self).__init__(play_context, new_stdin, *args, **kwargs)
super().__init__(play_context, new_stdin, *args, **kwargs)
self.jail = self._play_context.remote_addr
if self.modified_jailname_key in kwargs:
@ -93,7 +93,7 @@ class Connection(ConnectionBase):
def _connect(self):
""" connect to the jail; nothing to do here """
super(Connection, self)._connect()
super()._connect()
if not self._connected:
display.vvv(f"ESTABLISH JAIL CONNECTION FOR USER: {self._play_context.remote_user}", host=self.jail)
self._connected = True
@ -126,7 +126,7 @@ class Connection(ConnectionBase):
def exec_command(self, cmd, in_data=None, sudoable=False):
""" run a command on the jail """
super(Connection, self).exec_command(cmd, in_data=in_data, sudoable=sudoable)
super().exec_command(cmd, in_data=in_data, sudoable=sudoable)
p = self._buffered_exec_command(cmd)
@ -150,7 +150,7 @@ class Connection(ConnectionBase):
def put_file(self, in_path, out_path):
""" transfer a file from local to jail """
super(Connection, self).put_file(in_path, out_path)
super().put_file(in_path, out_path)
display.vvv(f"PUT {in_path} TO {out_path}", host=self.jail)
out_path = shlex_quote(self._prefix_login_path(out_path))
@ -176,7 +176,7 @@ class Connection(ConnectionBase):
def fetch_file(self, in_path, out_path):
""" fetch a file from jail to local """
super(Connection, self).fetch_file(in_path, out_path)
super().fetch_file(in_path, out_path)
display.vvv(f"FETCH {in_path} TO {out_path}", host=self.jail)
in_path = shlex_quote(self._prefix_login_path(in_path))
@ -200,5 +200,5 @@ class Connection(ConnectionBase):
def close(self):
""" terminate the connection; nothing to do here """
super(Connection, self).close()
super().close()
self._connected = False

12
plugins/connection/lxc.py
View file

@ -58,14 +58,14 @@ class Connection(ConnectionBase):
default_user = 'root'
def __init__(self, play_context, new_stdin, *args, **kwargs):
super(Connection, self).__init__(play_context, new_stdin, *args, **kwargs)
super().__init__(play_context, new_stdin, *args, **kwargs)
self.container_name = None
self.container = None
def _connect(self):
""" connect to the lxc; nothing to do here """
super(Connection, self)._connect()
super()._connect()
if not HAS_LIBLXC:
msg = "lxc python bindings are not installed"
@ -118,7 +118,7 @@ class Connection(ConnectionBase):
def exec_command(self, cmd, in_data=None, sudoable=False):
""" run a command on the chroot """
super(Connection, self).exec_command(cmd, in_data=in_data, sudoable=sudoable)
super().exec_command(cmd, in_data=in_data, sudoable=sudoable)
# python2-lxc needs bytes. python3-lxc needs text.
executable = to_native(self.get_option('executable'), errors='surrogate_or_strict')
@ -171,7 +171,7 @@ class Connection(ConnectionBase):
def put_file(self, in_path, out_path):
''' transfer a file from local to lxc '''
super(Connection, self).put_file(in_path, out_path)
super().put_file(in_path, out_path)
self._display.vvv(f"PUT {in_path} TO {out_path}", host=self.container_name)
in_path = to_bytes(in_path, errors='surrogate_or_strict')
out_path = to_bytes(out_path, errors='surrogate_or_strict')
@ -199,7 +199,7 @@ class Connection(ConnectionBase):
def fetch_file(self, in_path, out_path):
''' fetch a file from lxc to local '''
super(Connection, self).fetch_file(in_path, out_path)
super().fetch_file(in_path, out_path)
self._display.vvv(f"FETCH {in_path} TO {out_path}", host=self.container_name)
in_path = to_bytes(in_path, errors='surrogate_or_strict')
out_path = to_bytes(out_path, errors='surrogate_or_strict')
@ -230,5 +230,5 @@ class Connection(ConnectionBase):
def close(self):
''' terminate the connection; nothing to do here '''
super(Connection, self).close()
super().close()
self._connected = False

12
plugins/connection/lxd.py
View file

@ -89,7 +89,7 @@ class Connection(ConnectionBase):
has_pipelining = True
def __init__(self, play_context, new_stdin, *args, **kwargs):
super(Connection, self).__init__(play_context, new_stdin, *args, **kwargs)
super().__init__(play_context, new_stdin, *args, **kwargs)
try:
self._lxc_cmd = get_bin_path("lxc")
@ -102,7 +102,7 @@ class Connection(ConnectionBase):
def _connect(self):
"""connect to lxd (nothing to do here) """
super(Connection, self)._connect()
super()._connect()
if not self._connected:
self._display.vvv(f"ESTABLISH LXD CONNECTION FOR USER: {self.get_option('remote_user')}", host=self._host())
@ -134,7 +134,7 @@ class Connection(ConnectionBase):
def exec_command(self, cmd, in_data=None, sudoable=True):
""" execute a command on the lxd host """
super(Connection, self).exec_command(cmd, in_data=in_data, sudoable=sudoable)
super().exec_command(cmd, in_data=in_data, sudoable=sudoable)
self._display.vvv(f"EXEC {cmd}", host=self._host())
@ -181,7 +181,7 @@ class Connection(ConnectionBase):
def put_file(self, in_path, out_path):
""" put a file from local to lxd """
super(Connection, self).put_file(in_path, out_path)
super().put_file(in_path, out_path)
self._display.vvv(f"PUT {in_path} TO {out_path}", host=self._host())
@ -225,7 +225,7 @@ class Connection(ConnectionBase):
def fetch_file(self, in_path, out_path):
""" fetch a file from lxd to local """
super(Connection, self).fetch_file(in_path, out_path)
super().fetch_file(in_path, out_path)
self._display.vvv(f"FETCH {in_path} TO {out_path}", host=self._host())
@ -245,6 +245,6 @@ class Connection(ConnectionBase):
def close(self):
""" close the connection (nothing to do here) """
super(Connection, self).close()
super().close()
self._connected = False

12
plugins/connection/qubes.py
View file

@ -57,7 +57,7 @@ class Connection(ConnectionBase):
has_pipelining = True
def __init__(self, play_context, new_stdin, *args, **kwargs):
super(Connection, self).__init__(play_context, new_stdin, *args, **kwargs)
super().__init__(play_context, new_stdin, *args, **kwargs)
self._remote_vmname = self._play_context.remote_addr
self._connected = False
@ -103,13 +103,13 @@ class Connection(ConnectionBase):
def _connect(self):
"""No persistent connection is being maintained."""
super(Connection, self)._connect()
super()._connect()
self._connected = True
@ensure_connect # type: ignore # TODO: for some reason, the type infos for ensure_connect suck...
def exec_command(self, cmd, in_data=None, sudoable=False):
"""Run specified command in a running QubesVM """
super(Connection, self).exec_command(cmd, in_data=in_data, sudoable=sudoable)
super().exec_command(cmd, in_data=in_data, sudoable=sudoable)
display.vvvv(f"CMD IS: {cmd}")
@ -120,7 +120,7 @@ class Connection(ConnectionBase):
def put_file(self, in_path, out_path):
""" Place a local file located in 'in_path' inside VM at 'out_path' """
super(Connection, self).put_file(in_path, out_path)
super().put_file(in_path, out_path)
display.vvv(f"PUT {in_path} TO {out_path}", host=self._remote_vmname)
with open(in_path, "rb") as fobj:
@ -137,7 +137,7 @@ class Connection(ConnectionBase):
def fetch_file(self, in_path, out_path):
"""Obtain file specified via 'in_path' from the container and place it at 'out_path' """
super(Connection, self).fetch_file(in_path, out_path)
super().fetch_file(in_path, out_path)
display.vvv(f"FETCH {in_path} TO {out_path}", host=self._remote_vmname)
# We are running in dom0
@ -150,5 +150,5 @@ class Connection(ConnectionBase):
def close(self):
""" Closing the connection """
super(Connection, self).close()
super().close()
self._connected = False

8
plugins/connection/saltstack.py
View file

@ -39,7 +39,7 @@ class Connection(ConnectionBase):
transport = 'community.general.saltstack'
def __init__(self, play_context, new_stdin, *args, **kwargs):
super(Connection, self).__init__(play_context, new_stdin, *args, **kwargs)
super().__init__(play_context, new_stdin, *args, **kwargs)
self.host = self._play_context.remote_addr
def _connect(self):
@ -52,7 +52,7 @@ class Connection(ConnectionBase):
def exec_command(self, cmd, in_data=None, sudoable=False):
""" run a command on the remote minion """
super(Connection, self).exec_command(cmd, in_data=in_data, sudoable=sudoable)
super().exec_command(cmd, in_data=in_data, sudoable=sudoable)
if in_data:
raise errors.AnsibleError("Internal Error: this module does not support optimized module pipelining")
@ -76,7 +76,7 @@ class Connection(ConnectionBase):
def put_file(self, in_path, out_path):
""" transfer a file from local to remote """
super(Connection, self).put_file(in_path, out_path)
super().put_file(in_path, out_path)
out_path = self._normalize_path(out_path, '/')
self._display.vvv(f"PUT {in_path} TO {out_path}", host=self.host)
@ -88,7 +88,7 @@ class Connection(ConnectionBase):
def fetch_file(self, in_path, out_path):
""" fetch a file from remote to local """
super(Connection, self).fetch_file(in_path, out_path)
super().fetch_file(in_path, out_path)
in_path = self._normalize_path(in_path, '/')
self._display.vvv(f"FETCH {in_path} TO {out_path}", host=self.host)

4
plugins/connection/wsl.py
View file

@ -404,7 +404,7 @@ class Connection(ConnectionBase):
_log_channel: str | None = None
def __init__(self, play_context: PlayContext, new_stdin: io.TextIOWrapper | None = None, *args: t.Any, **kwargs: t.Any):
super(Connection, self).__init__(play_context, new_stdin, *args, **kwargs)
super().__init__(play_context, new_stdin, *args, **kwargs)
def _set_log_channel(self, name: str) -> None:
""" Mimic paramiko.SSHClient.set_log_channel """
@ -587,7 +587,7 @@ class Connection(ConnectionBase):
cmd = self._build_wsl_command(cmd)
super(Connection, self).exec_command(cmd, in_data=in_data, sudoable=sudoable) # type: ignore[safe-super]
super().exec_command(cmd, in_data=in_data, sudoable=sudoable) # type: ignore[safe-super]
bufsize = 4096

12
plugins/connection/zone.py
View file

@ -49,7 +49,7 @@ class Connection(ConnectionBase):
has_tty = False
def __init__(self, play_context, new_stdin, *args, **kwargs):
super(Connection, self).__init__(play_context, new_stdin, *args, **kwargs)
super().__init__(play_context, new_stdin, *args, **kwargs)
self.zone = self._play_context.remote_addr
@ -96,7 +96,7 @@ class Connection(ConnectionBase):
def _connect(self):
""" connect to the zone; nothing to do here """
super(Connection, self)._connect()
super()._connect()
if not self._connected:
display.vvv("THIS IS A LOCAL ZONE DIR", host=self.zone)
self._connected = True
@ -123,7 +123,7 @@ class Connection(ConnectionBase):
def exec_command(self, cmd, in_data=None, sudoable=False):
""" run a command on the zone """
super(Connection, self).exec_command(cmd, in_data=in_data, sudoable=sudoable)
super().exec_command(cmd, in_data=in_data, sudoable=sudoable)
p = self._buffered_exec_command(cmd)
@ -146,7 +146,7 @@ class Connection(ConnectionBase):
def put_file(self, in_path, out_path):
""" transfer a file from local to zone """
super(Connection, self).put_file(in_path, out_path)
super().put_file(in_path, out_path)
display.vvv(f"PUT {in_path} TO {out_path}", host=self.zone)
out_path = shlex_quote(self._prefix_login_path(out_path))
@ -172,7 +172,7 @@ class Connection(ConnectionBase):
def fetch_file(self, in_path, out_path):
""" fetch a file from zone to local """
super(Connection, self).fetch_file(in_path, out_path)
super().fetch_file(in_path, out_path)
display.vvv(f"FETCH {in_path} TO {out_path}", host=self.zone)
in_path = shlex_quote(self._prefix_login_path(in_path))
@ -196,5 +196,5 @@ class Connection(ConnectionBase):
def close(self):
""" terminate the connection; nothing to do here """
super(Connection, self).close()
super().close()
self._connected = False