x/tobiko
Code Issues Proposed changes

Pass mypy type verifications

Browse Source 
Change-Id: If5b0a9902bd3cb65068541dce0cfe1ce35b72afb
This commit is contained in:
Federico Ressi 2019-11-02 18:14:37 +01:00
parent f4658cbfb9
commit 3afec15af3
17 changed files with 92 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
tobiko/common/_exception.py
View File

@ -52,11 +52,11 @@ class TobikoException(Exception):
:attribute message: the message to be printed out.
"""
message = None
message = "unknown reason"
def __init__(self, message=None, **properties):
# pylint: disable=exception-message-attribute
message = message or self.message or "unknown reason"
message = message or self.message
if properties:
message = message.format(**properties)
self.message = message

3
tobiko/common/_select.py
View File

@ -54,7 +54,8 @@ class Selection(list):
return '{!s}({!r})'.format(type(self).__name__, list(self))
select = Selection.create
def select(objects):
return Selection.create(objects)
def filter_by_attributes(objects, exclude=False, **attributes):

3
tobiko/common/_skip.py
View File

@ -16,13 +16,14 @@ from __future__ import absolute_import
import functools
import inspect
import unittest
import typing # noqa
import testtools
from tobiko.common import _fixture
SkipException = testtools.TestCase.skipException
SkipException = testtools.TestCase.skipException # type: typing.Type
def skip(reason, *args, **kwargs):

3
tobiko/config.py
View File

@ -17,6 +17,7 @@ import importlib
import itertools
import logging
import os
import typing # noqa
from oslo_config import cfg
from oslo_log import log
@ -82,7 +83,7 @@ class GlobalConfig(object):
# this is a singletone
_instance = None
_sources = {}
_sources = {} # type: typing.Dict[str, typing.Any]
def __new__(cls):
if cls._instance is None:

4
tobiko/http/_session.py
View File

@ -42,10 +42,10 @@ def setup_http_session(session, ssh_client=None):
# All known keyword arguments that could be provided to the pool manager, its
# pools, or the underlying connections. This is used to construct a pool key.
_key_fields = poolmanager._key_fields + ('key_ssh_client',)
_key_fields = tuple(poolmanager._key_fields) + ('key_ssh_client',)
class PoolKey(collections.namedtuple("PoolKey", _key_fields)):
class PoolKey(collections.namedtuple("PoolKey", _key_fields)): # type: ignore
"""The namedtuple class used to construct keys for the connection pool.
All custom key schemes should include the fields in this key at a minimum.

8
tobiko/openstack/glance/_image.py
View File

@ -70,9 +70,9 @@ class GlanceImageStatus(object):
class GlanceImageFixture(tobiko.SharedFixture):
client = None
image_name = None
username = None
password = None
image_name = None # type: str
username = None # type: str
password = None # type: str
image = None
wait_interval = 5.
@ -319,7 +319,7 @@ class FileGlanceImageFixture(UploadGranceImageFixture):
class URLGlanceImageFixture(FileGlanceImageFixture):
image_url = None
image_url = None # type: str
def __init__(self, image_url=None, **kwargs):
super(URLGlanceImageFixture, self).__init__(**kwargs)

1
tobiko/openstack/heat/__init__.py
View File

@ -30,3 +30,4 @@ HeatTemplateFixture = _template.HeatTemplateFixture
HeatTemplateFileFixture = _template.HeatTemplateFileFixture
HeatStackFixture = _stack.HeatStackFixture
heat_stack_parameters = _stack.heat_stack_parameters

25
tobiko/openstack/heat/_stack.py
View File

@ -15,6 +15,7 @@ from __future__ import absolute_import
import collections
import time
import typing # noqa
from heatclient import exc
from oslo_log import log
@ -42,15 +43,17 @@ DELETE_FAILED = 'DELETE_FAILED'
TEMPLATE_FILE_SUFFIX = '.yaml'
def _stack_parameters(obj, stack=None):
if obj is None or isinstance(obj, collections.Mapping):
def heat_stack_parameters(obj, stack=None):
if isinstance(obj, HeatStackParametersFixture):
parameters = obj
elif obj is None or isinstance(obj, collections.Mapping):
parameters = HeatStackParametersFixture(stack, obj)
else:
parameters = tobiko.get_fixture(obj)
if not isinstance(parameters, HeatStackParametersFixture):
msg = "Object {!r} is not an HeatStackParametersFixture".format(
parameters)
raise TypeError(msg)
tobiko.check_valid_type(parameters, HeatStackParametersFixture)
if stack:
parameters.stack = parameters.stack or stack
tobiko.check_valid_type(parameters.stack, type(None), HeatStackFixture)
return parameters
@ -60,10 +63,10 @@ class HeatStackFixture(tobiko.SharedFixture):
client = None
retry_create_stack = 1
wait_interval = 5
stack_name = None
template = None
stack_name = None # type: str
template = None # type: _template.HeatTemplateFixture
stack = None
parameters = None
parameters = None # type: HeatStackParametersFixture
def __init__(self, stack_name=None, template=None, parameters=None,
wait_interval=None, client=None):
@ -73,7 +76,7 @@ class HeatStackFixture(tobiko.SharedFixture):
self.fixture_name)
self.template = _template.heat_template(template or self.template)
self.parameters = _stack_parameters(
self.parameters = heat_stack_parameters(
stack=self, obj=(parameters or self.parameters))
self.client = client or self.client
if config.get_bool_env('TOBIKO_PREVENT_CREATE'):
@ -282,7 +285,7 @@ class HeatStackNamespaceFixture(tobiko.SharedFixture):
def __init__(self, stack):
super(HeatStackNamespaceFixture, self).__init__()
if not isinstance(stack, HeatStackFixture):
if stack and not isinstance(stack, HeatStackFixture):
message = "Object {!r} is not an HeatStackFixture".format(stack)
raise TypeError(message)
self.stack = stack

5
tobiko/openstack/heat/_template.py
View File

@ -16,6 +16,7 @@ from __future__ import absolute_import
import collections
import os
import sys
import typing # noqa
from heatclient.common import template_utils
import yaml
@ -30,7 +31,7 @@ TEMPLATE_DIRS = list(sys.path)
class HeatTemplateFixture(tobiko.SharedFixture):
template = None
template = None # type: typing.Dict[str, typing.Any]
template_files = None
template_yaml = None
@ -68,8 +69,6 @@ class HeatTemplateFileFixture(HeatTemplateFixture):
template_file = None
template_dirs = None
template_files = None
template = None
template_yaml = None
def __init__(self, template_file=None, template_dirs=None):
super(HeatTemplateFileFixture, self).__init__()

32
tobiko/openstack/keystone/_clouds_file.py
View File

@ -30,10 +30,8 @@ JSON_SUFFIXES = ('.json',)
CLOUDS_FILE_SUFFIXES = JSON_SUFFIXES + YAML_SUFFIXES
try:
FileNotFound = FileNotFoundError
except NameError:
FileNotFound = OSError
class CloudsFileNotFoundError(tobiko.TobikoException):
message = "No such clouds file(s): {clouds_files!s}"
class DefaultCloudsFileConfig(tobiko.SharedFixture):
@ -198,22 +196,18 @@ class CloudsFileKeystoneCredentialsFixture(
def _get_clouds_file(self):
clouds_file = self.clouds_file
if not clouds_file:
clouds_files = self.clouds_files
for filename in clouds_files:
if os.path.exists(filename):
LOG.debug('Found clouds file at %r', filename)
self.clouds_file = clouds_file = filename
break
else:
message = 'No such clouds file: {!s}'.format(
', '.join(repr(f) for f in clouds_files))
raise FileNotFound(message)
if not os.path.exists(clouds_file):
message = 'Cloud file not found: {!r}'.format(clouds_file)
raise FileNotFound(message)
if clouds_file:
clouds_files = [self.clouds_file]
else:
clouds_files = list(self.clouds_files)
for filename in clouds_files:
if os.path.exists(filename):
LOG.debug('Found clouds file at %r', filename)
self.clouds_file = clouds_file = filename
break
else:
raise CloudsFileNotFoundError(clouds_files=', '.join(clouds_files))
return clouds_file

3
tobiko/openstack/stacks/_nova.py
View File

@ -16,6 +16,7 @@
from __future__ import absolute_import
import os
import typing # noqa
import six
@ -118,7 +119,7 @@ class ServerStackFixture(heat.HeatStackFixture):
port_security_enabled = False
#: Security groups to be associated to network ports
security_groups = []
security_groups = [] # type: typing.List[str]
@property
def key_name(self):

35
tobiko/shell/ping/_parameters.py
View File

@ -24,9 +24,27 @@ from oslo_log import log
LOG = log.getLogger(__name__)
PING_PARAMETER_NAMES = ['host', 'count', 'deadline', 'fragmentation',
'interval', 'ip_version', 'packet_size', 'source',
'timeout', 'network_namespace']
class PingParameters(collections.namedtuple('PingParameters',
['host',
'count',
'deadline',
'fragmentation',
'interval',
'ip_version',
'packet_size',
'source',
'timeout',
'network_namespace'])):
"""Recollect parameters to be used to format ping command line
PingParameters class is a data model recollecting parameters used to
create a ping command line. It provides the feature of copying default
values from another instance of PingParameters passed using constructor
parameter 'default'.
"""
PING_PARAMETERS_NAMES = PingParameters._fields
def get_ping_parameters(default=None, **ping_params):
@ -122,17 +140,6 @@ def default_ping_parameters():
timeout=CONF.tobiko.ping.timeout)
class PingParameters(collections.namedtuple('PingParameters',
PING_PARAMETER_NAMES)):
"""Recollect parameters to be used to format ping command line
PingParameters class is a data model recollecting parameters used to
create a ping command line. It provides the feature of copying default
values from another instance of PingParameters passed using constructor
parameter 'default'.
"""
def get_ping_ip_version(parameters):
ip_version = parameters.ip_version
if ip_version is not None:

15
tobiko/shell/sh/_path.py
View File

@ -15,8 +15,6 @@
# under the License.
from __future__ import absolute_import
import collections
from oslo_log import log
import tobiko
@ -27,18 +25,11 @@ LOG = log.getLogger(__name__)
class ExecutePathFixture(tobiko.SharedFixture):
executable_dirs = None
environ = None
def __init__(self, executable_dirs=None, environ=None):
super(ExecutePathFixture, self).__init__()
if executable_dirs:
self.executable_dirs = tuple(executable_dirs)
tobiko.check_valid_type(self.executable_dirs, collections.Iterable)
if environ is not None:
self.environ = environ
tobiko.check_valid_type(self.environ, collections.Mapping)
self.executable_dirs = list(executable_dirs or
[]) # type: typing.List[str]
self.environ = dict(environ or {}) # type: typing.Dict[str, str]
def setup_fixture(self):
missing_dirs = []

8
tobiko/shell/ssh/_forward.py
View File

@ -138,11 +138,7 @@ class SSHTunnelForwarderFixture(tobiko.SharedFixture):
# pylint: disable=protected-access
SSHForwardHandler = sshtunnel._ForwardHandler
# pylint: enable=protected-access
class SSHUnixForwardHandler(SSHForwardHandler):
class SSHUnixForwardHandler(sshtunnel._ForwardHandler):
transport = None
@ -178,6 +174,8 @@ class SSHUnixForwardHandler(SSHForwardHandler):
self.logger.log(sshtunnel.TRACE_LEVEL,
'{0} connection closed.'.format(self.info))
# pylint: enable=protected-access
class SSHTunnelForwarder(sshtunnel.SSHTunnelForwarder):

14
tobiko/tests/unit/openstack/heat/test_stack.py
View File

@ -29,7 +29,7 @@ from tobiko.tests.unit import openstack
class MyStack(heat.HeatStackFixture):
template = {'template': 'from-class'}
template = heat.heat_template({'template': 'from-class'})
class MyStackWithStackName(MyStack):
@ -37,7 +37,7 @@ class MyStackWithStackName(MyStack):
class MyStackWithParameters(MyStack):
parameters = {'param': 'from-class'}
parameters = heat.heat_stack_parameters({'param': 'from-class'})
class MyStackWithWaitInterval(MyStack):
@ -73,7 +73,8 @@ class HeatStackFixtureTest(openstack.OpenstackTest):
self.assertIsInstance(stack.parameters,
_stack.HeatStackParametersFixture)
self.assertEqual(parameters or fixture_class.parameters or {},
self.assertEqual(parameters or getattr(fixture_class.parameters,
'parameters', {}),
stack.parameters.parameters)
self.assertEqual(wait_interval or fixture_class.wait_interval,
stack.wait_interval)
@ -95,7 +96,7 @@ class HeatStackFixtureTest(openstack.OpenstackTest):
self.test_init(parameters={'my': 'value'})
def test_init_with_parameters_from_class(self):
self.test_init(fixture_class=MyStackWithParameters, )
self.test_init(fixture_class=MyStackWithParameters)
def test_init_with_wait_interval(self):
self.test_init(wait_interval=20)
@ -144,7 +145,10 @@ class HeatStackFixtureTest(openstack.OpenstackTest):
else:
client.stacks.delete.assert_not_called()
parameters = parameters or fixture_class.parameters or {}
parameters = (parameters or
(fixture_class.parameters and
fixture_class.parameters.values) or
{})
self.assertEqual(parameters, stack.parameters.values)
if call_create:
client.stacks.create.assert_called_once_with(

19
tobiko/tests/unit/openstack/keystone/test_clouds_file.py
View File

@ -17,6 +17,7 @@ from __future__ import absolute_import
import json
import os
import tempfile
import typing # noqa
import yaml
@ -38,9 +39,9 @@ def make_clouds_content(cloud_name, api_version=None, auth=None):
class CloudsFileFixture(tobiko.SharedFixture):
cloud_name = None
api_version = None
auth = None
cloud_name = None # type: str
api_version = None # type: str
auth = None # type: typing.Dict[str, typing.Any]
clouds_content = None
clouds_file = None
suffix = '.yaml'
@ -298,15 +299,15 @@ class CloudsFileKeystoneCredentialsFixtureTest(openstack.OpenstackTest):
self.patch(self.config, 'clouds_files', ['/a', '/b', '/c'])
fixture = keystone.CloudsFileKeystoneCredentialsFixture(
cloud_name='cloud-name')
ex = self.assertRaises(_clouds_file.FileNotFound, tobiko.setup_fixture,
fixture)
ex = self.assertRaises(_clouds_file.CloudsFileNotFoundError,
tobiko.setup_fixture, fixture)
self.assertEqual('cloud-name', fixture.cloud_name)
self.assertEqual("No such clouds file: '/a', '/b', '/c'", str(ex))
self.assertEqual("No such clouds file(s): /a, /b, /c", str(ex))
def test_setup_with_non_existing_clouds_file(self):
fixture = keystone.CloudsFileKeystoneCredentialsFixture(
clouds_file='/a.yaml',
cloud_name='cloud-name')
ex = self.assertRaises(_clouds_file.FileNotFound, tobiko.setup_fixture,
fixture)
self.assertEqual("Cloud file not found: '/a.yaml'", str(ex))
ex = self.assertRaises(_clouds_file.CloudsFileNotFoundError,
tobiko.setup_fixture, fixture)
self.assertEqual("No such clouds file(s): /a.yaml", str(ex))

1
zuul.d/mypy.yaml
View File

@ -1,7 +1,6 @@
- job:
name: tobiko-tox-mypy
parent: openstack-tox
voting: False
description: |
Run static mypy type checker for Tobiko project.
vars: