openstack/pyeclib
Code Issues Proposed changes

Add some CLI tools

Browse Source 
pyeclib-backend version
      Print the versions of pyeclib, liberasurecode, and Python

   pyeclib-backend list
      List the status of backends

   pyeclib-backend check
      Check the status of a particular backend

   pyeclib-backend verify
      Verify the ability to decode given some unavailable fragments

   pyeclib-backend bench
      Benchmarks available backends

Co-Authored-By: Matthew Oliver <matt@oliver.net.au>
Signed-off-by: Tim Burke <tim.burke@gmail.com>
Change-Id: Ibe47b4665bfe763c07e68d8be0f92983bc15dff0
This commit is contained in:
Tim Burke
2025-08-21 14:02:42 -07:00
parent 1c6dc6fcc6
commit 301e8f0cd7
12 changed files with 835 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
doc/source/cli.rst Normal file
View File

@@ -0,0 +1,56 @@
PyECLib CLI
===========
PyECLib provides a ``pyeclib-backend`` tool to provide various information about backends.
``version`` subcommand
----------------------
.. code:: text
pyeclib-backend [-V | version]
Displays the versions of pyeclib, liberasurecode, and python.
``list`` subcommand
-------------------
.. code:: text
pyeclib-backend list [-a | --available] [<ec_type>]
Displays the status (available, missing, or unknwon) of requested backends.
By default, all backends are displayed; if ``--available`` is provided, only
available backends are displayed, and status is not displayed.
``check`` subcommand
--------------------
.. code:: text
pyeclib-backend check [-q | --quiet] <ec_type>
Check whether a backend is available. Exits
- 0 if ``ec_type`` is available,
- 1 if ``ec_type`` is missing, or
- 2 if ``ec_type`` is not recognized
If ``--quiet`` is provided, output nothing; rely only on exit codes.
``verify`` subcommand
---------------------
.. code:: text
pyeclib-backend verify [-q | --quiet] [--ec-type=all]
[--n-data=10] [--n-parity=5] [--unavailable=2] [--segment-size=1024]
Verify the ability to decode all combinations of fragments given some number
of unavailable fragments.
``bench`` subcommand
--------------------
.. code:: text
pyeclib-backend bench [-e | --encode] [-d | --decode] [--ec-type=all]
[--n-data=10] [--n-parity=5] [--unavailable=2] [--segment-size=1048576]
[--iterations=200]
Benchmark one or more backends.

2
doc/source/index.rst
View File

@@ -6,6 +6,8 @@ Contents:
.. toctree::
:maxdepth: 2
cli
Indices and tables

92
pyeclib/cli/__init__.py Normal file
View File

@@ -0,0 +1,92 @@
# Copyright (c) 2025, NVIDIA
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# Redistributions of source code must retain the above copyright notice, this
# list of conditions and the following disclaimer.
#
# Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution. THIS SOFTWARE IS
# PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
# OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN
# NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
# THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
from pyeclib import ec_iface
_type_abbreviations = {
# name -> prefix
"all": "",
"isa-l": "isa_l_",
"isa_l": "isa_l_",
"isal": "isa_l_",
"jerasure": "jerasure_",
"flat-xor": "flat_xor_",
"flat_xor": "flat_xor_",
"flatxor": "flat_xor_",
"xor": "flat_xor_",
}
def expand_ec_types(user_types):
result = set(user_types or ["all"])
for abbrev, prefix in _type_abbreviations.items():
if abbrev in result:
result.remove(abbrev)
result.update(
ec_type
for ec_type in ec_iface.ALL_EC_TYPES
if ec_type.startswith(prefix)
)
return sorted(result)
def add_instance_args(parser, default_segment_size=1024):
"""
Add arguments to ``parser`` for instance instantiation.
"""
parser.add_argument(
"--ec-type",
action="append",
type=str,
)
parser.add_argument(
"--n-data",
"--ndata",
"-k",
metavar="K",
type=int,
default=10,
)
parser.add_argument(
"--n-parity",
"--nparity",
"-m",
metavar="M",
type=int,
default=5,
)
parser.add_argument(
"--unavailable",
"-u",
metavar="N",
type=int,
default=2,
)
parser.add_argument(
"--segment-size",
"-s",
metavar="BYTES",
type=int,
default=default_segment_size,
)

83
pyeclib/cli/__main__.py Normal file
View File

@@ -0,0 +1,83 @@
# Copyright (c) 2025, NVIDIA
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# Redistributions of source code must retain the above copyright notice, this
# list of conditions and the following disclaimer.
#
# Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution. THIS SOFTWARE IS
# PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
# OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN
# NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
# THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import argparse
import sys
from pyeclib.cli import bench
from pyeclib.cli import check
from pyeclib.cli import list as list_cli
from pyeclib.cli import verify
from pyeclib.cli import version
def main(args=None):
parser = argparse.ArgumentParser(
prog="pyeclib-backend",
description="tool to get various erasure coding information",
)
parser.add_argument(
"-V",
"--version",
action="store_const",
dest="func",
const=version.version_command,
help=version.version_description,
)
subparsers = parser.add_subparsers()
version_parser = subparsers.add_parser(
"version", help=version.version_description)
version_parser.set_defaults(func=version.version_command)
list_parser = subparsers.add_parser(
"list", help=list_cli.list_description)
list_parser.set_defaults(func=list_cli.list_command)
list_cli.add_list_args(list_parser)
check_parser = subparsers.add_parser(
"check", help=check.check_description)
check_parser.set_defaults(func=check.check_command)
check.add_check_args(check_parser)
verify_parser = subparsers.add_parser(
"verify", help=verify.verify_description)
verify_parser.set_defaults(func=verify.verify_command)
verify.add_verify_args(verify_parser)
bench_parser = subparsers.add_parser(
"bench", help=bench.bench_description)
bench_parser.set_defaults(func=bench.bench_command)
bench.add_bench_args(bench_parser)
parsed_args = parser.parse_args(args)
if parsed_args.func is None:
parser.error(
f"the following arguments are required: "
f"{{{','.join(subparsers.choices)}}}"
)
sys.exit(parsed_args.func(parsed_args))
if __name__ == "__main__":
main()

101
pyeclib/cli/bench.py Normal file
View File

@@ -0,0 +1,101 @@
# Copyright (c) 2025, NVIDIA
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# Redistributions of source code must retain the above copyright notice, this
# list of conditions and the following disclaimer.
#
# Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution. THIS SOFTWARE IS
# PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
# OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN
# NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
# THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import argparse
import os
import random
import time
from pyeclib import cli
from pyeclib import ec_iface
def add_bench_args(parser):
parser.add_argument("-e", "--encode", action="store_true")
parser.add_argument("-d", "--decode", action="store_true")
cli.add_instance_args(parser, default_segment_size=2**20)
parser.add_argument("--iterations", "-i", type=int, default=200)
def bench_command(args):
args.ec_type = cli.expand_ec_types(args.ec_type)
data = os.urandom(args.segment_size + args.iterations)
width = max(len(ec_type) for ec_type in args.ec_type)
print(f"Using {args.n_data} data + {args.n_parity} parity with "
f"{args.unavailable} unavailable frags")
for ec_type in args.ec_type:
if ec_type not in ec_iface.ALL_EC_TYPES:
print(f"{ec_type:<{width}} unknown")
continue
if ec_type not in ec_iface.VALID_EC_TYPES:
print(f"{ec_type:<{width}} not available")
continue
try:
instance = ec_iface.ECDriver(
ec_type=ec_type,
k=args.n_data,
m=args.n_parity,
)
except ec_iface.ECDriverError:
print(f"{ec_type:<{width}} could not be instantiated")
continue
frags = instance.encode(data[:args.segment_size])
if args.encode or not args.decode:
start = time.time()
for i in range(args.iterations):
_ = instance.encode(data[i:i + args.segment_size])
dt = time.time() - start
mb_encoded = args.iterations * args.segment_size / (2 ** 20)
print(f'{ec_type} (encode): {mb_encoded / dt:.1f}MB/s')
if args.decode or not args.encode:
start = time.time()
for i in range(args.iterations):
data_frags = random.sample(
frags[:args.n_data],
args.n_data - args.unavailable,
)
if ec_type.startswith('flat_xor'):
# The math is actually more complicated than this, but ...
parity_frags = frags[args.n_data:]
else:
parity_frags = random.sample(
frags[args.n_data:],
args.unavailable,
)
_ = instance.decode(data_frags + parity_frags)
dt = time.time() - start
mb_encoded = args.iterations * args.segment_size / (2 ** 20)
print(f'{ec_type} (decode): {mb_encoded / dt:.1f}MB/s')
bench_description = "benchmark EC schemas"
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=bench_description)
add_bench_args(parser)
args = parser.parse_args()
bench_command(args)

58
pyeclib/cli/check.py Normal file
View File

@@ -0,0 +1,58 @@
# Copyright (c) 2025, NVIDIA
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# Redistributions of source code must retain the above copyright notice, this
# list of conditions and the following disclaimer.
#
# Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution. THIS SOFTWARE IS
# PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
# OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN
# NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
# THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import argparse
import sys
from pyeclib import ec_iface
def add_check_args(parser):
parser.add_argument("-q", "--quiet", action="store_true")
parser.add_argument("ec_type")
def check_command(args):
if args.ec_type in ec_iface.VALID_EC_TYPES:
if not args.quiet:
print(args.ec_type, 'is available')
return 0
if args.ec_type in ec_iface.ALL_EC_TYPES:
if not args.quiet:
print(args.ec_type, 'is missing')
return 1
if not args.quiet:
print(args.ec_type, 'is unknown')
return 2
check_description = "check for backend availability"
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=check_description)
add_check_args(parser)
args = parser.parse_args()
sys.exit(check_command(args))

66
pyeclib/cli/list.py Normal file
View File

@@ -0,0 +1,66 @@
# Copyright (c) 2025, NVIDIA
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# Redistributions of source code must retain the above copyright notice, this
# list of conditions and the following disclaimer.
#
# Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution. THIS SOFTWARE IS
# PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
# OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN
# NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
# THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import argparse
import sys
from pyeclib import cli
from pyeclib import ec_iface
def add_list_args(parser):
parser.add_argument("-a", "--available", action="store_true",
help="display only available backends")
parser.add_argument("ec_type", nargs="*", type=str,
help="display these backends (default: all)")
def list_command(args):
args.ec_type = cli.expand_ec_types(args.ec_type)
width = max(len(backend) for backend in args.ec_type)
found = 0
for backend in args.ec_type:
if args.available:
if backend in ec_iface.VALID_EC_TYPES:
print(backend)
found += 1
else:
if backend not in ec_iface.ALL_EC_TYPES:
print(f"{backend:<{width}} unknown")
elif backend in ec_iface.VALID_EC_TYPES:
print(f"{backend:<{width}} available")
found += 1
else:
print(f"{backend:<{width}} missing")
return 0 if found else 1
list_description = "list availability of EC backends"
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=list_description)
add_list_args(parser)
args = parser.parse_args()
sys.exit(list_command(args))

105
pyeclib/cli/verify.py Normal file
View File

@@ -0,0 +1,105 @@
# Copyright (c) 2025, NVIDIA
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# Redistributions of source code must retain the above copyright notice, this
# list of conditions and the following disclaimer.
#
# Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution. THIS SOFTWARE IS
# PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
# OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN
# NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
# THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import argparse
import itertools
import os
import sys
from pyeclib import cli
from pyeclib import ec_iface
def add_verify_args(parser):
parser.add_argument("-q", "--quiet", action="store_true")
cli.add_instance_args(parser)
def verify_command(args):
args.ec_type = cli.expand_ec_types(args.ec_type)
total_failures = 0
total_corrupt = 0
data = os.urandom(args.segment_size)
width = max(len(ec_type) for ec_type in args.ec_type)
print(f"Using {args.n_data} data + {args.n_parity} parity with "
f"{args.unavailable} unavailable frags")
for ec_type in args.ec_type:
if ec_type not in ec_iface.ALL_EC_TYPES:
print(f"{ec_type:<{width}} unknown")
continue
if ec_type not in ec_iface.VALID_EC_TYPES:
print(f"{ec_type:<{width}} not available")
continue
try:
instance = ec_iface.ECDriver(
ec_type=ec_type,
k=args.n_data,
m=args.n_parity,
)
except ec_iface.ECDriverError:
print(f"{ec_type:<{width}} could not be instantiated")
continue
frags = instance.encode(data)
combinations, failures, corrupt = check_instance(
instance, frags, args.unavailable, data)
total_failures += failures
total_corrupt += corrupt
if corrupt:
print(f"\x1b[91;40m{ec_type:<{width}} {combinations=}, "
f"{failures=}, {corrupt=}\x1b[0m")
elif failures:
print(f"\x1b[1;91m{ec_type:<{width}} {combinations=}, "
f"{failures=}, {corrupt=}\x1b[0m")
else:
print(f"{ec_type:<{width}} {combinations=}")
if total_corrupt:
return 3
if total_failures:
return 1
return 0
def check_instance(instance, frags, unavailable, data):
combinations = corrupt = failures = 0
for to_decode in itertools.combinations(frags, len(frags) - unavailable):
combinations += 1
try:
rebuilt = instance.decode(to_decode)
if rebuilt != data:
corrupt += 1
except ec_iface.ECDriverError:
failures += 1
# Might want to log?
return combinations, failures, corrupt
verify_description = "validate reconstructability of EC schemas"
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=verify_description)
add_verify_args(parser)
args = parser.parse_args()
sys.exit(verify_command(args))

38
pyeclib/cli/version.py Normal file
View File

@@ -0,0 +1,38 @@
# Copyright (c) 2025, NVIDIA
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# Redistributions of source code must retain the above copyright notice, this
# list of conditions and the following disclaimer.
#
# Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution. THIS SOFTWARE IS
# PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
# OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN
# NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
# THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import platform
from pyeclib import ec_iface
version_description = "print pyeclib and liberasurecode versions"
def version_command(args=None):
print(f"pyeclib {ec_iface.__version__}")
print(f"liberasurecode {ec_iface.LIBERASURECODE_VERSION}")
print(f"{platform.python_implementation()} {platform.python_version()}")
if __name__ == "__main__":
version_command()

1
pyeclib/ec_iface.py
View File

@@ -52,6 +52,7 @@ PYECLIB_MINOR = 6
PYECLIB_REV = 4
PYECLIB_VERSION = PyECLibVersion(PYECLIB_MAJOR, PYECLIB_MINOR,
PYECLIB_REV)
__version__ = '%d.%d.%d' % (PYECLIB_MAJOR, PYECLIB_MINOR, PYECLIB_REV)
PYECLIB_MAX_DATA = 32

5
pyproject.toml
View File

@@ -28,13 +28,16 @@ classifiers = [
"License :: OSI Approved :: BSD License",
]
[project.scripts]
pyeclib-backend = "pyeclib.cli.__main__:main"
[project.urls]
Homepage = "https://opendev.org/openstack/pyeclib"
"Bug Tracker" = "https://bugs.launchpad.net/pyeclib"
"Release Notes" = "https://opendev.org/openstack/pyeclib/src/branch/master/ChangeLog"
[tool.setuptools]
packages = ["pyeclib"]
packages = ["pyeclib", "pyeclib.cli"]
platforms = ["Linux"]
[[tool.setuptools.ext-modules]]

229
test/test_pyeclib_cli.py Normal file
View File

@@ -0,0 +1,229 @@
# Copyright (c) 2025, NVIDIA
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# Redistributions of source code must retain the above copyright notice, this
# list of conditions and the following disclaimer.
#
# Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution. THIS SOFTWARE IS
# PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
# OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN
# NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
# THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import io
import platform
import re
import unittest
from unittest import mock
from pyeclib.cli.__main__ import main
from pyeclib import ec_iface
class TestVersion(unittest.TestCase):
def _test_version(self, args):
with mock.patch("sys.stdout", new=io.StringIO()) as stdout, \
self.assertRaises(SystemExit) as caught:
main(args)
self.assertTrue(stdout.getvalue().endswith("\n"))
line_parts = [
line.split(" ")
for line in stdout.getvalue()[:-1].split("\n")
]
self.assertEqual([prog for prog, vers in line_parts], [
'pyeclib',
'liberasurecode',
platform.python_implementation(),
])
re_version = re.compile(r"\d+\.\d+\.\d+")
self.assertEqual(
[bool(re_version.match(vers)) for prog, vers in line_parts],
[True] * 3)
self.assertEqual(caught.exception.code, None)
def test_subcommand(self):
self._test_version(["version"])
def test_opt(self):
self._test_version(["-V"])
class TestList(unittest.TestCase):
def test_list(self):
with mock.patch("sys.stdout", new=io.StringIO()) as stdout, \
self.assertRaises(SystemExit) as caught:
main(["list"])
self.assertTrue(stdout.getvalue().endswith("\n"))
line_parts = [
line.split()
for line in stdout.getvalue()[:-1].split("\n")
]
self.assertEqual(
[backend for backend, status in line_parts],
sorted(ec_iface.ALL_EC_TYPES))
self.assertEqual(
{status for backend, status in line_parts},
{"available", "missing"})
self.assertEqual(caught.exception.code, 0)
def test_list_one(self):
with mock.patch("sys.stdout", new=io.StringIO()) as stdout, \
self.assertRaises(SystemExit) as caught:
main(["list", "liberasurecode_rs_vand"])
self.assertTrue(stdout.getvalue().endswith("\n"))
line_parts = [
line.split()
for line in stdout.getvalue()[:-1].split("\n")
]
self.assertEqual(line_parts, [["liberasurecode_rs_vand", "available"]])
self.assertEqual(caught.exception.code, 0)
def test_list_one_unknown(self):
with mock.patch("sys.stdout", new=io.StringIO()) as stdout, \
self.assertRaises(SystemExit) as caught:
main(["list", "missing-backend"])
self.assertTrue(stdout.getvalue().endswith("\n"))
line_parts = [
line.split()
for line in stdout.getvalue()[:-1].split("\n")
]
self.assertEqual(line_parts, [["missing-backend", "unknown"]])
self.assertEqual(caught.exception.code, 1)
def test_list_multiple_mixed_status(self):
with mock.patch("sys.stdout", new=io.StringIO()) as stdout, \
self.assertRaises(SystemExit) as caught:
main(["list", "missing-backend", "liberasurecode_rs_vand"])
self.assertTrue(stdout.getvalue().endswith("\n"))
line_parts = [
line.split()
for line in stdout.getvalue()[:-1].split("\n")
]
self.assertEqual(line_parts, [
["liberasurecode_rs_vand", "available"],
["missing-backend", "unknown"],
])
self.assertEqual(caught.exception.code, 0)
def test_list_abbreviation(self):
with mock.patch("sys.stdout", new=io.StringIO()) as stdout, \
self.assertRaises(SystemExit) as caught:
main(["list", "isal"])
self.assertTrue(stdout.getvalue().endswith("\n"))
line_parts = [
line.split()
for line in stdout.getvalue()[:-1].split("\n")
]
self.assertEqual(
[backend for backend, status in line_parts],
sorted(backend for backend in ec_iface.ALL_EC_TYPES
if backend.startswith("isa_l_")))
found_status = {status for backend, status in line_parts}
self.assertEqual(len(found_status), 1, found_status)
found_status = list(found_status)[0]
self.assertIn(found_status, {"available", "missing"})
self.assertEqual(caught.exception.code,
0 if found_status == "available" else 1)
def test_list_available(self):
with mock.patch("sys.stdout", new=io.StringIO()) as stdout, \
self.assertRaises(SystemExit) as caught:
main(["list", "--available"])
self.assertTrue(stdout.getvalue().endswith("\n"))
self.assertEqual(
stdout.getvalue()[:-1].split("\n"),
sorted(ec_iface.VALID_EC_TYPES))
self.assertEqual(caught.exception.code, 0)
def test_list_available_abbreviation(self):
with mock.patch("sys.stdout", new=io.StringIO()) as stdout, \
self.assertRaises(SystemExit) as caught:
main(["list", "--available", "flatxor"])
self.assertTrue(stdout.getvalue().endswith("\n"))
self.assertEqual(stdout.getvalue()[:-1].split("\n"), [
"flat_xor_hd_3",
"flat_xor_hd_4",
])
self.assertEqual(caught.exception.code, 0)
class TestCheck(unittest.TestCase):
def test_check_backend_required(self):
with mock.patch("sys.stderr", new=io.StringIO()) as stderr, \
self.assertRaises(SystemExit) as caught:
main(["check"])
self.assertIn("the following arguments are required: ec_type",
stderr.getvalue())
self.assertEqual(caught.exception.code, 2)
def test_check_available(self):
with mock.patch("sys.stdout", new=io.StringIO()) as stdout, \
mock.patch("sys.stderr", new=io.StringIO()) as stderr, \
self.assertRaises(SystemExit) as caught:
main(["check", "liberasurecode_rs_vand"])
self.assertEqual("liberasurecode_rs_vand is available\n",
stdout.getvalue())
self.assertEqual("", stderr.getvalue())
self.assertEqual(caught.exception.code, 0)
def test_check_missing(self):
with mock.patch("sys.stdout", new=io.StringIO()) as stdout, \
mock.patch("sys.stderr", new=io.StringIO()) as stderr, \
mock.patch("pyeclib.ec_iface.VALID_EC_TYPES", []), \
self.assertRaises(SystemExit) as caught:
main(["check", "liberasurecode_rs_vand"])
self.assertEqual("liberasurecode_rs_vand is missing\n",
stdout.getvalue())
self.assertEqual("", stderr.getvalue())
self.assertEqual(caught.exception.code, 1)
def test_check_unknown(self):
with mock.patch("sys.stdout", new=io.StringIO()) as stdout, \
mock.patch("sys.stderr", new=io.StringIO()) as stderr, \
mock.patch("pyeclib.ec_iface.VALID_EC_TYPES", []), \
self.assertRaises(SystemExit) as caught:
main(["check", "unknown-backend"])
self.assertEqual("unknown-backend is unknown\n",
stdout.getvalue())
self.assertEqual("", stderr.getvalue())
self.assertEqual(caught.exception.code, 2)
def test_check_quiet_available(self):
with mock.patch("sys.stdout", new=io.StringIO()) as stdout, \
mock.patch("sys.stderr", new=io.StringIO()) as stderr, \
self.assertRaises(SystemExit) as caught:
main(["check", "-q", "liberasurecode_rs_vand"])
self.assertEqual("", stdout.getvalue())
self.assertEqual("", stderr.getvalue())
self.assertEqual(caught.exception.code, 0)
def test_check_quiet_missing(self):
with mock.patch("sys.stdout", new=io.StringIO()) as stdout, \
mock.patch("sys.stderr", new=io.StringIO()) as stderr, \
mock.patch("pyeclib.ec_iface.VALID_EC_TYPES", []), \
self.assertRaises(SystemExit) as caught:
main(["check", "--quiet", "liberasurecode_rs_vand"])
self.assertEqual("", stdout.getvalue())
self.assertEqual("", stderr.getvalue())
self.assertEqual(caught.exception.code, 1)
def test_check_quiet_unknown(self):
with mock.patch("sys.stdout", new=io.StringIO()) as stdout, \
mock.patch("sys.stderr", new=io.StringIO()) as stderr, \
mock.patch("pyeclib.ec_iface.VALID_EC_TYPES", []), \
self.assertRaises(SystemExit) as caught:
main(["check", "-q", "unknown-backend"])
self.assertEqual("", stdout.getvalue())
self.assertEqual("", stderr.getvalue())
self.assertEqual(caught.exception.code, 2)