openstack
/
octavia
Code Issues Proposed changes

Add health monitor API validation 

The WSME package is not validating null/None fields correctly,
especially for Enum fields.
This patch strengthens the API validaton for health monitors.

It also removes an un-used status update method in the API
controllers.

Change-Id: I892dd81b166e9ab1a26bb80e8b0da1cb2e5c7482
This commit is contained in:
Michael Johnson 2018-06-07 21:15:00 -07:00
parent 13eab152fb
commit e1e1640af5
6 changed files with 306 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
octavia/api/v2/controllers/health_monitor.py
View File

@ -27,7 +27,7 @@ from octavia.api.drivers import driver_factory
from octavia.api.drivers import utils as driver_utils
from octavia.api.v2.controllers import base
from octavia.api.v2.types import health_monitor as hm_types
from octavia.common import constants
from octavia.common import constants as consts
from octavia.common import data_models
from octavia.common import exceptions
from octavia.db import api as db_api
@ -39,7 +39,7 @@ LOG = logging.getLogger(__name__)
class HealthMonitorController(base.BaseController):
RBAC_TYPE = constants.RBAC_HEALTHMONITOR
RBAC_TYPE = consts.RBAC_HEALTHMONITOR
def __init__(self):
super(HealthMonitorController, self).__init__()
@ -52,7 +52,7 @@ class HealthMonitorController(base.BaseController):
db_hm = self._get_db_hm(context.session, id, show_deleted=False)
self._auth_validate_action(context, db_hm.project_id,
constants.RBAC_GET_ONE)
consts.RBAC_GET_ONE)
result = self._convert_db_to_type(
db_hm, hm_types.HealthMonitorResponse)
@ -71,7 +71,7 @@ class HealthMonitorController(base.BaseController):
db_hm, links = self.repositories.health_monitor.get_all(
context.session, show_deleted=False,
pagination_helper=pcontext.get(constants.PAGINATION_HELPER),
pagination_helper=pcontext.get(consts.PAGINATION_HELPER),
**query_filter)
result = self._convert_db_to_type(
db_hm, [hm_types.HealthMonitorResponse])
@ -94,7 +94,7 @@ class HealthMonitorController(base.BaseController):
load_balancer_id = pool.load_balancer_id
if not self.repositories.test_and_set_lb_and_listeners_prov_status(
session, load_balancer_id,
constants.PENDING_UPDATE, constants.PENDING_UPDATE,
consts.PENDING_UPDATE, consts.PENDING_UPDATE,
listener_ids=self._get_affected_listener_ids(session, hm),
pool_id=hm.pool_id):
LOG.info("Health Monitor cannot be created or modified because "
@ -102,23 +102,42 @@ class HealthMonitorController(base.BaseController):
raise exceptions.ImmutableObject(resource='Load Balancer',
id=load_balancer_id)
def _reset_lb_listener_pool_statuses(self, session, hm):
# Setting LB + listener + pool back to active because this should be a
# recoverable error
pool = self._get_db_pool(session, hm.pool_id)
load_balancer_id = pool.load_balancer_id
self.repositories.load_balancer.update(
session, load_balancer_id,
provisioning_status=constants.ACTIVE)
for listener in self._get_affected_listener_ids(session, hm):
self.repositories.listener.update(
session, listener,
provisioning_status=constants.ACTIVE)
self.repositories.pool.update(session, hm.pool_id,
provisioning_status=constants.ACTIVE)
def _validate_create_hm(self, lock_session, hm_dict):
"""Validate creating health monitor on pool."""
mandatory_fields = (consts.TYPE, consts.DELAY, consts.TIMEOUT,
consts.POOL_ID)
for field in mandatory_fields:
if hm_dict.get(field, None) is None:
raise exceptions.InvalidOption(value='None', option=field)
# MAX_RETRIES is renamed fall_threshold so handle is special
if hm_dict.get(consts.RISE_THRESHOLD, None) is None:
raise exceptions.InvalidOption(value='None',
option=consts.MAX_RETRIES)
if hm_dict[consts.TYPE] != consts.HEALTH_MONITOR_HTTP:
if hm_dict.get(consts.HTTP_METHOD, None):
raise exceptions.InvalidOption(
value=consts.HTTP_METHOD, option='health monitors of '
'type {}'.format(hm_dict[consts.TYPE]))
if hm_dict.get(consts.URL_PATH, None):
raise exceptions.InvalidOption(
value=consts.URL_PATH, option='health monitors of '
'type {}'.format(hm_dict[consts.TYPE]))
if hm_dict.get(consts.EXPECTED_CODES, None):
raise exceptions.InvalidOption(
value=consts.EXPECTED_CODES, option='health monitors of '
'type {}'.format(hm_dict[consts.TYPE]))
else:
if not hm_dict.get(consts.HTTP_METHOD, None):
hm_dict[consts.HTTP_METHOD] = (
consts.HEALTH_MONITOR_HTTP_DEFAULT_METHOD)
if not hm_dict.get(consts.URL_PATH, None):
hm_dict[consts.URL_PATH] = (
consts.HEALTH_MONITOR_DEFAULT_URL_PATH)
if not hm_dict.get(consts.EXPECTED_CODES, None):
hm_dict[consts.EXPECTED_CODES] = (
consts.HEALTH_MONITOR_DEFAULT_EXPECTED_CODES)
try:
return self.repositories.health_monitor.create(
lock_session, **hm_dict)
@ -138,9 +157,9 @@ class HealthMonitorController(base.BaseController):
health_monitor = health_monitor_.healthmonitor
if (not CONF.api_settings.allow_ping_health_monitors and
health_monitor.type == constants.HEALTH_MONITOR_PING):
health_monitor.type == consts.HEALTH_MONITOR_PING):
raise exceptions.DisabledOption(
option='type', value=constants.HEALTH_MONITOR_PING)
option='type', value=consts.HEALTH_MONITOR_PING)
pool = self._get_db_pool(context.session, health_monitor.pool_id)
@ -148,7 +167,7 @@ class HealthMonitorController(base.BaseController):
context.session, pool.load_balancer_id)
self._auth_validate_action(context, health_monitor.project_id,
constants.RBAC_POST)
consts.RBAC_POST)
# Load the driver early as it also provides validation
driver = driver_factory.get_driver(provider)
@ -198,6 +217,29 @@ class HealthMonitorController(base.BaseController):
return db_hm
def _validate_update_hm(self, db_hm, health_monitor):
if db_hm.type != consts.HEALTH_MONITOR_HTTP:
if health_monitor.http_method != wtypes.Unset:
raise exceptions.InvalidOption(
value=consts.HTTP_METHOD, option='health monitors of '
'type {}'.format(db_hm.type))
if health_monitor.url_path != wtypes.Unset:
raise exceptions.InvalidOption(
value=consts.URL_PATH, option='health monitors of '
'type {}'.format(db_hm.type))
if health_monitor.expected_codes != wtypes.Unset:
raise exceptions.InvalidOption(
value=consts.URL_PATH, option='health monitors of '
'type {}'.format(db_hm.type))
else:
# For HTTP health monitor these cannot be null/None
if health_monitor.http_method is None:
health_monitor.http_method = wtypes.Unset
if health_monitor.url_path is None:
health_monitor.url_path = wtypes.Unset
if health_monitor.expected_codes is None:
health_monitor.expected_codes = wtypes.Unset
@wsme_pecan.wsexpose(hm_types.HealthMonitorRootResponse, wtypes.text,
body=hm_types.HealthMonitorRootPUT, status_code=200)
def put(self, id, health_monitor_):
@ -210,7 +252,9 @@ class HealthMonitorController(base.BaseController):
project_id, provider = self._get_lb_project_id_provider(
context.session, pool.load_balancer_id)
self._auth_validate_action(context, project_id, constants.RBAC_PUT)
self._auth_validate_action(context, project_id, consts.RBAC_PUT)
self._validate_update_hm(db_hm, health_monitor)
# Load the driver early as it also provides validation
driver = driver_factory.get_driver(provider)
@ -233,7 +277,7 @@ class HealthMonitorController(base.BaseController):
driver_dm.HealthMonitor.from_dict(provider_healthmon_dict))
# Update the database to reflect what the driver just accepted
health_monitor.provisioning_status = constants.PENDING_UPDATE
health_monitor.provisioning_status = consts.PENDING_UPDATE
db_hm_dict = health_monitor.to_dict(render_unsets=False)
self.repositories.health_monitor.update(lock_session, id,
**db_hm_dict)
@ -256,9 +300,9 @@ class HealthMonitorController(base.BaseController):
project_id, provider = self._get_lb_project_id_provider(
context.session, pool.load_balancer_id)
self._auth_validate_action(context, project_id, constants.RBAC_DELETE)
self._auth_validate_action(context, project_id, consts.RBAC_DELETE)
if db_hm.provisioning_status == constants.DELETED:
if db_hm.provisioning_status == consts.DELETED:
return
# Load the driver early as it also provides validation
@ -270,7 +314,7 @@ class HealthMonitorController(base.BaseController):
self.repositories.health_monitor.update(
lock_session, db_hm.id,
provisioning_status=constants.PENDING_DELETE)
provisioning_status=consts.PENDING_DELETE)
LOG.info("Sending delete Health Monitor %s to provider %s",
id, driver.name)

