util/ctx: extract suppress_oserror()
Extract the `suppress_oserror()` function from the ObjectManager and make it available as utility for other code as well. This also adds a bunch of tests that verify it works as expected.
This commit is contained in:
parent
6e02488a9f
commit
8a195d7502
3 changed files with 80 additions and 15 deletions
|
|
@ -7,7 +7,7 @@ import tempfile
|
|||
import weakref
|
||||
from typing import Optional
|
||||
|
||||
import osbuild.util.rmrf as rmrf
|
||||
from osbuild.util import ctx, rmrf
|
||||
from . import treesum
|
||||
|
||||
|
||||
|
|
@ -16,19 +16,6 @@ __all__ = [
|
|||
]
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def suppress_oserror(*errnos):
|
||||
"""A context manager that suppresses any OSError with an errno in `errnos`.
|
||||
|
||||
Like contextlib.suppress, but can differentiate between OSErrors.
|
||||
"""
|
||||
try:
|
||||
yield
|
||||
except OSError as e:
|
||||
if e.errno not in errnos:
|
||||
raise e
|
||||
|
||||
|
||||
def mount(source, target, bind=True, ro=True, private=True, mode="0755"):
|
||||
options = []
|
||||
if bind:
|
||||
|
|
@ -145,7 +132,7 @@ class Object:
|
|||
self._check_readers()
|
||||
self._check_writer()
|
||||
self.init()
|
||||
with suppress_oserror(errno.ENOTEMPTY, errno.EEXIST):
|
||||
with ctx.suppress_oserror(errno.ENOTEMPTY, errno.EEXIST):
|
||||
os.rename(self._tree, destination)
|
||||
self.reset()
|
||||
|
||||
|
|
|
|||
35
osbuild/util/ctx.py
Normal file
35
osbuild/util/ctx.py
Normal file
|
|
@ -0,0 +1,35 @@
|
|||
"""ContextManager Utilities
|
||||
|
||||
This module implements helpers around python context-managers, with-statements,
|
||||
and RAII. It is meant as a supplement to `contextlib` from the python standard
|
||||
library.
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
|
||||
|
||||
__all__ = [
|
||||
"suppress_oserror",
|
||||
]
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def suppress_oserror(*errnos):
|
||||
"""Suppress OSError Exceptions
|
||||
|
||||
This is an extension to `contextlib.suppress()` from the python standard
|
||||
library. It catches any `OSError` exceptions and suppresses them. However,
|
||||
it only catches the exceptions that match the specified error numbers.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
errnos
|
||||
A list of error numbers to match on. If none are specified, this
|
||||
function has no effect.
|
||||
"""
|
||||
|
||||
try:
|
||||
yield
|
||||
except OSError as e:
|
||||
if e.errno not in errnos:
|
||||
raise e
|
||||
43
test/mod/test_util_ctx.py
Normal file
43
test/mod/test_util_ctx.py
Normal file
|
|
@ -0,0 +1,43 @@
|
|||
#
|
||||
# Tests for the 'osbuild.util.ctx' module.
|
||||
#
|
||||
|
||||
import errno
|
||||
import unittest
|
||||
|
||||
from osbuild.util import ctx
|
||||
|
||||
|
||||
class TestUtilCtx(unittest.TestCase):
|
||||
def test_suppress_oserror(self):
|
||||
#
|
||||
# Verify the `suppress_oserror()` function.
|
||||
#
|
||||
|
||||
# Empty list and empty statement is a no-op.
|
||||
with ctx.suppress_oserror():
|
||||
pass
|
||||
|
||||
# Single errno matches raised errno.
|
||||
with ctx.suppress_oserror(errno.EPERM):
|
||||
raise OSError(errno.EPERM, "Operation not permitted")
|
||||
|
||||
# Many errnos match raised errno regardless of their order.
|
||||
with ctx.suppress_oserror(errno.EPERM, errno.ENOENT, errno.ESRCH):
|
||||
raise OSError(errno.EPERM, "Operation not permitted")
|
||||
with ctx.suppress_oserror(errno.ENOENT, errno.EPERM, errno.ESRCH):
|
||||
raise OSError(errno.EPERM, "Operation not permitted")
|
||||
with ctx.suppress_oserror(errno.ENOENT, errno.ESRCH, errno.EPERM):
|
||||
raise OSError(errno.EPERM, "Operation not permitted")
|
||||
|
||||
# Empty list re-raises exceptions.
|
||||
with self.assertRaises(OSError):
|
||||
with ctx.suppress_oserror():
|
||||
raise OSError(errno.EPERM, "Operation not permitted")
|
||||
|
||||
# Non-matching lists re-raise exceptions.
|
||||
with self.assertRaises(OSError):
|
||||
with ctx.suppress_oserror(errno.ENOENT):
|
||||
raise OSError(errno.EPERM, "Operation not permitted")
|
||||
with ctx.suppress_oserror(errno.ENOENT, errno.ESRCH):
|
||||
raise OSError(errno.EPERM, "Operation not permitted")
|
||||
Loading…
Add table
Add a link
Reference in a new issue