Code Cleanup

Also, this allows to pass several times the --resource argument.

Change-Id: I54b055b03b51ce0a3527c140d615024e0d9bd851
Related-bug: #1685630
This commit is contained in:
Jordan Pittier 2017-04-23 18:02:00 +02:00 committed by Jordan Pittier
parent 0b4977a8d1
commit 57481bc598
4 changed files with 48 additions and 69 deletions

View File

@ -63,11 +63,10 @@ def create_argument_parser() -> argparse.ArgumentParser:
"authenticated user."
)
parser.add_argument(
"--resource", choices=["Networks", "Ports", "SecurityGroups",
"FloatingIPs", "Routers", "RouterInterfaces",
"Servers", "Images", "Backups", "Snapshots",
"Volumes"],
help="Name of Resource type to be checked, instead of all resources "
"--resource", action="append",
choices=[cls.__name__ for cls in utils.get_resource_classes()],
help="Purge only the specified resource type. Repeat to delete "
"several types at once."
)
group = parser.add_mutually_exclusive_group(required=True)
@ -179,14 +178,18 @@ def runner(
logging.info("Going to delete %s",
resource_mngr.to_str(resource))
# If we are in dry run mode, don't actually delete the resource
if options.dry_run:
continue
# If we want to delete only specific resources, many things
# can go wrong, so we basically ignore all exceptions.
if options.resource:
utils.call_and_ignore_all(resource_mngr.delete, resource)
exc = shade.OpenStackCloudException
else:
utils.call_and_ignore_notfound(resource_mngr.delete,
resource)
exc = shade.OpenStackCloudResourceNotFound
utils.call_and_ignore_exc(exc, resource_mngr.delete, resource)
except Exception as exc:
log = logging.error
@ -226,17 +229,11 @@ def main() -> None:
creds_manager.ensure_enabled_project()
creds_manager.ensure_role_on_project()
if options.resource:
resource_managers = sorted(
[cls(creds_manager)
for cls in utils.get_resource_classes(options.resource)],
key=operator.methodcaller('order')
)
else:
resource_managers = sorted(
[cls(creds_manager) for cls in utils.get_all_resource_classes()],
key=operator.methodcaller('order')
)
resource_managers = sorted(
[cls(creds_manager)
for cls in utils.get_resource_classes(options.resource)],
key=operator.methodcaller('order')
)
# This is an `Event` used to signal whether one of the threads encountered
# an unrecoverable error, at which point all threads should exit because

View File

@ -61,13 +61,11 @@ class TestFunctions(unittest.TestCase):
self.assertIsInstance(parser, argparse.ArgumentParser)
options = parser.parse_args([
'--resource', 'Networks', '--purge-project', 'foo'
'--resource', 'Networks', '--purge-project', 'foo',
'--resource', 'Volumes'
])
self.assertEqual(False, options.verbose)
self.assertEqual(False, options.dry_run)
self.assertEqual(False, options.delete_shared_resources)
self.assertEqual('foo', options.purge_project)
self.assertEqual('Networks', options.resource)
self.assertEqual(['Networks', 'Volumes'], options.resource)
def test_runner(self):
resources = [mock.Mock(), mock.Mock(), mock.Mock()]

View File

@ -44,7 +44,7 @@ class TestUtils(unittest.TestCase):
})
def test_get_all_resource_classes(self):
classes = utils.get_all_resource_classes()
classes = utils.get_resource_classes()
self.assertIsInstance(classes, typing.List)
for klass in classes:
self.assertTrue(issubclass(klass, ServiceResource))
@ -60,20 +60,15 @@ class TestUtils(unittest.TestCase):
def raiser():
raise shade.exc.OpenStackCloudResourceNotFound("")
self.assertIsNone(utils.call_and_ignore_notfound(raiser))
self.assertIsNone(
utils.call_and_ignore_exc(
shade.exc.OpenStackCloudResourceNotFound, raiser
)
)
m = mock.Mock()
utils.call_and_ignore_notfound(m, 42)
self.assertEqual([mock.call(42)], m.call_args_list)
def test_call_and_ignore_all(self):
def raiser():
raise shade.exc.OpenStackCloudException("")
self.assertIsNone(utils.call_and_ignore_all(raiser))
m = mock.Mock()
utils.call_and_ignore_all(m, 42)
utils.call_and_ignore_exc(
shade.exc.OpenStackCloudResourceNotFound, m, 42)
self.assertEqual([mock.call(42)], m.call_args_list)
@mock.patch('logging.getLogger', autospec=True)

View File

@ -14,24 +14,27 @@ import functools
import importlib
import logging
import pkgutil
import sys
import re
from typing import Any
from typing import Callable
from typing import cast
from typing import Dict
from typing import Iterable
from typing import List
from typing import Optional
from typing import TypeVar
import shade
from ospurge.resources import base
def get_all_resource_classes() -> List:
def get_resource_classes(resources: Optional[Iterable[str]]=None) -> List:
"""
Import all the modules in the `resources` package and return all the
subclasses of the `ServiceResource` Abstract Base Class. This way we can
easily extend OSPurge by just adding a new file in the `resources` dir.
subclasses of the `ServiceResource` ABC that match the `resources` arg.
This way we can easily extend OSPurge by just adding a new file in the
`resources` dir.
"""
iter_modules = pkgutil.iter_modules(
['ospurge/resources'], prefix='ospurge.resources.'
@ -40,23 +43,17 @@ def get_all_resource_classes() -> List:
if not ispkg:
importlib.import_module(name)
return base.ServiceResource.__subclasses__()
all_classes = base.ServiceResource.__subclasses__()
# If we don't want to filter out which classes to return, use a global
# wildcard regex.
if not resources:
regex = re.compile(".*")
# Otherwise, build a regex by concatenation.
else:
regex = re.compile('|'.join(resources))
def get_resource_classes(resources) -> List:
"""
Import the modules by subclass name and return the subclasses
of `ServiceResource` Abstract Base Class.
"""
iter_modules = pkgutil.iter_modules(
['ospurge/resources'], prefix='ospurge.resources.'
)
for (_, name, ispkg) in iter_modules:
if not ispkg:
importlib.import_module(name)
return [s for s in base.ServiceResource.__subclasses__()
if s.__name__ in resources]
return [c for c in all_classes if regex.match(c.__name__)]
F = TypeVar('F', bound=Callable[..., Any])
@ -87,19 +84,11 @@ def monkeypatch_oscc_logging_warning(f: F) -> F:
return cast(F, wrapper)
def call_and_ignore_notfound(f: Callable, *args: List) -> None:
def call_and_ignore_exc(exc: type, f: Callable, *args: List) -> None:
try:
f(*args)
except shade.exc.OpenStackCloudResourceNotFound:
pass
def call_and_ignore_all(f: Callable, *args: List) -> None:
try:
f(*args)
except shade.exc.OpenStackCloudException:
print(sys.exc_info())
pass
except exc as e:
logging.debug("The following exception was ignored: %r", e)
def replace_project_info(config: Dict, new_project_id: str) -> Dict[str, Any]: