mirror of
https://github.com/ansible-collections/community.general.git
synced 2026-02-04 07:51:50 +00:00
Clean up other Python files (#11379)
* Address issues found by ruff check. * Make mypy happy; remove some Python 2 compat code. * Also declare port1.
This commit is contained in:
parent
75234597bc
commit
b3dc06a7dd
16 changed files with 52 additions and 47 deletions
|
|
@ -43,20 +43,20 @@ def main():
|
|||
jobs[label] = max(attempt, jobs.get(label, 0))
|
||||
|
||||
for label, attempt in jobs.items():
|
||||
name = "Coverage {attempt} {label}".format(label=label, attempt=attempt)
|
||||
name = f"Coverage {attempt} {label}"
|
||||
source = os.path.join(source_directory, name)
|
||||
source_files = os.listdir(source)
|
||||
|
||||
for source_file in source_files:
|
||||
source_path = os.path.join(source, source_file)
|
||||
destination_path = os.path.join(destination_directory, source_file + "." + label)
|
||||
print('"%s" -> "%s"' % (source_path, destination_path))
|
||||
print(f'"{source_path}" -> "{destination_path}"')
|
||||
shutil.copyfile(source_path, destination_path)
|
||||
count += 1
|
||||
|
||||
print("Coverage file count: %d" % count)
|
||||
print("##vso[task.setVariable variable=coverageFileCount]%d" % count)
|
||||
print("##vso[task.setVariable variable=outputPath]%s" % output_path)
|
||||
print(f"Coverage file count: {count}")
|
||||
print(f"##vso[task.setVariable variable=coverageFileCount]{count}")
|
||||
print(f"##vso[task.setVariable variable=outputPath]{output_path}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
|
|
@ -15,7 +15,6 @@ import pathlib
|
|||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
import typing as t
|
||||
import urllib.request
|
||||
|
||||
|
||||
|
|
@ -23,7 +22,7 @@ import urllib.request
|
|||
class CoverageFile:
|
||||
name: str
|
||||
path: pathlib.Path
|
||||
flags: t.List[str]
|
||||
flags: list[str]
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
|
|
@ -46,7 +45,7 @@ def parse_args() -> Args:
|
|||
return Args(**kwargs)
|
||||
|
||||
|
||||
def process_files(directory: pathlib.Path) -> t.Tuple[CoverageFile, ...]:
|
||||
def process_files(directory: pathlib.Path) -> tuple[CoverageFile, ...]:
|
||||
processed = []
|
||||
for file in directory.joinpath("reports").glob("coverage*.xml"):
|
||||
name = file.stem.replace("coverage=", "")
|
||||
|
|
@ -62,7 +61,7 @@ def process_files(directory: pathlib.Path) -> t.Tuple[CoverageFile, ...]:
|
|||
return tuple(processed)
|
||||
|
||||
|
||||
def upload_files(codecov_bin: pathlib.Path, files: t.Tuple[CoverageFile, ...], dry_run: bool = False) -> None:
|
||||
def upload_files(codecov_bin: pathlib.Path, files: tuple[CoverageFile, ...], dry_run: bool = False) -> None:
|
||||
for file in files:
|
||||
cmd = [
|
||||
str(codecov_bin),
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@ def main():
|
|||
|
||||
for line in sys.stdin:
|
||||
seconds = time.time() - start
|
||||
sys.stdout.write("%02d:%02d %s" % (seconds // 60, seconds % 60, line))
|
||||
sys.stdout.write(f"{seconds // 60:02}:{seconds % 60:02} {line}")
|
||||
sys.stdout.flush()
|
||||
|
||||
|
||||
|
|
|
|||
20
.mypy.ini
20
.mypy.ini
|
|
@ -14,6 +14,8 @@ warn_redundant_casts = True
|
|||
# warn_return_any = True
|
||||
warn_unreachable = True
|
||||
|
||||
exclude = tests/integration/targets/django_.*/files/.*
|
||||
|
||||
[mypy-ansible.*]
|
||||
# ansible-core has partial typing information
|
||||
follow_untyped_imports = True
|
||||
|
|
@ -24,6 +26,10 @@ follow_untyped_imports = True
|
|||
# 3. That have no types and type stubs.
|
||||
[mypy-aerospike.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-antsibull_nox.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-asyncore.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-boto3.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-bs4.*]
|
||||
|
|
@ -38,6 +44,8 @@ ignore_missing_imports = True
|
|||
ignore_missing_imports = True
|
||||
[mypy-crypt.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-daemon.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-datadog.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-dbus.*]
|
||||
|
|
@ -122,6 +130,10 @@ ignore_missing_imports = True
|
|||
ignore_missing_imports = True
|
||||
[mypy-nomad.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-nopackagewiththisname.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-nox.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-oci.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-oneandone.*]
|
||||
|
|
@ -174,6 +186,8 @@ ignore_missing_imports = True
|
|||
ignore_missing_imports = True
|
||||
[mypy-rpm.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-ruamel.yaml.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-salt.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-selinux.*]
|
||||
|
|
@ -186,6 +200,10 @@ ignore_missing_imports = True
|
|||
ignore_missing_imports = True
|
||||
[mypy-sha.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-smtpd.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-smtpd_tls.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-SoftLayer.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-spotinst_sdk.*]
|
||||
|
|
@ -202,6 +220,8 @@ ignore_missing_imports = True
|
|||
ignore_missing_imports = True
|
||||
[mypy-vexatapi.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-voluptuous.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-websocket.*]
|
||||
ignore_missing_imports = True
|
||||
[mypy-XenAPI.*]
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@
|
|||
import sys
|
||||
from io import StringIO
|
||||
|
||||
from ruamel.yaml import YAML
|
||||
from ruamel.yaml import YAML # type: ignore[import-not-found]
|
||||
|
||||
|
||||
def main() -> None:
|
||||
|
|
|
|||
|
|
@ -9,14 +9,14 @@
|
|||
import os
|
||||
import sys
|
||||
|
||||
import nox
|
||||
import nox # type: ignore[import-not-found]
|
||||
|
||||
# Whether the noxfile is running in CI:
|
||||
IN_CI = os.environ.get("CI") == "true"
|
||||
|
||||
|
||||
try:
|
||||
import antsibull_nox
|
||||
import antsibull_nox # type: ignore[import-not-found]
|
||||
except ImportError:
|
||||
print("You need to install antsibull-nox in the same Python environment as nox.")
|
||||
sys.exit(1)
|
||||
|
|
|
|||
|
|
@ -12,7 +12,6 @@ def callback_results_extractor(outputs_results):
|
|||
expected_output = result["test"]["expected_output"]
|
||||
stdout_lines = result["stdout_lines"]
|
||||
for i in range(max(len(expected_output), len(stdout_lines))):
|
||||
line = "line_%s" % (i + 1)
|
||||
test_line = stdout_lines[i] if i < len(stdout_lines) else None
|
||||
expected_lines = expected_output[i] if i < len(expected_output) else None
|
||||
if not isinstance(expected_lines, str) and expected_lines is not None:
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ from ansible.playbook.conditional import Conditional
|
|||
from ansible.plugins.action import ActionBase
|
||||
|
||||
try:
|
||||
from ansible.utils.datatag import trust_value as _trust_value
|
||||
from ansible.utils.datatag import trust_value as _trust_value # type: ignore[import-not-found]
|
||||
except ImportError:
|
||||
|
||||
def _trust_value(input):
|
||||
|
|
|
|||
|
|
@ -51,11 +51,10 @@ def main():
|
|||
with runner.context(p["arg_order"], check_mode_skip=p["check_mode_skip"]) as ctx:
|
||||
result = ctx.run(**p["arg_values"])
|
||||
info = ctx.run_info
|
||||
check = "check"
|
||||
rc, out, err = result if result is not None else (None, None, None)
|
||||
|
||||
module.exit_json(rc=rc, out=out, err=err, info=info)
|
||||
except Exception as exc:
|
||||
except Exception:
|
||||
module.fail_json(rc=1, module_stderr=traceback.format_exc(), msg="Module crashed with exception")
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -9,13 +9,8 @@ import os
|
|||
import posixpath
|
||||
import sys
|
||||
|
||||
try:
|
||||
from http.server import SimpleHTTPRequestHandler, HTTPServer
|
||||
from urllib.parse import unquote
|
||||
except ImportError:
|
||||
from SimpleHTTPServer import SimpleHTTPRequestHandler
|
||||
from BaseHTTPServer import HTTPServer
|
||||
from urllib import unquote
|
||||
from http.server import SimpleHTTPRequestHandler, HTTPServer
|
||||
from urllib.parse import unquote
|
||||
|
||||
|
||||
# Argument parsing
|
||||
|
|
@ -23,8 +18,8 @@ if len(sys.argv) != 4:
|
|||
print(f"Syntax: {sys.argv[0]} <bind> <port> <path>")
|
||||
sys.exit(-1)
|
||||
|
||||
HOST, PORT, PATH = sys.argv[1:4]
|
||||
PORT = int(PORT)
|
||||
HOST, PORT_str, PATH = sys.argv[1:4]
|
||||
PORT = int(PORT_str)
|
||||
|
||||
|
||||
# The HTTP request handler
|
||||
|
|
@ -40,7 +35,7 @@ class Handler(SimpleHTTPRequestHandler):
|
|||
trailing_slash = path.rstrip().endswith("/")
|
||||
try:
|
||||
path = unquote(path, errors="surrogatepass")
|
||||
except (UnicodeDecodeError, TypeError) as exc:
|
||||
except (UnicodeDecodeError, TypeError):
|
||||
path = unquote(path)
|
||||
path = posixpath.normpath(path)
|
||||
words = path.split("/")
|
||||
|
|
|
|||
|
|
@ -10,15 +10,11 @@ import sys
|
|||
root_dir = sys.argv[1]
|
||||
port = int(sys.argv[2])
|
||||
|
||||
try:
|
||||
from BaseHTTPServer import HTTPServer
|
||||
from SimpleHTTPServer import SimpleHTTPRequestHandler
|
||||
except ModuleNotFoundError:
|
||||
from http.server import HTTPServer, SimpleHTTPRequestHandler
|
||||
from http.server import HTTPServer, SimpleHTTPRequestHandler
|
||||
|
||||
httpd = HTTPServer(("localhost", port), SimpleHTTPRequestHandler)
|
||||
try:
|
||||
httpd.socket = ssl.wrap_socket(
|
||||
httpd.socket = ssl.wrap_socket( # type: ignore[attr-defined]
|
||||
httpd.socket,
|
||||
server_side=True,
|
||||
certfile=os.path.join(root_dir, "cert.pem"),
|
||||
|
|
|
|||
|
|
@ -27,6 +27,8 @@ port = "25:465"
|
|||
if len(sys.argv) > 1:
|
||||
port = sys.argv[1]
|
||||
ports = port.split(":")
|
||||
port1: int
|
||||
port2: int | None
|
||||
if len(ports) > 1:
|
||||
port1, port2 = int(ports[0]), int(ports[1])
|
||||
else:
|
||||
|
|
@ -58,6 +60,6 @@ else:
|
|||
print("Start SMTP server on port", port1)
|
||||
smtp_server1 = smtpd.DebuggingServer(("127.0.0.1", port1), None) # pylint: disable=used-before-assignment
|
||||
if port2:
|
||||
print("WARNING: TLS is NOT supported on this system, not listening on port %s." % port2)
|
||||
print(f"WARNING: TLS is NOT supported on this system, not listening on port {port2}.")
|
||||
|
||||
asyncore.loop()
|
||||
|
|
|
|||
|
|
@ -32,7 +32,7 @@ from ansible_collections.community.general.plugins.module_utils import deps
|
|||
from ansible_collections.community.general.plugins.module_utils.module_helper import ModuleHelper
|
||||
|
||||
with deps.declare("nopackagewiththisname"):
|
||||
import nopackagewiththisname # noqa: F401, pylint: disable=unused-import
|
||||
import nopackagewiththisname # noqa: F401, pylint: disable=unused-import # type: ignore[import-not-found]
|
||||
|
||||
|
||||
class MSimple(ModuleHelper):
|
||||
|
|
|
|||
|
|
@ -6,15 +6,10 @@ from __future__ import annotations
|
|||
|
||||
import daemon
|
||||
|
||||
try:
|
||||
from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
|
||||
def write_to_output(stream, content):
|
||||
stream.write(content)
|
||||
except ImportError:
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
|
||||
def write_to_output(stream, content):
|
||||
def write_to_output(stream, content):
|
||||
stream.write(bytes(content, "utf-8"))
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -22,4 +22,4 @@ else:
|
|||
url = "http://127.0.0.1:9001/RPC2"
|
||||
|
||||
server = ServerProxy(url, verbose=True)
|
||||
server.supervisor.sendProcessStdin(proc, "import sys; print(%s); sys.stdout.flush();\n" % value)
|
||||
server.supervisor.sendProcessStdin(proc, f"import sys; print({value}); sys.stdout.flush();\n")
|
||||
|
|
|
|||
|
|
@ -56,7 +56,7 @@ IGNORE_NO_MAINTAINERS = [
|
|||
|
||||
|
||||
class BotmetaCheck:
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
self.errors: list[str] = []
|
||||
self.botmeta_filename = ".github/BOTMETA.yml"
|
||||
self.list_entries = frozenset(("supershipit", "maintainers", "labels", "keywords", "notify", "ignore"))
|
||||
|
|
@ -121,8 +121,8 @@ class BotmetaCheck:
|
|||
)
|
||||
):
|
||||
maintainers = self.read_authors(filename)
|
||||
for maintainer in maintainers:
|
||||
maintainer = self.extract_author_name(maintainer)
|
||||
for maintainer_str in maintainers:
|
||||
maintainer = self.extract_author_name(maintainer_str)
|
||||
if maintainer is not None and maintainer not in all_maintainers:
|
||||
others = ", ".join(all_maintainers)
|
||||
msg = f"Author {maintainer} not mentioned as active or inactive maintainer for {filename} (mentioned are: {others})"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue