saltstack/salt
1
0
Fork 0
mirror of https://github.com/saltstack/salt.git synced 2025-04-16 17:50:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Format source code using latest black

Browse source 
Signed-off-by: Pedro Algarvio <palgarvio@vmware.com>
This commit is contained in:
Pedro Algarvio 2022-07-20 13:50:28 +01:00 • committed by Megan Wilhite
parent 59ef42aa9e
commit b35e7d46dc
45 changed files with 172 additions and 210 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
doc/topics/troubleshooting/yaml_idiosyncrasies.rst
View file

@ -347,7 +347,7 @@ Python can also be used to discover the Unicode number for a character:
.. code-block:: python
repr(u"Text with wrong characters i need to figure out")
repr("Text with wrong characters i need to figure out")
This shell command can find wrong characters in your SLS files:

1
salt/_logging/handlers.py
View file

@ -275,7 +275,6 @@ if sys.version_info < (3, 7):
record.exc_text = None
return record
else:
class QueueHandler(

10
salt/grains/core.py
View file

@ -527,11 +527,11 @@ def _osx_memdata():
.replace(",", ".")
)
if swap_total.endswith("K"):
_power = 2 ** 10
_power = 2**10
elif swap_total.endswith("M"):
_power = 2 ** 20
_power = 2**20
elif swap_total.endswith("G"):
_power = 2 ** 30
_power = 2**30
swap_total = float(swap_total[:-1]) * _power
grains["mem_total"] = int(mem) // 1024 // 1024
@ -625,7 +625,7 @@ def _windows_memdata():
# get the Total Physical memory as reported by msinfo32
tot_bytes = win32api.GlobalMemoryStatusEx()["TotalPhys"]
# return memory info in gigabytes
grains["mem_total"] = int(tot_bytes / (1024 ** 2))
grains["mem_total"] = int(tot_bytes / (1024**2))
return grains
@ -3020,7 +3020,7 @@ def get_server_id():
return {}
id_ = __opts__.get("id", "")
hash_ = int(hashlib.sha256(id_.encode()).hexdigest(), 16)
return {"server_id": abs(hash_ % (2 ** 31))}
return {"server_id": abs(hash_ % (2**31))}
def get_master():

2
salt/modules/aptpkg.py
View file

@ -358,7 +358,7 @@ def _call_apt(args, scope=True, **kwargs):
while "Could not get lock" in cmd_ret.get("stderr", "") and count < 10:
count += 1
log.warning("Waiting for dpkg lock release: retrying... %s/100", count)
time.sleep(2 ** count)
time.sleep(2**count)
cmd_ret = __salt__["cmd.run_all"](cmd, **params)
return cmd_ret

4
salt/modules/bcache.py
View file

@ -910,7 +910,7 @@ def _size_map(size):
if re.search(r"[Kk]", size):
size = 1024 * float(re.sub(r"[Kk]", "", size))
elif re.search(r"[Mm]", size):
size = 1024 ** 2 * float(re.sub(r"[Mm]", "", size))
size = 1024**2 * float(re.sub(r"[Mm]", "", size))
size = int(size)
return size
except Exception: # pylint: disable=broad-except
@ -999,7 +999,7 @@ def _wipe(dev):
endres += _run_all(cmd, "warn", wipe_failmsg)
# Some stuff (<cough>GPT</cough>) writes stuff at the end of a dev as well
cmd += " seek={}".format((size / 1024 ** 2) - blocks)
cmd += " seek={}".format((size / 1024**2) - blocks)
endres += _run_all(cmd, "warn", wipe_failmsg)
elif wiper == "blkdiscard":

2
salt/modules/boto_kinesis.py
View file

@ -643,4 +643,4 @@ def _jittered_backoff(attempt, max_retry_delay):
salt myminion boto_kinesis._jittered_backoff current_attempt_number max_delay_in_seconds
"""
return min(random.random() * (2 ** attempt), max_retry_delay)
return min(random.random() * (2**attempt), max_retry_delay)

2
salt/modules/consul.py
View file

@ -280,7 +280,7 @@ def put(consul_url=None, token=None, key=None, value=None, **kwargs):
_current = get(consul_url=consul_url, token=token, key=key)
if "flags" in kwargs:
if kwargs["flags"] >= 0 and kwargs["flags"] <= 2 ** 64:
if kwargs["flags"] >= 0 and kwargs["flags"] <= 2**64:
query_params["flags"] = kwargs["flags"]
if "cas" in kwargs:

16
salt/modules/debian_service.py
View file

@ -29,16 +29,12 @@ def __virtual__():
"""
Only work on Debian and when systemd isn't running
"""
if (
__grains__["os"]
in (
"Debian",
"Raspbian",
"Devuan",
"NILinuxRT",
)
and not salt.utils.systemd.booted(__context__)
):
if __grains__["os"] in (
"Debian",
"Raspbian",
"Devuan",
"NILinuxRT",
) and not salt.utils.systemd.booted(__context__):
return __virtualname__
else:
return (

12
salt/modules/file.py
View file

@ -5498,14 +5498,10 @@ def check_managed_changes(
__clean_tmp(sfn)
return False, comments
if sfn and source and keep_mode:
if (
urllib.parse.urlparse(source).scheme
in (
"salt",
"file",
)
or source.startswith("/")
):
if urllib.parse.urlparse(source).scheme in (
"salt",
"file",
) or source.startswith("/"):
try:
mode = __salt__["cp.stat_file"](source, saltenv=saltenv, octal=True)
except Exception as exc: # pylint: disable=broad-except

23
salt/modules/lxc.py
View file

@ -1631,20 +1631,17 @@ def init(
run(name, "rm -f '{}'".format(SEED_MARKER), path=path, python_shell=False)
gid = "/.lxc.initial_seed"
gids = [gid, "/lxc.initial_seed"]
if (
any(
retcode(
name,
"test -e {}".format(x),
path=path,
chroot_fallback=True,
ignore_retcode=True,
)
== 0
for x in gids
if any(
retcode(
name,
"test -e {}".format(x),
path=path,
chroot_fallback=True,
ignore_retcode=True,
)
or not ret.get("result", True)
):
== 0
for x in gids
) or not ret.get("result", True):
pass
elif seed or seed_cmd:
if seed:

2
salt/modules/mac_user.py
View file

@ -65,7 +65,7 @@ def _dscl(cmd, ctype="create"):
def _first_avail_uid():
uids = {x.pw_uid for x in pwd.getpwall()}
for idx in range(501, 2 ** 24):
for idx in range(501, 2**24):
if idx not in uids:
return idx

18
salt/modules/network.py
View file

@ -1822,17 +1822,13 @@ def default_route(family=None):
if __grains__["kernel"] == "Linux":
default_route["inet"] = ["0.0.0.0", "default"]
default_route["inet6"] = ["::/0", "default"]
elif (
__grains__["os"]
in [
"FreeBSD",
"NetBSD",
"OpenBSD",
"MacOS",
"Darwin",
]
or __grains__["kernel"] in ("SunOS", "AIX")
):
elif __grains__["os"] in [
"FreeBSD",
"NetBSD",
"OpenBSD",
"MacOS",
"Darwin",
] or __grains__["kernel"] in ("SunOS", "AIX"):
default_route["inet"] = ["default"]
default_route["inet6"] = ["default"]
else:

4
salt/modules/openvswitch.py
View file

@ -407,7 +407,7 @@ def port_create_gre(br, port, id, remote):
salt '*' openvswitch.port_create_gre br0 gre1 5001 192.168.1.10
"""
if not 0 <= id < 2 ** 32:
if not 0 <= id < 2**32:
return False
elif not __salt__["dig.check_ip"](remote):
return False
@ -451,7 +451,7 @@ def port_create_vxlan(br, port, id, remote, dst_port=None):
salt '*' openvswitch.port_create_vxlan br0 vx1 5001 192.168.1.10 8472
"""
dst_port = " options:dst_port=" + str(dst_port) if 0 < dst_port <= 65535 else ""
if not 0 <= id < 2 ** 64:
if not 0 <= id < 2**64:
return False
elif not __salt__["dig.check_ip"](remote):
return False

16
salt/modules/win_system.py
View file

@ -508,16 +508,16 @@ def get_system_info():
def byte_calc(val):
val = float(val)
if val < 2 ** 10:
if val < 2**10:
return "{:.3f}B".format(val)
elif val < 2 ** 20:
return "{:.3f}KB".format(val / 2 ** 10)
elif val < 2 ** 30:
return "{:.3f}MB".format(val / 2 ** 20)
elif val < 2 ** 40:
return "{:.3f}GB".format(val / 2 ** 30)
elif val < 2**20:
return "{:.3f}KB".format(val / 2**10)
elif val < 2**30:
return "{:.3f}MB".format(val / 2**20)
elif val < 2**40:
return "{:.3f}GB".format(val / 2**30)
else:
return "{:.3f}TB".format(val / 2 ** 40)
return "{:.3f}TB".format(val / 2**40)
# Lookup dicts for Win32_OperatingSystem
os_type = {1: "Work Station", 2: "Domain Controller", 3: "Server"}

2
salt/modules/win_task.py
View file

@ -676,7 +676,7 @@ def create_task_from_xml(
except pythoncom.com_error as error:
hr, msg, exc, arg = error.args # pylint: disable=W0633
error_code = hex(exc[5] + 2 ** 32)
error_code = hex(exc[5] + 2**32)
fc = {
0x80041319: "Required element or attribute missing",
0x80041318: "Value incorrectly formatted or out of range",

2
salt/platform/win.py
View file

@ -212,7 +212,7 @@ class LARGE_INTEGER(wintypes.LARGE_INTEGER):
@classmethod
def from_time(cls, t):
time100ns = int(t * 10 ** 7)
time100ns = int(t * 10**7)
return cls(time100ns + cls._unix_epoch)

3
salt/serializers/msgpack.py
View file

@ -27,7 +27,6 @@ if not available:
def _deserialize(stream_or_string, **options):
_fail()
elif salt.utils.msgpack.version >= (1, 0, 0):
def _serialize(obj, **options):
@ -44,7 +43,6 @@ elif salt.utils.msgpack.version >= (1, 0, 0):
except Exception as error: # pylint: disable=broad-except
raise DeserializationError(error)
elif salt.utils.msgpack.version >= (0, 2, 0):
def _serialize(obj, **options):
@ -61,7 +59,6 @@ elif salt.utils.msgpack.version >= (0, 2, 0):
except Exception as error: # pylint: disable=broad-except
raise DeserializationError(error)
else: # msgpack.version < 0.2.0
def _encoder(obj):

13
salt/states/boto_apigateway.py
View file

@ -1576,12 +1576,13 @@ class _Swagger:
ret["result"] = False
ret["abort"] = True
if "error" in update_model_schema_response:
ret[
"comment"
] = "Failed to update existing model {} with schema {}, " "error: {}".format(
model,
_dict_to_json_pretty(schema),
update_model_schema_response["error"]["message"],
ret["comment"] = (
"Failed to update existing model {} with schema {}, "
"error: {}".format(
model,
_dict_to_json_pretty(schema),
update_model_schema_response["error"]["message"],
)
)
return ret

4
salt/states/esxi.py
View file

@ -1691,13 +1691,13 @@ def host_cache_configured(
)
)
else:
if (existing_datastore["capacity"] / 1024.0 ** 2) < swap_size_MiB:
if (existing_datastore["capacity"] / 1024.0**2) < swap_size_MiB:
raise ArgumentValueError(
"Capacity of host cache datastore '{}' ({} MiB) is "
"smaller than the required swap size ({} MiB)".format(
existing_datastore["name"],
existing_datastore["capacity"] / 1024.0 ** 2,
existing_datastore["capacity"] / 1024.0**2,
swap_size_MiB,
)
)

4
salt/states/openvswitch_port.py
View file

@ -199,7 +199,7 @@ def present(
def _check_gre():
interface_options = __salt__["openvswitch.interface_get_options"](name)
interface_type = __salt__["openvswitch.interface_get_type"](name)
if not 0 <= id <= 2 ** 32:
if not 0 <= id <= 2**32:
ret["result"] = False
ret["comment"] = comments["comment_gre_invalid_id"]
elif not __salt__["dig.check_ip"](remote):
@ -223,7 +223,7 @@ def present(
def _check_vxlan():
interface_options = __salt__["openvswitch.interface_get_options"](name)
interface_type = __salt__["openvswitch.interface_get_type"](name)
if not 0 <= id <= 2 ** 64:
if not 0 <= id <= 2**64:
ret["result"] = False
ret["comment"] = comments["comment_vxlan_invalid_id"]
elif not __salt__["dig.check_ip"](remote):

11
salt/states/zone.py
View file

@ -734,11 +734,12 @@ def export(name, path, replace=False):
if __salt__["file.file_exists"](cfg_tmp):
__salt__["file.remove"](cfg_tmp)
ret["result"] = False
ret[
"comment"
] = "Unable to be re-export zone configuration for {}" " to {}!".format(
name,
path,
ret["comment"] = (
"Unable to be re-export zone configuration for {}"
" to {}!".format(
name,
path,
)
)
else:
ret["result"] = True

1
salt/utils/atomicfile.py
View file

@ -98,7 +98,6 @@ if os.name == "nt": # pragma: no cover
except Exception: # pylint: disable=broad-except
pass
else:
atomic_rename = os.rename # pylint: disable=C0103
CAN_RENAME_OPEN_FILE = True

2
salt/utils/aws.py
View file

@ -68,7 +68,7 @@ def sleep_exponential_backoff(attempts):
A failure rate of >10% is observed when using the salt-api with an asynchronous client
specified (runner_async).
"""
time.sleep(random.uniform(1, 2 ** attempts))
time.sleep(random.uniform(1, 2**attempts))
def creds(provider):

