mirror of
https://github.com/ansible-collections/community.general.git
synced 2026-04-20 18:59:08 +00:00
[PR #11879/77509be2 backport][stable-12] Replace .format() calls with f-strings across multiple plugins (#11881)
Replace .format() calls with f-strings across multiple plugins (#11879)
* Replace .format() calls with f-strings across multiple plugins
* Add changelog fragment for PR 11879
---------
(cherry picked from commit 77509be2aa)
Co-authored-by: Alexei Znamensky <103110+russoz@users.noreply.github.com>
Co-authored-by: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
patchback[bot]
GitHub
parent
6458abb9c1
commit
748882dfa8
15 changed files with 88 additions and 114 deletions
12
changelogs/fragments/11879-convert-format-to-fstrings.yml
Normal file
12
changelogs/fragments/11879-convert-format-to-fstrings.yml
Normal file
|
|
@ -0,0 +1,12 @@
|
|||
minor_changes:
|
||||
- cobbler_sync - minor code cleanup (https://github.com/ansible-collections/community.general/pull/11879).
|
||||
- cobbler_system - minor code cleanup (https://github.com/ansible-collections/community.general/pull/11879).
|
||||
- manageiq_alert_profiles - minor code cleanup (https://github.com/ansible-collections/community.general/pull/11879).
|
||||
- manageiq_alerts - minor code cleanup (https://github.com/ansible-collections/community.general/pull/11879).
|
||||
- oneview_san_manager - minor code cleanup (https://github.com/ansible-collections/community.general/pull/11879).
|
||||
- packet_device - minor code cleanup (https://github.com/ansible-collections/community.general/pull/11879).
|
||||
- packet_ip_subnet - minor code cleanup (https://github.com/ansible-collections/community.general/pull/11879).
|
||||
- pubnub_blocks - minor code cleanup (https://github.com/ansible-collections/community.general/pull/11879).
|
||||
- terraform - minor code cleanup (https://github.com/ansible-collections/community.general/pull/11879).
|
||||
- manageiq module utils - minor code cleanup (https://github.com/ansible-collections/community.general/pull/11879).
|
||||
- oneview module utils - minor code cleanup (https://github.com/ansible-collections/community.general/pull/11879).
|
||||
|
|
@ -53,7 +53,6 @@ def check_client(module: AnsibleModule) -> None:
|
|||
|
||||
def validate_connection_params(module: AnsibleModule) -> dict[str, t.Any]:
|
||||
params: dict[str, t.Any] = module.params["manageiq_connection"]
|
||||
error_str = "missing required argument: manageiq_connection[{}]"
|
||||
url: str | None = params["url"]
|
||||
token: str | None = params["token"]
|
||||
username: str | None = params["username"]
|
||||
|
|
@ -63,7 +62,7 @@ def validate_connection_params(module: AnsibleModule) -> dict[str, t.Any]:
|
|||
return params
|
||||
for arg in ["url", "username", "password"]:
|
||||
if params[arg] in (None, ""):
|
||||
module.fail_json(msg=error_str.format(arg))
|
||||
module.fail_json(msg=f"missing required argument: manageiq_connection[{arg}]")
|
||||
raise AssertionError("should be unreachable")
|
||||
|
||||
|
||||
|
|
@ -218,9 +217,9 @@ class ManageIQPolicies:
|
|||
|
||||
def query_resource_profiles(self):
|
||||
"""Returns a set of the profile objects objects assigned to the resource"""
|
||||
url = "{resource_url}/policy_profiles?expand=resources"
|
||||
url = f"{self.resource_url}/policy_profiles?expand=resources"
|
||||
try:
|
||||
response = self.client.get(url.format(resource_url=self.resource_url))
|
||||
response = self.client.get(url)
|
||||
except Exception as e:
|
||||
msg = f"Failed to query {self.resource_type} policies: {e}"
|
||||
self.module.fail_json(msg=msg)
|
||||
|
|
@ -235,9 +234,9 @@ class ManageIQPolicies:
|
|||
|
||||
def query_profile_policies(self, profile_id):
|
||||
"""Returns a set of the policy objects assigned to the resource"""
|
||||
url = "{api_url}/policy_profiles/{profile_id}?expand=policies"
|
||||
url = f"{self.api_url}/policy_profiles/{profile_id}?expand=policies"
|
||||
try:
|
||||
response = self.client.get(url.format(api_url=self.api_url, profile_id=profile_id))
|
||||
response = self.client.get(url)
|
||||
except Exception as e:
|
||||
msg = f"Failed to query {self.resource_type} policies: {e}"
|
||||
self.module.fail_json(msg=msg)
|
||||
|
|
@ -370,9 +369,9 @@ class ManageIQTags:
|
|||
|
||||
def query_resource_tags(self):
|
||||
"""Returns a set of the tag objects assigned to the resource"""
|
||||
url = "{resource_url}/tags?expand=resources&attributes=categorization"
|
||||
url = f"{self.resource_url}/tags?expand=resources&attributes=categorization"
|
||||
try:
|
||||
response = self.client.get(url.format(resource_url=self.resource_url))
|
||||
response = self.client.get(url)
|
||||
except Exception as e:
|
||||
msg = f"Failed to query {self.resource_type} tags: {e}"
|
||||
self.module.fail_json(msg=msg)
|
||||
|
|
|
|||
|
|
@ -186,7 +186,6 @@ class OneViewModuleBase(metaclass=abc.ABCMeta):
|
|||
MSG_DELETED = "Resource deleted successfully."
|
||||
MSG_ALREADY_PRESENT = "Resource is already present."
|
||||
MSG_ALREADY_ABSENT = "Resource is already absent."
|
||||
MSG_DIFF_AT_KEY = "Difference found at key '{0}'. "
|
||||
|
||||
ONEVIEW_COMMON_ARGS = dict(
|
||||
config=dict(type="path"),
|
||||
|
|
@ -407,7 +406,7 @@ class OneViewModuleBase(metaclass=abc.ABCMeta):
|
|||
if key not in resource2:
|
||||
if resource1[key] is not None:
|
||||
# Inexistent key is equivalent to exist with value None
|
||||
self.module.log(self.MSG_DIFF_AT_KEY.format(key) + debug_resources)
|
||||
self.module.log(f"Difference found at key '{key}'. {debug_resources}")
|
||||
return False
|
||||
# If both values are null, empty or False it will be considered equal.
|
||||
elif not resource1[key] and not resource2[key]:
|
||||
|
|
@ -415,15 +414,15 @@ class OneViewModuleBase(metaclass=abc.ABCMeta):
|
|||
elif isinstance(resource1[key], Mapping):
|
||||
# recursive call
|
||||
if not self.compare(resource1[key], resource2[key]):
|
||||
self.module.log(self.MSG_DIFF_AT_KEY.format(key) + debug_resources)
|
||||
self.module.log(f"Difference found at key '{key}'. {debug_resources}")
|
||||
return False
|
||||
elif isinstance(resource1[key], list):
|
||||
# change comparison function to compare_list
|
||||
if not self.compare_list(resource1[key], resource2[key]):
|
||||
self.module.log(self.MSG_DIFF_AT_KEY.format(key) + debug_resources)
|
||||
self.module.log(f"Difference found at key '{key}'. {debug_resources}")
|
||||
return False
|
||||
elif _standardize_value(resource1[key]) != _standardize_value(resource2[key]):
|
||||
self.module.log(self.MSG_DIFF_AT_KEY.format(key) + debug_resources)
|
||||
self.module.log(f"Difference found at key '{key}'. {debug_resources}")
|
||||
return False
|
||||
|
||||
# Checks all keys in the second dict, looking for missing elements
|
||||
|
|
@ -431,7 +430,7 @@ class OneViewModuleBase(metaclass=abc.ABCMeta):
|
|||
if key not in resource1:
|
||||
if resource2[key] is not None:
|
||||
# Inexistent key is equivalent to exist with value None
|
||||
self.module.log(self.MSG_DIFF_AT_KEY.format(key) + debug_resources)
|
||||
self.module.log(f"Difference found at key '{key}'. {debug_resources}")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
|
|
|||
|
|
@ -99,9 +99,9 @@ def main():
|
|||
use_ssl = module.params["use_ssl"]
|
||||
validate_certs = module.params["validate_certs"]
|
||||
|
||||
module.params["proto"] = "https" if use_ssl else "http"
|
||||
proto = "https" if use_ssl else "http"
|
||||
if not port:
|
||||
module.params["port"] = "443" if use_ssl else "80"
|
||||
port = "443" if use_ssl else "80"
|
||||
|
||||
result = dict(
|
||||
changed=True,
|
||||
|
|
@ -111,17 +111,13 @@ def main():
|
|||
|
||||
ssl_context = None if validate_certs or not use_ssl else ssl._create_unverified_context()
|
||||
|
||||
url = "{proto}://{host}:{port}/cobbler_api".format(**module.params)
|
||||
url = f"{proto}://{module.params['host']}:{port}/cobbler_api"
|
||||
conn = xmlrpc_client.ServerProxy(url, context=ssl_context)
|
||||
|
||||
try:
|
||||
token = conn.login(username, password)
|
||||
except xmlrpc_client.Fault as e:
|
||||
module.fail_json(
|
||||
msg="Failed to log in to Cobbler '{url}' as '{username}'. {error}".format(
|
||||
url=url, error=f"{e}", **module.params
|
||||
)
|
||||
)
|
||||
module.fail_json(msg=f"Failed to log in to Cobbler '{url}' as '{username}'. {e}")
|
||||
except Exception as e:
|
||||
module.fail_json(msg=f"Connection to '{url}' failed. {e}")
|
||||
|
||||
|
|
|
|||
|
|
@ -220,9 +220,9 @@ def main():
|
|||
name = module.params["name"]
|
||||
state = module.params["state"]
|
||||
|
||||
module.params["proto"] = "https" if use_ssl else "http"
|
||||
proto = "https" if use_ssl else "http"
|
||||
if not port:
|
||||
module.params["port"] = "443" if use_ssl else "80"
|
||||
port = "443" if use_ssl else "80"
|
||||
|
||||
result = dict(
|
||||
changed=False,
|
||||
|
|
@ -232,17 +232,13 @@ def main():
|
|||
|
||||
ssl_context = None if validate_certs or not use_ssl else ssl._create_unverified_context()
|
||||
|
||||
url = "{proto}://{host}:{port}/cobbler_api".format(**module.params)
|
||||
url = f"{proto}://{module.params['host']}:{port}/cobbler_api"
|
||||
conn = xmlrpc_client.ServerProxy(url, context=ssl_context)
|
||||
|
||||
try:
|
||||
token = conn.login(username, password)
|
||||
except xmlrpc_client.Fault as e:
|
||||
module.fail_json(
|
||||
msg="Failed to log in to Cobbler '{url}' as '{username}'. {error}".format(
|
||||
url=url, error=f"{e}", **module.params
|
||||
)
|
||||
)
|
||||
module.fail_json(msg=f"Failed to log in to Cobbler '{url}' as '{username}'. {e}")
|
||||
except Exception as e:
|
||||
module.fail_json(msg=f"Connection to '{url}' failed. {e}")
|
||||
|
||||
|
|
|
|||
|
|
@ -135,8 +135,7 @@ class ManageIQAlertProfiles:
|
|||
# now that it has been created, we can assign the alerts
|
||||
self.assign_or_unassign(result["results"][0], alerts, "assign")
|
||||
|
||||
msg = "Profile {name} created successfully"
|
||||
msg = msg.format(name=profile["name"])
|
||||
msg = f"Profile {profile['name']} created successfully"
|
||||
return dict(changed=True, msg=msg)
|
||||
|
||||
def delete_profile(self, profile):
|
||||
|
|
@ -161,12 +160,14 @@ class ManageIQAlertProfiles:
|
|||
try:
|
||||
result = self.client.post(subcollection_url, resources=alerts, action=action)
|
||||
if len(result["results"]) != len(alerts):
|
||||
msg = "Failed to {action} alerts to profile '{name}',expected {expected} alerts to be {action}ed,but only {changed} were {action}ed"
|
||||
msg = msg.format(action=action, name=profile["name"], expected=len(alerts), changed=result["results"])
|
||||
msg = (
|
||||
f"Failed to {action} alerts to profile '{profile['name']}', "
|
||||
f"expected {len(alerts)} alerts to be {action}ed, "
|
||||
f"but only {result['results']} were {action}ed"
|
||||
)
|
||||
self.module.fail_json(msg=msg)
|
||||
except Exception as e:
|
||||
msg = "Failed to {action} alerts to profile '{name}': {error}"
|
||||
msg = msg.format(action=action, name=profile["name"], error=e)
|
||||
msg = f"Failed to {action} alerts to profile '{profile['name']}': {e}"
|
||||
self.module.fail_json(msg=msg)
|
||||
|
||||
return result["results"]
|
||||
|
|
@ -220,8 +221,7 @@ class ManageIQAlertProfiles:
|
|||
try:
|
||||
self.client.post(old_profile["href"], resource=profile_dict, action="edit")
|
||||
except Exception as e:
|
||||
msg = "Updating profile '{name}' failed: {error}"
|
||||
msg = msg.format(name=old_profile["name"], error=e)
|
||||
msg = f"Updating profile '{old_profile['name']}' failed: {e}"
|
||||
self.module.fail_json(msg=msg)
|
||||
|
||||
if changed:
|
||||
|
|
@ -280,8 +280,7 @@ def main():
|
|||
res_args = manageiq_alert_profiles.delete_profile(existing_profile)
|
||||
else:
|
||||
# This alert profile does not exist in ManageIQ, and that's okay
|
||||
msg = "Alert profile '{name}' does not exist in ManageIQ"
|
||||
msg = msg.format(name=name)
|
||||
msg = f"Alert profile '{name}' does not exist in ManageIQ"
|
||||
res_args = dict(changed=False, msg=msg)
|
||||
|
||||
module.exit_json(**res_args)
|
||||
|
|
|
|||
|
|
@ -213,30 +213,25 @@ class ManageIQAlerts:
|
|||
try:
|
||||
result = self.client.post(self.alerts_url, action="create", resource=alert)
|
||||
|
||||
msg = "Alert {description} created successfully: {details}"
|
||||
msg = msg.format(description=alert["description"], details=result)
|
||||
msg = f"Alert {alert['description']} created successfully: {result}"
|
||||
return dict(changed=True, msg=msg)
|
||||
except Exception as e:
|
||||
msg = "Creating alert {description} failed: {error}"
|
||||
description = alert["description"]
|
||||
if "Resource expression needs be specified" in str(e):
|
||||
# Running on an older version of ManageIQ and trying to create a hash expression
|
||||
msg = msg.format(
|
||||
description=alert["description"], error="Your version of ManageIQ does not support hash_expression"
|
||||
)
|
||||
msg = f"Creating alert {description} failed: Your version of ManageIQ does not support hash_expression"
|
||||
else:
|
||||
msg = msg.format(description=alert["description"], error=e)
|
||||
msg = f"Creating alert {description} failed: {e}"
|
||||
self.module.fail_json(msg=msg)
|
||||
|
||||
def delete_alert(self, alert):
|
||||
"""Delete an alert"""
|
||||
try:
|
||||
result = self.client.post(f"{self.alerts_url}/{alert['id']}", action="delete")
|
||||
msg = "Alert {description} deleted: {details}"
|
||||
msg = msg.format(description=alert["description"], details=result)
|
||||
msg = f"Alert {alert['description']} deleted: {result}"
|
||||
return dict(changed=True, msg=msg)
|
||||
except Exception as e:
|
||||
msg = "Deleting alert {description} failed: {error}"
|
||||
msg = msg.format(description=alert["description"], error=e)
|
||||
msg = f"Deleting alert {alert['description']} failed: {e}"
|
||||
self.module.fail_json(msg=msg)
|
||||
|
||||
def update_alert(self, existing_alert, new_alert):
|
||||
|
|
@ -254,27 +249,22 @@ class ManageIQAlerts:
|
|||
# the result to the expected result.
|
||||
if new_alert_obj == ManageIQAlert(result):
|
||||
# success!
|
||||
msg = "Alert {description} updated successfully: {details}"
|
||||
msg = msg.format(description=existing_alert["description"], details=result)
|
||||
msg = f"Alert {existing_alert['description']} updated successfully: {result}"
|
||||
|
||||
return dict(changed=True, msg=msg)
|
||||
else:
|
||||
# unexpected result
|
||||
msg = "Updating alert {description} failed, unexpected result {details}"
|
||||
msg = msg.format(description=existing_alert["description"], details=result)
|
||||
msg = f"Updating alert {existing_alert['description']} failed, unexpected result {result}"
|
||||
|
||||
self.module.fail_json(msg=msg)
|
||||
|
||||
except Exception as e:
|
||||
msg = "Updating alert {description} failed: {error}"
|
||||
description = existing_alert["description"]
|
||||
if "Resource expression needs be specified" in str(e):
|
||||
# Running on an older version of ManageIQ and trying to update a hash expression
|
||||
msg = msg.format(
|
||||
description=existing_alert["description"],
|
||||
error="Your version of ManageIQ does not support hash_expression",
|
||||
)
|
||||
msg = f"Updating alert {description} failed: Your version of ManageIQ does not support hash_expression"
|
||||
else:
|
||||
msg = msg.format(description=existing_alert["description"], error=e)
|
||||
msg = f"Updating alert {description} failed: {e}"
|
||||
self.module.fail_json(msg=msg)
|
||||
|
||||
|
||||
|
|
@ -337,8 +327,7 @@ def main():
|
|||
res_args = manageiq_alerts.delete_alert(existing_alert)
|
||||
else:
|
||||
# it doesn't exist, and that's okay
|
||||
msg = "Alert '{description}' does not exist in ManageIQ"
|
||||
msg = msg.format(description=description)
|
||||
msg = f"Alert '{description}' does not exist in ManageIQ"
|
||||
res_args = dict(changed=False, msg=msg)
|
||||
|
||||
module.exit_json(**res_args)
|
||||
|
|
|
|||
|
|
@ -140,7 +140,6 @@ class SanManagerModule(OneViewModuleBase):
|
|||
MSG_DELETED = "SAN Manager deleted successfully."
|
||||
MSG_ALREADY_PRESENT = "SAN Manager is already present."
|
||||
MSG_ALREADY_ABSENT = "SAN Manager is already absent."
|
||||
MSG_SAN_MANAGER_PROVIDER_DISPLAY_NAME_NOT_FOUND = "The provider '{0}' was not found."
|
||||
|
||||
argument_spec = dict(
|
||||
state=dict(type="str", default="present", choices=["absent", "present", "connection_information_set"]),
|
||||
|
|
@ -161,7 +160,7 @@ class SanManagerModule(OneViewModuleBase):
|
|||
else:
|
||||
msg = 'A "name" or "connectionInfo" must be provided inside the "data" field for this operation. '
|
||||
msg += 'If a "connectionInfo" is provided, the "Host" name is considered as the "name" for the resource.'
|
||||
raise OneViewModuleValueError(msg.format())
|
||||
raise OneViewModuleValueError(msg)
|
||||
|
||||
resource = self.resource_client.get_by_name(resource_name)
|
||||
|
||||
|
|
@ -211,7 +210,7 @@ class SanManagerModule(OneViewModuleBase):
|
|||
provider_uri = self.resource_client.get_provider_uri(display_name)
|
||||
|
||||
if not provider_uri:
|
||||
raise OneViewModuleValueError(self.MSG_SAN_MANAGER_PROVIDER_DISPLAY_NAME_NOT_FOUND.format(display_name))
|
||||
raise OneViewModuleValueError(f"The provider '{display_name}' was not found.")
|
||||
|
||||
return provider_uri
|
||||
|
||||
|
|
|
|||
|
|
@ -286,7 +286,7 @@ except ImportError:
|
|||
HAS_PACKET_SDK = False
|
||||
|
||||
|
||||
NAME_RE = r"({0}|{0}{1}*{0})".format(r"[a-zA-Z0-9]", r"[a-zA-Z0-9\-]")
|
||||
NAME_RE = r"([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]*[a-zA-Z0-9])"
|
||||
HOSTNAME_RE = rf"({NAME_RE}\.)*{NAME_RE}$"
|
||||
MAX_DEVICES = 100
|
||||
|
||||
|
|
|
|||
|
|
@ -158,7 +158,7 @@ except ImportError:
|
|||
HAS_PACKET_SDK = False
|
||||
|
||||
|
||||
NAME_RE = r"({0}|{0}{1}*{0})".format(r"[a-zA-Z0-9]", r"[a-zA-Z0-9\-]")
|
||||
NAME_RE = r"([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]*[a-zA-Z0-9])"
|
||||
HOSTNAME_RE = rf"({NAME_RE}\.)*{NAME_RE}$"
|
||||
PROJECT_MAX_DEVICES = 100
|
||||
|
||||
|
|
|
|||
|
|
@ -275,12 +275,14 @@ def pubnub_account(module, user):
|
|||
account_name = params.get("account")
|
||||
account = user.account(name=params.get("account"))
|
||||
if account is None:
|
||||
err_frmt = (
|
||||
"It looks like there is no '{0}' account for "
|
||||
"authorized user. Please make sure what correct "
|
||||
"name has been passed during module configuration."
|
||||
module.fail_json(
|
||||
msg="Missing account.",
|
||||
description=(
|
||||
f"It looks like there is no '{account_name}' account for authorized user. "
|
||||
"Please make sure what correct name has been passed during module configuration."
|
||||
),
|
||||
changed=False,
|
||||
)
|
||||
module.fail_json(msg="Missing account.", description=err_frmt.format(account_name), changed=False)
|
||||
else:
|
||||
account = user.accounts()[0]
|
||||
|
||||
|
|
@ -312,14 +314,15 @@ def pubnub_application(module, account):
|
|||
module.fail_json(msg=exc_msg, description=exc_descr, changed=account.changed, module_cache=dict(account))
|
||||
|
||||
if application is None:
|
||||
err_fmt = (
|
||||
"There is no '{0}' application for {1}. Make sure what "
|
||||
"correct application name has been passed. If application "
|
||||
"doesn't exist you can create it on admin.pubnub.com."
|
||||
)
|
||||
email = account.owner.email
|
||||
module.fail_json(
|
||||
msg=err_fmt.format(params["application"], email), changed=account.changed, module_cache=dict(account)
|
||||
msg=(
|
||||
f"There is no '{params['application']}' application for {email}. "
|
||||
"Make sure what correct application name has been passed. "
|
||||
"If application doesn't exist you can create it on admin.pubnub.com."
|
||||
),
|
||||
changed=account.changed,
|
||||
module_cache=dict(account),
|
||||
)
|
||||
|
||||
return application
|
||||
|
|
@ -346,13 +349,14 @@ def pubnub_keyset(module, account, application):
|
|||
params = module.params
|
||||
keyset = application.keyset(params["keyset"])
|
||||
if keyset is None:
|
||||
err_fmt = (
|
||||
"There is no '{0}' keyset for '{1}' application. Make "
|
||||
"sure what correct keyset name has been passed. If keyset "
|
||||
"doesn't exist you can create it on admin.pubnub.com."
|
||||
)
|
||||
module.fail_json(
|
||||
msg=err_fmt.format(params["keyset"], application.name), changed=account.changed, module_cache=dict(account)
|
||||
msg=(
|
||||
f"There is no '{params['keyset']}' keyset for '{application.name}' application. "
|
||||
"Make sure what correct keyset name has been passed. "
|
||||
"If keyset doesn't exist you can create it on admin.pubnub.com."
|
||||
),
|
||||
changed=account.changed,
|
||||
module_cache=dict(account),
|
||||
)
|
||||
|
||||
return keyset
|
||||
|
|
|
|||
|
|
@ -605,7 +605,8 @@ def main():
|
|||
if vars is None:
|
||||
return "null"
|
||||
elif isinstance(vars, str):
|
||||
return '"{string}"'.format(string=vars.replace("\\", "\\\\").replace('"', '\\"')).replace("\n", "\\n")
|
||||
escaped = vars.replace("\\", "\\\\").replace('"', '\\"')
|
||||
return f'"{escaped}"'.replace("\n", "\\n")
|
||||
elif isinstance(vars, bool):
|
||||
if vars:
|
||||
return "true"
|
||||
|
|
|
|||
|
|
@ -16,7 +16,7 @@ username = sys.argv[3]
|
|||
password = sys.argv[4]
|
||||
|
||||
if username:
|
||||
url = "http://{}:{}@127.0.0.1:9001/RPC2".format(quote(username, safe=""), quote(password, safe=""))
|
||||
url = f"http://{quote(username, safe='')}:{quote(password, safe='')}@127.0.0.1:9001/RPC2"
|
||||
else:
|
||||
url = "http://127.0.0.1:9001/RPC2"
|
||||
|
||||
|
|
|
|||
|
|
@ -63,17 +63,10 @@ def test_dzdo(mocker, parser, reset_cli_args):
|
|||
task["become_pass"] = "testpass"
|
||||
cmd = call_become_plugin(task, var_options, cmd=default_cmd, executable=default_exe)
|
||||
print(cmd)
|
||||
password_pattern = r"\"\[dzdo via ansible, key=.+?\] password:\""
|
||||
assert (
|
||||
re.match(
|
||||
"""{} {} -p {} -u {} {} -c 'echo {}; {}'""".format(
|
||||
dzdo_exe,
|
||||
dzdo_flags,
|
||||
r"\"\[dzdo via ansible, key=.+?\] password:\"",
|
||||
task["become_user"],
|
||||
default_exe,
|
||||
success,
|
||||
default_cmd,
|
||||
),
|
||||
f"""{dzdo_exe} {dzdo_flags} -p {password_pattern} -u {task["become_user"]} {default_exe} -c 'echo {success}; {default_cmd}'""",
|
||||
cmd,
|
||||
)
|
||||
is not None
|
||||
|
|
@ -112,17 +105,10 @@ def test_dzdo_varoptions(mocker, parser, reset_cli_args):
|
|||
var_options["ansible_become_pass"] = "testpass"
|
||||
cmd = call_become_plugin(task, var_options, cmd=default_cmd, executable=default_exe)
|
||||
print(cmd)
|
||||
password_pattern = r"\"\[dzdo via ansible, key=.+?\] password:\""
|
||||
assert (
|
||||
re.match(
|
||||
"""{} {} -p {} -u {} {} -c 'echo {}; {}'""".format(
|
||||
dzdo_exe,
|
||||
dzdo_flags,
|
||||
r"\"\[dzdo via ansible, key=.+?\] password:\"",
|
||||
var_options["ansible_become_user"],
|
||||
default_exe,
|
||||
success,
|
||||
default_cmd,
|
||||
),
|
||||
f"""{dzdo_exe} {dzdo_flags} -p {password_pattern} -u {var_options["ansible_become_user"]} {default_exe} -c 'echo {success}; {default_cmd}'""",
|
||||
cmd,
|
||||
)
|
||||
is not None
|
||||
|
|
|
|||
|
|
@ -50,17 +50,11 @@ def test_sudosu(mocker, parser, reset_cli_args):
|
|||
var_options = {}
|
||||
cmd = call_become_plugin(task, var_options, cmd=default_cmd, executable=default_exe)
|
||||
print(cmd)
|
||||
password_pattern = r"\[sudo via ansible, key=.+?\] password:"
|
||||
assert (
|
||||
re.match(
|
||||
"""{} {} -p "{}" su -l {} {} -c 'echo {}; {}'""".format(
|
||||
sudo_exe,
|
||||
sudo_flags.replace("-n", ""),
|
||||
r"\[sudo via ansible, key=.+?\] password:",
|
||||
task["become_user"],
|
||||
default_exe,
|
||||
success,
|
||||
default_cmd,
|
||||
),
|
||||
f"""{sudo_exe} {sudo_flags.replace("-n", "")} -p "{password_pattern}" """
|
||||
f"""su -l {task["become_user"]} {default_exe} -c 'echo {success}; {default_cmd}'""",
|
||||
cmd,
|
||||
)
|
||||
is not None
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue