From c06fdaa102ad8af9440038f5bc1fcaed4c79c701 Mon Sep 17 00:00:00 2001 From: Philip Guyton Date: Wed, 20 Mar 2024 16:37:51 +0000 Subject: [PATCH] [t] Add Group with custom GID fails with type error #2807 Add type hints to run_command() to ensure callers are passing the expected argument types. --- src/rockstor/system/directory_services.py | 6 +++--- src/rockstor/system/osi.py | 23 ++++++++++++----------- 2 files changed, 15 insertions(+), 14 deletions(-) diff --git a/src/rockstor/system/directory_services.py b/src/rockstor/system/directory_services.py index 9f6d17911..79c34602d 100644 --- a/src/rockstor/system/directory_services.py +++ b/src/rockstor/system/directory_services.py @@ -267,7 +267,7 @@ def join_domain(config, method="sssd"): cmd[-3:-3] = cmd_options if method == "winbind": cmd = [NET, "ads", "join", "-U", admin] - return run_command(cmd, input=("{}\n".format(config.get("password"))), log=True) + return run_command(cmd, pinput=("{}\n".format(config.get("password"))), log=True) def leave_domain(config, method="sssd"): @@ -282,10 +282,10 @@ def leave_domain(config, method="sssd"): if method == "winbind": cmd = [NET, "ads", "leave", "-U", config.get("username")] try: - return run_command(cmd, input=pstr) + return run_command(cmd, pinput=pstr) except: status_cmd = [NET, "ads", "status", "-U", config.get("username")] - o, e, rc = run_command(status_cmd, input=pstr, throw=False) + o, e, rc = run_command(status_cmd, pinput=pstr, throw=False) if rc != 0: return logger.debug( "Status shows not joined. out: %s err: %s rc: %d" % (o, e, rc) diff --git a/src/rockstor/system/osi.py b/src/rockstor/system/osi.py index b0f3068e1..840acb9b5 100644 --- a/src/rockstor/system/osi.py +++ b/src/rockstor/system/osi.py @@ -31,6 +31,7 @@ from struct import pack from tempfile import mkstemp from distutils.util import strtobool +from typing import AnyStr, IO from django.conf import settings @@ -214,16 +215,16 @@ def replace_pattern_inline(source_file, target_file, pattern, replacement): def run_command( - cmd, - shell=False, ## Default - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - stdin=subprocess.PIPE, - throw=True, - log=False, - input=None, - raw=False, -): + cmd: list[str], + shell: bool = False, + stdout: None | int | IO = subprocess.PIPE, + stderr: None | int | IO = subprocess.PIPE, + stdin: None | int | IO = subprocess.PIPE, + throw: bool = True, + log: bool = False, + pinput: AnyStr | None = None, + raw: bool = False, +) -> (list[str] | str, list[str], int): try: # We force run_command to always use en_US # to avoid issues on date and number formats @@ -243,7 +244,7 @@ def run_command( env=fake_env, universal_newlines=True, # 3.7 adds text parameter universal_newlines alias ) - out, err = p.communicate(input=input) + out, err = p.communicate(input=pinput) # raw=True allows parsing of a JSON output directly, for instance if not raw: out = out.split("\n")