15
octavia/api/v2/controllers/member.py
View File

@ -110,21 +110,6 @@ class MemberController(base.BaseController):
raise exceptions.ImmutableObject(resource='Load Balancer',
id=load_balancer_id)
def _reset_lb_listener_pool_statuses(self, session, member=None):
# Setting LB + listener + pool back to active because this should be a
# recoverable error
pool = self._get_db_pool(session, self.pool_id)
load_balancer_id = pool.load_balancer_id
self.repositories.load_balancer.update(
session, load_balancer_id,
provisioning_status=constants.ACTIVE)
for listener in self._get_affected_listener_ids(session, member):
self.repositories.listener.update(
session, listener,
provisioning_status=constants.ACTIVE)
self.repositories.pool.update(session, self.pool_id,
provisioning_status=constants.ACTIVE)
def _validate_create_member(self, lock_session, member_dict):
"""Validate creating member on pool."""
try:

19
octavia/api/v2/types/health_monitor.py
View File

@ -89,14 +89,11 @@ class HealthMonitorPOST(BaseHealthMonitorType):
maximum=constants.MAX_HM_RETRIES),
mandatory=True)
http_method = wtypes.wsattr(
wtypes.Enum(str, *constants.SUPPORTED_HEALTH_MONITOR_HTTP_METHODS),
default=constants.HEALTH_MONITOR_HTTP_DEFAULT_METHOD)
wtypes.Enum(str, *constants.SUPPORTED_HEALTH_MONITOR_HTTP_METHODS))
url_path = wtypes.wsattr(
types.URLPathType(),
default=constants.HEALTH_MONITOR_DEFAULT_URL_PATH)
types.URLPathType())
expected_codes = wtypes.wsattr(
wtypes.StringType(pattern=r'^(\d{3}(\s*,\s*\d{3})*)$|^(\d{3}-\d{3})$'),
default=constants.HEALTH_MONITOR_DEFAULT_EXPECTED_CODES)
wtypes.StringType(pattern=r'^(\d{3}(\s*,\s*\d{3})*)$|^(\d{3}-\d{3})$'))
admin_state_up = wtypes.wsattr(bool, default=True)
# TODO(johnsom) Remove after deprecation (R series)
project_id = wtypes.wsattr(wtypes.StringType(max_length=36))
@ -146,14 +143,10 @@ class HealthMonitorSingleCreate(BaseHealthMonitorType):
maximum=constants.MAX_HM_RETRIES),
mandatory=True)
http_method = wtypes.wsattr(
wtypes.Enum(str, *constants.SUPPORTED_HEALTH_MONITOR_HTTP_METHODS),
default=constants.HEALTH_MONITOR_HTTP_DEFAULT_METHOD)
url_path = wtypes.wsattr(
types.URLPathType(),
default=constants.HEALTH_MONITOR_DEFAULT_URL_PATH)
wtypes.Enum(str, *constants.SUPPORTED_HEALTH_MONITOR_HTTP_METHODS))
url_path = wtypes.wsattr(types.URLPathType())
expected_codes = wtypes.wsattr(
wtypes.StringType(pattern=r'^(\d{3}(\s*,\s*\d{3})*)$|^(\d{3}-\d{3})$'),
default=constants.HEALTH_MONITOR_DEFAULT_EXPECTED_CODES)
wtypes.StringType(pattern=r'^(\d{3}(\s*,\s*\d{3})*)$|^(\d{3}-\d{3})$'))
admin_state_up = wtypes.wsattr(bool, default=True)

8
octavia/common/constants.py
View File

@ -52,6 +52,14 @@ SUPPORTED_HEALTH_MONITOR_HTTP_METHODS = (
HEALTH_MONITOR_HTTP_METHOD_PATCH)
HEALTH_MONITOR_DEFAULT_EXPECTED_CODES = '200'
HEALTH_MONITOR_DEFAULT_URL_PATH = '/'
TYPE = 'type'
URL_PATH = 'url_path'
HTTP_METHOD = 'http_method'
EXPECTED_CODES = 'expected_codes'
DELAY = 'delay'
TIMEOUT = 'timeout'
MAX_RETRIES = 'max_retries'
RISE_THRESHOLD = 'rise_threshold'
UPDATE_STATS = 'UPDATE_STATS'
UPDATE_HEALTH = 'UPDATE_HEALTH'

215
octavia/tests/functional/api/v2/test_health_monitor.py
View File

@ -640,6 +640,22 @@ class TestHealthMonitor(base.BaseAPITest):
self.assertEqual('/', api_hm.get('url_path'))
self.assertEqual('200', api_hm.get('expected_codes'))
def test_create_http_full(self):
api_hm = self.create_health_monitor(
self.pool_id, constants.HEALTH_MONITOR_HTTP,
1, 1, 1, 1, admin_state_up=False, expected_codes='200',
http_method='GET', name='Test HM', url_path='/').get(self.root_tag)
self.assertEqual(constants.HEALTH_MONITOR_HTTP, api_hm.get('type'))
self.assertEqual(1, api_hm.get('delay'))
self.assertEqual(1, api_hm.get('timeout'))
self.assertEqual(1, api_hm.get('max_retries_down'))
self.assertEqual(1, api_hm.get('max_retries'))
self.assertFalse(api_hm.get('admin_state_up'))
self.assertEqual('Test HM', api_hm.get('name'))
self.assertEqual('GET', api_hm.get('http_method'))
self.assertEqual('/', api_hm.get('url_path'))
self.assertEqual('200', api_hm.get('expected_codes'))
def test_create_authorized(self):
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
@ -793,6 +809,110 @@ class TestHealthMonitor(base.BaseAPITest):
self.assertIn('Provider \'bad_driver\' reports error: broken',
response.json.get('faultstring'))
def test_create_with_type_none(self):
req_dict = {'pool_id': self.pool_id,
'type': None,
'delay': 1,
'timeout': 1,
'max_retries_down': 1,
'max_retries': 1,
'url_path': '/'}
self.post(self.HMS_PATH, self._build_body(req_dict), status=400)
self.assert_correct_status(
lb_id=self.lb_id, listener_id=self.listener_id,
pool_id=self.pool_id)
def test_create_with_delay_none(self):
req_dict = {'pool_id': self.pool_id,
'type': constants.HEALTH_MONITOR_HTTP,
'delay': None,
'timeout': 1,
'max_retries_down': 1,
'max_retries': 1,
'url_path': '/'}
self.post(self.HMS_PATH, self._build_body(req_dict), status=400)
self.assert_correct_status(
lb_id=self.lb_id, listener_id=self.listener_id,
pool_id=self.pool_id)
def test_create_with_max_retries_none(self):
req_dict = {'pool_id': self.pool_id,
'type': constants.HEALTH_MONITOR_HTTP,
'delay': 1,
'timeout': 1,
'max_retries_down': 1,
'max_retries': None,
'url_path': '/'}
self.post(self.HMS_PATH, self._build_body(req_dict), status=400)
self.assert_correct_status(
lb_id=self.lb_id, listener_id=self.listener_id,
pool_id=self.pool_id)
def test_create_with_timeout_none(self):
req_dict = {'pool_id': self.pool_id,
'type': constants.HEALTH_MONITOR_HTTP,
'delay': 1,
'timeout': None,
'max_retries_down': 1,
'max_retries': 1,
'url_path': '/'}
self.post(self.HMS_PATH, self._build_body(req_dict), status=400)
self.assert_correct_status(
lb_id=self.lb_id, listener_id=self.listener_id,
pool_id=self.pool_id)
def test_create_with_pool_id_none(self):
req_dict = {'pool_id': None,
'type': constants.HEALTH_MONITOR_HTTP,
'delay': 1,
'timeout': 1,
'max_retries_down': 1,
'max_retries': 1,
'url_path': '/'}
self.post(self.HMS_PATH, self._build_body(req_dict), status=404)
self.assert_correct_status(
lb_id=self.lb_id, listener_id=self.listener_id,
pool_id=self.pool_id)
def test_create_TCP_with_http_method(self):
req_dict = {'pool_id': self.pool_id,
'type': constants.HEALTH_MONITOR_TCP,
'delay': 1,
'timeout': 1,
'max_retries_down': 1,
'max_retries': 1,
'http_method': constants.HEALTH_MONITOR_HTTP_METHOD_GET}
self.post(self.HMS_PATH, self._build_body(req_dict), status=400)
self.assert_correct_status(
lb_id=self.lb_id, listener_id=self.listener_id,
pool_id=self.pool_id)
def test_create_TCP_with_url_path(self):
req_dict = {'pool_id': self.pool_id,
'type': constants.HEALTH_MONITOR_TCP,
'delay': 1,
'timeout': 1,
'max_retries_down': 1,
'max_retries': 1,
'url_path': '/'}
self.post(self.HMS_PATH, self._build_body(req_dict), status=400)
self.assert_correct_status(
lb_id=self.lb_id, listener_id=self.listener_id,
pool_id=self.pool_id)
def test_create_TCP_with_expected_codes(self):
req_dict = {'pool_id': self.pool_id,
'type': constants.HEALTH_MONITOR_TCP,
'delay': 1,
'timeout': 1,
'max_retries_down': 1,
'max_retries': 1,
'expected_codes': '200'}
self.post(self.HMS_PATH, self._build_body(req_dict), status=400)
self.assert_correct_status(
lb_id=self.lb_id, listener_id=self.listener_id,
pool_id=self.pool_id)
def test_duplicate_create(self):
self.create_health_monitor(
self.pool_id, constants.HEALTH_MONITOR_HTTP, 1, 1, 1, 1)
@ -827,6 +947,29 @@ class TestHealthMonitor(base.BaseAPITest):
listener_prov_status=constants.PENDING_UPDATE,
pool_prov_status=constants.PENDING_UPDATE,
hm_prov_status=constants.PENDING_UPDATE)
response = self.get(self.HM_PATH.format(
healthmonitor_id=api_hm.get('id'))).json.get(self.root_tag)
self.assertEqual(2, response[constants.MAX_RETRIES])
def test_update_TCP(self):
api_hm = self.create_health_monitor(
self.pool_with_listener_id,
constants.HEALTH_MONITOR_TCP, 1, 1, 1, 1).get(self.root_tag)
self.set_lb_status(self.lb_id)
new_hm = {'max_retries': 2}
self.put(
self.HM_PATH.format(healthmonitor_id=api_hm.get('id')),
self._build_body(new_hm))
self.assert_correct_status(
lb_id=self.lb_id, listener_id=self.listener_id,
pool_id=self.pool_with_listener_id, hm_id=api_hm.get('id'),
lb_prov_status=constants.PENDING_UPDATE,
listener_prov_status=constants.PENDING_UPDATE,
pool_prov_status=constants.PENDING_UPDATE,
hm_prov_status=constants.PENDING_UPDATE)
response = self.get(self.HM_PATH.format(
healthmonitor_id=api_hm.get('id'))).json.get(self.root_tag)
self.assertEqual(2, response[constants.MAX_RETRIES])
def test_update_authorized(self):
api_hm = self.create_health_monitor(
@ -918,6 +1061,78 @@ class TestHealthMonitor(base.BaseAPITest):
self.assertIn('Provider \'bad_driver\' reports error: broken',
response.json.get('faultstring'))
def test_update_TCP_setting_http_method(self):
api_hm = self.create_health_monitor(
self.pool_with_listener_id,
constants.HEALTH_MONITOR_TCP, 1, 1, 1, 1).get(self.root_tag)
self.set_lb_status(self.lb_id)
new_hm = {'http_method': constants.HEALTH_MONITOR_HTTP_METHOD_GET}
self.put(
self.HM_PATH.format(healthmonitor_id=api_hm.get('id')),
self._build_body(new_hm), status=400)
def test_update_TCP_setting_url_path(self):
api_hm = self.create_health_monitor(
self.pool_with_listener_id,
constants.HEALTH_MONITOR_TCP, 1, 1, 1, 1).get(self.root_tag)
self.set_lb_status(self.lb_id)
new_hm = {'url_path': '/'}
self.put(
self.HM_PATH.format(healthmonitor_id=api_hm.get('id')),
self._build_body(new_hm), status=400)
def test_update_TCP_setting_expected_codes(self):
api_hm = self.create_health_monitor(
self.pool_with_listener_id,
constants.HEALTH_MONITOR_TCP, 1, 1, 1, 1).get(self.root_tag)
self.set_lb_status(self.lb_id)
new_hm = {'expected_codes': '200'}
self.put(
self.HM_PATH.format(healthmonitor_id=api_hm.get('id')),
self._build_body(new_hm), status=400)
def test_update_HTTP_http_method_none(self):
api_hm = self.create_health_monitor(
self.pool_with_listener_id,
constants.HEALTH_MONITOR_HTTP, 1, 1, 1, 1).get(self.root_tag)
self.set_lb_status(self.lb_id)
new_hm = {'http_method': None}
self.put(
self.HM_PATH.format(healthmonitor_id=api_hm.get('id')),
self._build_body(new_hm))
response = self.get(self.HM_PATH.format(
healthmonitor_id=api_hm.get('id'))).json.get(self.root_tag)
self.assertEqual(constants.HEALTH_MONITOR_HTTP_METHOD_GET,
response['http_method'])
def test_update_HTTP_url_path_none(self):
api_hm = self.create_health_monitor(
self.pool_with_listener_id,
constants.HEALTH_MONITOR_HTTP, 1, 1, 1, 1).get(self.root_tag)
self.set_lb_status(self.lb_id)
new_hm = {'url_path': None}
self.put(
self.HM_PATH.format(healthmonitor_id=api_hm.get('id')),
self._build_body(new_hm))
response = self.get(self.HM_PATH.format(
healthmonitor_id=api_hm.get('id'))).json.get(self.root_tag)
self.assertEqual(constants.HEALTH_MONITOR_DEFAULT_URL_PATH,
response['url_path'])
def test_update_HTTP_expected_codes_none(self):
api_hm = self.create_health_monitor(
self.pool_with_listener_id,
constants.HEALTH_MONITOR_HTTP, 1, 1, 1, 1).get(self.root_tag)
self.set_lb_status(self.lb_id)
new_hm = {'expected_codes': None}
self.put(
self.HM_PATH.format(healthmonitor_id=api_hm.get('id')),
self._build_body(new_hm))
response = self.get(self.HM_PATH.format(
healthmonitor_id=api_hm.get('id'))).json.get(self.root_tag)
self.assertEqual(constants.HEALTH_MONITOR_DEFAULT_EXPECTED_CODES,
response['expected_codes'])
def test_delete(self):
api_hm = self.create_health_monitor(
self.pool_with_listener_id,

10
octavia/tests/unit/api/v2/types/test_health_monitors.py
View File

@ -152,9 +152,9 @@ class TestHealthMonitorPOST(base.BaseTypesTest, TestHealthMonitor):
"timeout": 1, "max_retries": 1,
"pool_id": uuidutils.generate_uuid()}
hmpost = wsme_json.fromjson(self._type, body)
self.assertEqual('GET', hmpost.http_method)
self.assertEqual('/', hmpost.url_path)
self.assertEqual('200', hmpost.expected_codes)
self.assertEqual(wsme_types.Unset, hmpost.http_method)
self.assertEqual(wsme_types.Unset, hmpost.url_path)
self.assertEqual(wsme_types.Unset, hmpost.expected_codes)
self.assertEqual(3, hmpost.max_retries_down)
self.assertTrue(hmpost.admin_state_up)
@ -165,9 +165,9 @@ class TestHealthMonitorPOST(base.BaseTypesTest, TestHealthMonitor):
"pool_id": uuidutils.generate_uuid(),
"url_path": url_path}
hmpost = wsme_json.fromjson(self._type, body)
self.assertEqual('GET', hmpost.http_method)
self.assertEqual(wsme_types.Unset, hmpost.http_method)
self.assertEqual(url_path, hmpost.url_path)
self.assertEqual('200', hmpost.expected_codes)
self.assertEqual(wsme_types.Unset, hmpost.expected_codes)
self.assertEqual(3, hmpost.max_retries_down)
self.assertTrue(hmpost.admin_state_up)