2
salt/utils/dateutils.py
View file

@ -89,4 +89,4 @@ def total_seconds(td):
represented by the object. Wrapper for the total_seconds()
method which does not exist in versions of Python < 2.7.
"""
return (td.microseconds + (td.seconds + td.days * 24 * 3600) * 10 ** 6) / 10 ** 6
return (td.microseconds + (td.seconds + td.days * 24 * 3600) * 10**6) / 10**6

10
salt/utils/find.py
View file

@ -193,11 +193,11 @@ def _parse_size(value):
if scalar:
multiplier = {
"b": 2 ** 0,
"k": 2 ** 10,
"m": 2 ** 20,
"g": 2 ** 30,
"t": 2 ** 40,
"b": 2**0,
"k": 2**10,
"m": 2**20,
"g": 2**30,
"t": 2**40,
}.get(scalar[-1].lower())
if multiplier:
scalar = scalar[:-1].strip()

2
salt/utils/iam.py
View file

@ -39,7 +39,7 @@ def _retry_get_url(url, num_retries=10, timeout=5):
log.warning("Caught exception reading from URL. Retry no. %s", i)
log.warning(pprint.pformat(exc))
time.sleep(2 ** i)
time.sleep(2**i)
log.error("Failed to read from URL for %s times. Giving up.", num_retries)
return ""

1
salt/utils/odict.py
View file

@ -29,7 +29,6 @@ try:
class OrderedDict(collections.OrderedDict):
__hash__ = None
except (ImportError, AttributeError):
try:
import ordereddict

2
tests/pytests/functional/transport/ipc/test_client.py
View file

@ -122,7 +122,7 @@ async def test_send_many(channel):
async def test_very_big_message(channel):
long_str = "".join([str(num) for num in range(10 ** 5)])
long_str = "".join([str(num) for num in range(10**5)])
msg = {"long_str": long_str, "stop": True}
await channel.send(msg)
assert channel.payloads[0] == msg

40
tests/pytests/unit/modules/test_beacons.py
View file

@ -90,20 +90,17 @@ def test_add():
mock = MagicMock(return_value=True)
with patch.dict(beacons.__salt__, {"event.fire": mock}):
with patch.object(SaltEvent, "get_event", side_effect=event_returns):
assert (
beacons.add(
"ps",
[
{
"processes": {
"salt-master": "stopped",
"apache2": "stopped",
}
assert beacons.add(
"ps",
[
{
"processes": {
"salt-master": "stopped",
"apache2": "stopped",
}
],
)
== {"comment": comm1, "result": True}
)
}
],
) == {"comment": comm1, "result": True}
@pytest.mark.slow_test
@ -241,16 +238,13 @@ def test_add_beacon_module():
mock = MagicMock(return_value=True)
with patch.dict(beacons.__salt__, {"event.fire": mock}):
with patch.object(SaltEvent, "get_event", side_effect=event_returns):
assert (
beacons.add(
"watch_salt_master",
[
{"processes": {"salt-master": "stopped"}},
{"beacon_module": "ps"},
],
)
== {"comment": comm1, "result": True}
)
assert beacons.add(
"watch_salt_master",
[
{"processes": {"salt-master": "stopped"}},
{"beacon_module": "ps"},
],
) == {"comment": comm1, "result": True}
@pytest.mark.slow_test

24
tests/pytests/unit/modules/virt/test_domain.py
View file

@ -1354,7 +1354,7 @@ def test_update_add_cpu_topology(make_mock_vm):
assert setxml.find("./cpu/feature[@name='lahf']").get("policy") == "optional"
assert setxml.find("./cpu/numa/cell/[@id='0']").get("cpus") == "0,1,2,3"
assert setxml.find("./cpu/numa/cell/[@id='0']").get("memory") == str(1024 ** 3)
assert setxml.find("./cpu/numa/cell/[@id='0']").get("memory") == str(1024**3)
assert setxml.find("./cpu/numa/cell/[@id='0']").get("unit") == "bytes"
assert setxml.find("./cpu/numa/cell/[@id='0']").get("discard") == "yes"
assert (
@ -1383,7 +1383,7 @@ def test_update_add_cpu_topology(make_mock_vm):
)
assert setxml.find("./cpu/numa/cell/[@id='1']").get("cpus") == "4,5,6"
assert setxml.find("./cpu/numa/cell/[@id='1']").get("memory") == str(
int(1024 ** 3 / 2)
int(1024**3 / 2)
)
assert setxml.find("./cpu/numa/cell/[@id='1']").get("unit") == "bytes"
assert setxml.find("./cpu/numa/cell/[@id='1']").get("discard") == "no"
@ -1571,10 +1571,10 @@ def test_update_add_memtune(make_mock_vm):
assert ret["definition"]
setxml = ET.fromstring(virt.libvirt.openAuth().defineXML.call_args[0][0])
assert_equal_unit(setxml.find("memtune/soft_limit"), int(0.5 * 1024 ** 3), "bytes")
assert_equal_unit(setxml.find("memtune/hard_limit"), 1024 * 1024 ** 2, "bytes")
assert_equal_unit(setxml.find("memtune/swap_hard_limit"), 2048 * 1024 ** 2, "bytes")
assert_equal_unit(setxml.find("memtune/min_guarantee"), 1 * 1024 ** 3, "bytes")
assert_equal_unit(setxml.find("memtune/soft_limit"), int(0.5 * 1024**3), "bytes")
assert_equal_unit(setxml.find("memtune/hard_limit"), 1024 * 1024**2, "bytes")
assert_equal_unit(setxml.find("memtune/swap_hard_limit"), 2048 * 1024**2, "bytes")
assert_equal_unit(setxml.find("memtune/min_guarantee"), 1 * 1024**3, "bytes")
def test_update_add_memtune_invalid_unit(make_mock_vm):
@ -1625,7 +1625,7 @@ def test_update_mem_simple(make_mock_vm):
assert ret["definition"]
assert ret["mem"]
setxml = ET.fromstring(virt.libvirt.openAuth().defineXML.call_args[0][0])
assert setxml.find("memory").text == str(2048 * 1024 ** 2)
assert setxml.find("memory").text == str(2048 * 1024**2)
assert setxml.find("memory").get("unit") == "bytes"
assert domain_mock.setMemoryFlags.call_args[0][0] == 2048 * 1024
@ -1644,9 +1644,9 @@ def test_update_mem(make_mock_vm):
assert ret["mem"]
setxml = ET.fromstring(virt.libvirt.openAuth().defineXML.call_args[0][0])
assert setxml.find("memory").get("unit") == "bytes"
assert setxml.find("memory").text == str(int(0.5 * 1024 ** 3))
assert setxml.find("maxMemory").text == str(1 * 1024 ** 3)
assert setxml.find("currentMemory").text == str(2 * 1024 ** 3)
assert setxml.find("memory").text == str(int(0.5 * 1024**3))
assert setxml.find("maxMemory").text == str(1 * 1024**3)
assert setxml.find("currentMemory").text == str(2 * 1024**3)
def test_update_add_mem_backing(make_mock_vm):
@ -1676,8 +1676,8 @@ def test_update_add_mem_backing(make_mock_vm):
p.get("nodeset"): {"size": p.get("size"), "unit": p.get("unit")}
for p in setxml.findall("memoryBacking/hugepages/page")
} == {
"1,2,3,5": {"size": str(1024 ** 3), "unit": "bytes"},
"4": {"size": str(2 * 1024 ** 3), "unit": "bytes"},
"1,2,3,5": {"size": str(1024**3), "unit": "bytes"},
"4": {"size": str(2 * 1024**3), "unit": "bytes"},
}
assert setxml.find("./memoryBacking/nosharepages") is not None
assert setxml.find("./memoryBacking/nosharepages").text is None

2
tests/pytests/unit/utils/templates/test_jinja.py
View file

@ -60,7 +60,7 @@ def test_jinja_undefined_error_context(minion_opts, local_salt):
# exception was raised is to match on the initial wording of the exception
# message.
match_regex = re.compile(
fr"^Jinja variable .*; line .*{marker}$", re.DOTALL | re.MULTILINE
rf"^Jinja variable .*; line .*{marker}$", re.DOTALL | re.MULTILINE
)
with pytest.raises(SaltRenderError, match=match_regex):
render_jinja_tmpl(

18
tests/pytests/unit/utils/test_stringutils.py
View file

@ -722,14 +722,14 @@ def test_human_to_bytes(unit):
# second multiplier is metric/decimal
conversion = {
"B": (1, 1),
"K": (2 ** 10, 10 ** 3),
"M": (2 ** 20, 10 ** 6),
"G": (2 ** 30, 10 ** 9),
"T": (2 ** 40, 10 ** 12),
"P": (2 ** 50, 10 ** 15),
"E": (2 ** 60, 10 ** 18),
"Z": (2 ** 70, 10 ** 21),
"Y": (2 ** 80, 10 ** 24),
"K": (2**10, 10**3),
"M": (2**20, 10**6),
"G": (2**30, 10**9),
"T": (2**40, 10**12),
"P": (2**50, 10**15),
"E": (2**60, 10**18),
"Z": (2**70, 10**21),
"Y": (2**80, 10**24),
}
idx = 0
@ -774,7 +774,7 @@ def test_human_to_bytes_edge_cases():
# no unit - bytes
assert salt.utils.stringutils.human_to_bytes("32") == 32
# no unit - default MB
assert salt.utils.stringutils.human_to_bytes("32", default_unit="M") == 32 * 2 ** 20
assert salt.utils.stringutils.human_to_bytes("32", default_unit="M") == 32 * 2**20
# bad value
assert salt.utils.stringutils.human_to_bytes("32-1") == 0
assert salt.utils.stringutils.human_to_bytes("3.4.MB") == 0

2
tests/pytests/unit/utils/test_win_functions.py
View file

@ -14,7 +14,6 @@ try:
class WinError(win32net.error):
winerror = 0
except ImportError:
HAS_WIN32 = False
@ -26,7 +25,6 @@ try:
class PyWinError(pywintypes.error):
pywinerror = 0
except ImportError:
HAS_PYWIN = False

2
tests/support/helpers.py
View file

@ -265,7 +265,7 @@ def flaky(caller=None, condition=True, attempts=4):
teardown = getattr(cls, "tearDown", None)
if callable(teardown):
teardown()
backoff_time = attempt ** 2
backoff_time = attempt**2
log.info("Found Exception. Waiting %s seconds to retry.", backoff_time)
time.sleep(backoff_time)
return cls

2
tests/support/pytest/helpers.py
View file

@ -735,7 +735,7 @@ class EntropyGenerator:
"-out",
target_file.name,
"-base64",
str(int(2 ** 30 * 3 / 4)), # 1GB
str(int(2**30 * 3 / 4)), # 1GB
],
shell=False,
check=True,

1
tests/support/xmlunit.py
View file

@ -89,7 +89,6 @@ try:
self.stream.writeln("Finished generating XML reports")
return result
except ImportError:
HAS_XMLRUNNER = False

26
tests/unit/ext/test_ipaddress.py
View file

@ -174,8 +174,8 @@ class CommonTestMixin_v4(CommonTestMixin):
def test_large_ints_rejected(self):
msg = "%d (>= 2**32) is not permitted as an IPv4 address"
with self.assertAddressError(re.escape(msg % 2 ** 32)):
self.factory(2 ** 32)
with self.assertAddressError(re.escape(msg % 2**32)):
self.factory(2**32)
def test_bad_packed_length(self):
def assertBadLength(length):
@ -213,8 +213,8 @@ class CommonTestMixin_v6(CommonTestMixin):
def test_large_ints_rejected(self):
msg = "%d (>= 2**128) is not permitted as an IPv6 address"
with self.assertAddressError(re.escape(msg % 2 ** 128)):
self.factory(2 ** 128)
with self.assertAddressError(re.escape(msg % 2**128)):
self.factory(2**128)
def test_bad_packed_length(self):
def assertBadLength(length):
@ -1275,30 +1275,30 @@ class IpaddrUnitTest(TestCase):
ipaddress.IPv4Address("1.1.1.1") - 256, ipaddress.IPv4Address("1.1.0.1")
)
self.assertEqual(
ipaddress.IPv6Address("::1") + (2 ** 16 - 2),
ipaddress.IPv6Address("::1") + (2**16 - 2),
ipaddress.IPv6Address("::ffff"),
)
self.assertEqual(
ipaddress.IPv6Address("::ffff") - (2 ** 16 - 2),
ipaddress.IPv6Address("::ffff") - (2**16 - 2),
ipaddress.IPv6Address("::1"),
)
self.assertNotEqual(
ipaddress.IPv6Address("::1%scope") + (2 ** 16 - 2),
ipaddress.IPv6Address("::1%scope") + (2**16 - 2),
ipaddress.IPv6Address("::ffff%scope"),
)
self.assertNotEqual(
ipaddress.IPv6Address("::ffff%scope") - (2 ** 16 - 2),
ipaddress.IPv6Address("::ffff%scope") - (2**16 - 2),
ipaddress.IPv6Address("::1%scope"),
)
def testInvalidIntToBytes(self):
self.assertRaises(ValueError, ipaddress.v4_int_to_packed, -1)
self.assertRaises(
ValueError, ipaddress.v4_int_to_packed, 2 ** ipaddress.IPV4LENGTH
ValueError, ipaddress.v4_int_to_packed, 2**ipaddress.IPV4LENGTH
)
self.assertRaises(ValueError, ipaddress.v6_int_to_packed, -1)
self.assertRaises(
ValueError, ipaddress.v6_int_to_packed, 2 ** ipaddress.IPV6LENGTH
ValueError, ipaddress.v6_int_to_packed, 2**ipaddress.IPV6LENGTH
)
def testInternals(self):
@ -2462,7 +2462,7 @@ class IpaddrUnitTest(TestCase):
def testReservedIpv6(self):
self.assertEqual(True, ipaddress.ip_network("ffff::").is_multicast)
self.assertEqual(True, ipaddress.ip_network(2 ** 128 - 1).is_multicast)
self.assertEqual(True, ipaddress.ip_network(2**128 - 1).is_multicast)
self.assertEqual(True, ipaddress.ip_network("ff00::").is_multicast)
self.assertEqual(False, ipaddress.ip_network("fdff::").is_multicast)
@ -2496,7 +2496,7 @@ class IpaddrUnitTest(TestCase):
self.assertEqual(True, ipaddress.ip_network("200::1/128").is_global)
# test addresses
self.assertEqual(True, ipaddress.ip_address("ffff::").is_multicast)
self.assertEqual(True, ipaddress.ip_address(2 ** 128 - 1).is_multicast)
self.assertEqual(True, ipaddress.ip_address(2**128 - 1).is_multicast)
self.assertEqual(True, ipaddress.ip_address("ff00::").is_multicast)
self.assertEqual(False, ipaddress.ip_address("fdff::").is_multicast)
@ -2603,7 +2603,7 @@ class IpaddrUnitTest(TestCase):
net = self.ipv4_network
self.assertEqual("1.2.3.0/24", net.compressed)
net = self.ipv6_network
self.assertRaises(ValueError, net._string_from_ip_int, 2 ** 128 + 1)
self.assertRaises(ValueError, net._string_from_ip_int, 2**128 + 1)
def testIPv6NetworkHelpers(self):
net = self.ipv6_network

2
tests/unit/modules/test_boto_secgroup.py
View file

@ -69,7 +69,7 @@ boto_conn_parameters = {
def _random_group_id():
group_id = "sg-{:x}".format(random.randrange(2 ** 32))
group_id = "sg-{:x}".format(random.randrange(2**32))
return group_id

36
tests/unit/modules/test_kubeadm.py
View file

@ -639,16 +639,13 @@ class KubeAdmTestCase(TestCase, LoaderModuleMockMixin):
"cmd.run_all": MagicMock(return_value=result),
}
with patch.dict(kubeadm.__salt__, salt_mock):
assert (
kubeadm.config_images_list(
config="/kubeadm.cfg",
feature_gates="k=v",
kubernetes_version="version",
kubeconfig="/kube.cfg",
rootfs="/mnt",
)
== ["image1", "image2"]
)
assert kubeadm.config_images_list(
config="/kubeadm.cfg",
feature_gates="k=v",
kubernetes_version="version",
kubeconfig="/kube.cfg",
rootfs="/mnt",
) == ["image1", "image2"]
salt_mock["cmd.run_all"].assert_called_with(
[
"kubeadm",
@ -709,17 +706,14 @@ class KubeAdmTestCase(TestCase, LoaderModuleMockMixin):
"cmd.run_all": MagicMock(return_value=result),
}
with patch.dict(kubeadm.__salt__, salt_mock):
assert (
kubeadm.config_images_pull(
config="/kubeadm.cfg",
cri_socket="socket",
feature_gates="k=v",
kubernetes_version="version",
kubeconfig="/kube.cfg",
rootfs="/mnt",
)
== ["image1", "image2"]
)
assert kubeadm.config_images_pull(
config="/kubeadm.cfg",
cri_socket="socket",
feature_gates="k=v",
kubernetes_version="version",
kubeconfig="/kube.cfg",
rootfs="/mnt",
) == ["image1", "image2"]
salt_mock["cmd.run_all"].assert_called_with(
[
"kubeadm",

1
tests/unit/modules/test_libcloud_compute.py
View file

@ -194,7 +194,6 @@ if HAS_LIBCLOUD:
assert key_pair.name == "test_key"
return True
else:
MockComputeDriver = object

1
tests/unit/modules/test_libcloud_loadbalancer.py
View file

@ -77,7 +77,6 @@ if HAS_LIBCLOUD:
assert balancer.id == "test_id"
return [self._TEST_MEMBER]
else:
MockLBDriver = object

1
tests/unit/modules/test_libcloud_storage.py
View file

@ -53,7 +53,6 @@ if HAS_LIBCLOUD:
assert object_name == "test_obj"
return self._TEST_OBJECT
else:
MockStorageDriver = object

46
tests/unit/modules/test_virt.py
View file

@ -562,7 +562,7 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
self.assertFalse("slots" in root.find("maxMemory").keys())
self.assert_equal_unit(root.find("memtune/hard_limit"), 1024 * 1024)
self.assert_equal_unit(root.find("memtune/soft_limit"), 512 * 1024)
self.assert_equal_unit(root.find("memtune/swap_hard_limit"), 1024 ** 2)
self.assert_equal_unit(root.find("memtune/swap_hard_limit"), 1024**2)
self.assert_equal_unit(root.find("memtune/min_guarantee"), 256 * 1024)
self.assertEqual(
[
@ -705,7 +705,7 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
cell0 = root.find("cpu/numa/cell[@id='0']")
self.assertEqual(cell0.get("cpus"), "0-3")
self.assertIsNone(cell0.get("unit"))
self.assertEqual(cell0.get("memory"), str(1024 ** 2))
self.assertEqual(cell0.get("memory"), str(1024**2))
self.assertEqual(cell0.get("discard"), "yes")
self.assertEqual(
{d.get("id"): d.get("value") for d in cell0.findall("distances/sibling")},
@ -715,7 +715,7 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
cell1 = root.find("cpu/numa/cell[@id='1']")
self.assertEqual(cell1.get("cpus"), "4-7")
self.assertIsNone(cell0.get("unit"))
self.assertEqual(cell1.get("memory"), str(2 * 1024 ** 2))
self.assertEqual(cell1.get("memory"), str(2 * 1024**2))
self.assertFalse("discard" in cell1.keys())
self.assertEqual(
{d.get("id"): d.get("value") for d in cell1.findall("distances/sibling")},
@ -3253,7 +3253,7 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
)
self.assertEqual(
setxml.find("./cpu/numa/cell/[@id='0']").attrib["memory"],
str(512 * 1024 ** 2),
str(512 * 1024**2),
)
self.assertEqual(
setxml.find("./cpu/numa/cell/[@id='0']").get("unit"),
@ -3292,7 +3292,7 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
)
self.assertEqual(
setxml.find("./cpu/numa/cell/[@id='1']").attrib["memory"],
str(int(2 * 1024 ** 3)),
str(int(2 * 1024**3)),
)
self.assertEqual(
setxml.find("./cpu/numa/cell/[@id='1']").get("unit"),
@ -3360,7 +3360,7 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
)
self.assertEqual(
setxml.find("./cpu/numa/cell/[@id='0']").attrib["memory"],
str(512 * 1024 ** 2),
str(512 * 1024**2),
)
self.assertEqual(
setxml.find("./cpu/numa/cell/[@id='0']").get("unit"),
@ -3396,7 +3396,7 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
)
self.assertEqual(
setxml.find("./cpu/numa/cell/[@id='1']").attrib["memory"],
str(int(2 * 1024 ** 3)),
str(int(2 * 1024**3)),
)
self.assertEqual(
setxml.find("./cpu/numa/cell/[@id='1']").attrib["discard"], "yes"
@ -3429,7 +3429,7 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
)
self.assertEqual(
setxml.find("./cpu/numa/cell/[@id='1']").attrib["memory"],
str(int(1024 ** 3 * 2)),
str(int(1024**3 * 2)),
)
self.assertEqual(
setxml.find("./cpu/numa/cell/[@id='1']").attrib["discard"], "yes"
@ -3506,7 +3506,7 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
virt.update("vm_with_memtune_param", mem=memtune_new_val),
)
self.assertEqual(
domain_mock.setMemoryFlags.call_args[0][0], int(2.5 * 1024 ** 2)
domain_mock.setMemoryFlags.call_args[0][0], int(2.5 * 1024**2)
)
setxml = ET.fromstring(define_mock.call_args[0][0])
@ -3518,22 +3518,22 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
)
self.assertEqual(
setxml.find("memtune").find("swap_hard_limit").text,
str(int(2.5 * 1024 ** 2)),
str(int(2.5 * 1024**2)),
)
self.assertEqual(
setxml.find("memtune").find("swap_hard_limit").get("unit"),
"KiB",
)
self.assertEqual(
setxml.find("memtune").find("min_guarantee").text, str(1 * 1024 ** 3)
setxml.find("memtune").find("min_guarantee").text, str(1 * 1024**3)
)
self.assertEqual(
setxml.find("memtune").find("min_guarantee").attrib.get("unit"), "bytes"
)
self.assertEqual(setxml.find("maxMemory").text, str(3096 * 1024 ** 2))
self.assertEqual(setxml.find("maxMemory").text, str(3096 * 1024**2))
self.assertEqual(setxml.find("maxMemory").attrib.get("slots"), "10")
self.assertEqual(setxml.find("currentMemory").text, str(int(2.5 * 1024 ** 3)))
self.assertEqual(setxml.find("memory").text, str(int(0.7 * 1024 ** 3)))
self.assertEqual(setxml.find("currentMemory").text, str(int(2.5 * 1024**3)))
self.assertEqual(setxml.find("memory").text, str(int(0.7 * 1024**3)))
max_slot_reverse = {
"slots": "10",
@ -3548,7 +3548,7 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
virt.update("vm_with_memtune_param", mem=max_slot_reverse),
)
setxml = ET.fromstring(define_mock.call_args[0][0])
self.assertEqual(setxml.find("maxMemory").text, str(3096 * 1024 ** 2))
self.assertEqual(setxml.find("maxMemory").text, str(3096 * 1024**2))
self.assertEqual(setxml.find("maxMemory").get("unit"), "bytes")
self.assertEqual(setxml.find("maxMemory").attrib.get("slots"), "10")
@ -3574,7 +3574,7 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
virt.update("vm_with_memtune_param", mem=max_swap_none),
)
self.assertEqual(
domain_mock.setMemoryFlags.call_args[0][0], int(2.5 * 1024 ** 2)
domain_mock.setMemoryFlags.call_args[0][0], int(2.5 * 1024**2)
)
setxml = ET.fromstring(define_mock.call_args[0][0])
@ -3586,14 +3586,14 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
)
self.assertEqual(setxml.find("memtune").find("swap_hard_limit"), None)
self.assertEqual(
setxml.find("memtune").find("min_guarantee").text, str(1 * 1024 ** 3)
setxml.find("memtune").find("min_guarantee").text, str(1 * 1024**3)
)
self.assertEqual(
setxml.find("memtune").find("min_guarantee").attrib.get("unit"), "bytes"
)
self.assertEqual(setxml.find("maxMemory").text, None)
self.assertEqual(setxml.find("currentMemory").text, str(int(2.5 * 1024 ** 3)))
self.assertEqual(setxml.find("memory").text, str(int(0.7 * 1024 ** 3)))
self.assertEqual(setxml.find("currentMemory").text, str(int(2.5 * 1024**3)))
self.assertEqual(setxml.find("memory").text, str(int(0.7 * 1024**3)))
memtune_none = {
"soft_limit": None,
@ -3632,8 +3632,8 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
setxml = ET.fromstring(define_mock.call_args[0][0])
self.assertEqual(setxml.find("maxMemory"), None)
self.assertEqual(setxml.find("currentMemory").text, str(int(1 * 1024 ** 2)))
self.assertEqual(setxml.find("memory").text, str(int(1 * 1024 ** 2)))
self.assertEqual(setxml.find("currentMemory").text, str(int(1 * 1024**2)))
self.assertEqual(setxml.find("memory").text, str(int(1 * 1024**2)))
def test_update_exist_memorybacking_params(self):
"""
@ -3698,8 +3698,8 @@ class VirtTestCase(TestCase, LoaderModuleMockMixin):
for p in setxml.findall("memoryBacking/hugepages/page")
},
{
"1,2,4": {"size": str(1024 ** 3), "unit": "bytes"},
"3": {"size": str(2 * 1024 ** 3), "unit": "bytes"},
"1,2,4": {"size": str(1024**3), "unit": "bytes"},
"3": {"size": str(2 * 1024**3), "unit": "bytes"},
},
)
self.assertEqual(setxml.find("./memoryBacking/nosharepages"), None)

4
tests/unit/utils/test_find.py
View file

@ -255,7 +255,7 @@ class TestFind(TestCase):
self.assertEqual(option.match("", "", [1] * 9), False)
option = salt.utils.find.MtimeOption("mtime", "-1s")
self.assertEqual(option.match("", "", [10 ** 10] * 9), True)
self.assertEqual(option.match("", "", [10**10] * 9), True)
class TestGrepOption(TestCase):
@ -376,7 +376,7 @@ class TestPrintOption(TestCase):
self.assertEqual(option.execute("", [0] * 10), "root")
option = salt.utils.find.PrintOption("print", "user")
self.assertEqual(option.execute("", [2 ** 31] * 10), 2 ** 31)
self.assertEqual(option.execute("", [2**31] * 10), 2**31)
@skipIf(sys.platform.startswith("win"), "grp not available on Windows")
def test_print_group(self):

1
tests/unit/utils/test_win_functions.py
View file

@ -11,7 +11,6 @@ try:
class WinError(win32net.error):
winerror = 0
except ImportError:
HAS_WIN32 = False