wbrawner/ansible
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

cloud: huawei: Add new module hwc_network_vpc (#54102)

Browse source
This commit is contained in:
zhongjun2 2019-03-31 00:20:54 +08:00 committed by René Moser
parent 51f41a2298
commit 5599b0484f
9 changed files with 1314 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

324
lib/ansible/module_utils/hwc_utils.py Normal file
View file

@ -0,0 +1,324 @@
# Copyright (c), Google Inc, 2017
# Simplified BSD License (see licenses/simplified_bsd.txt or
# https://opensource.org/licenses/BSD-2-Clause)
import traceback
REQUESTS_IMP_ERR = None
try:
import requests
HAS_REQUESTS = True
except ImportError:
REQUESTS_IMP_ERR = traceback.format_exc()
HAS_REQUESTS = False
THIRD_LIBRARIES_IMP_ERR = None
try:
from keystoneauth1.adapter import Adapter
from keystoneauth1.identity import v3
from keystoneauth1 import session
HAS_THIRD_LIBRARIES = True
except ImportError:
THIRD_LIBRARIES_IMP_ERR = traceback.format_exc()
HAS_THIRD_LIBRARIES = False
from ansible.module_utils.basic import AnsibleModule, env_fallback, missing_required_lib
from ansible.module_utils._text import to_text
def navigate_hash(source, path, default=None):
if not (source and path):
return None
key = path[0]
path = path[1:]
if key not in source:
return default
result = source[key]
if path:
return navigate_hash(result, path, default)
else:
return result
class HwcRequestException(Exception):
pass
def remove_empty_from_dict(obj):
return _DictClean(
obj,
lambda v: v is not None and v != {} and v != []
)()
def remove_nones_from_dict(obj):
return _DictClean(obj, lambda v: v is not None)()
def replace_resource_dict(item, value):
""" Handles the replacement of dicts with values -> the needed value for HWC API"""
if isinstance(item, list):
items = []
for i in item:
items.append(replace_resource_dict(i, value))
return items
else:
if not item:
return item
return item.get(value)
def are_dicts_different(expect, actual):
"""Remove all output-only from actual."""
actual_vals = {}
for k, v in actual.items():
if k in expect:
actual_vals[k] = v
expect_vals = {}
for k, v in expect.items():
if k in actual:
expect_vals[k] = v
return DictComparison(expect_vals) != DictComparison(actual_vals)
class HwcSession(object):
"""Handles all authentation and HTTP sessions for HWC API calls."""
def __init__(self, module, product):
self.module = module
self.product = product
self._validate()
self._session = self._credentials()
self._adapter = Adapter(self._session)
self._endpoints = {}
self._project_id = ""
def get(self, url, body=None):
try:
return self._adapter.get(
url, json=body,
headers=self._headers(), raise_exc=False)
except getattr(requests.exceptions, 'RequestException') as inst:
self.module.fail_json(msg=inst.message)
def post(self, url, body=None):
try:
return self._adapter.post(
url, json=body,
headers=self._headers(), raise_exc=False)
except getattr(requests.exceptions, 'RequestException') as inst:
self.module.fail_json(msg=inst.message)
def delete(self, url, body=None):
try:
return self._adapter.delete(
url, json=body,
headers=self._headers(), raise_exc=False)
except getattr(requests.exceptions, 'RequestException') as inst:
self.module.fail_json(msg=inst.message)
def put(self, url, body=None):
try:
return self._adapter.put(
url, json=body,
headers=self._headers(), raise_exc=False)
except getattr(requests.exceptions, 'RequestException') as inst:
self.module.fail_json(msg=inst.message)
def get_service_endpoint(self, service_type):
if self._endpoints.get(service_type):
return self._endpoints.get(service_type)
e = None
try:
e = self._session.get_endpoint_data(
service_type=service_type,
region_name=self.module.params['region']
)
except getattr(requests.exceptions, 'RequestException') as inst:
self.module.fail_json(msg=inst.message)
if not e or e.url == "":
self.module.fail_json(
msg="Can not find the enpoint for %s" % service_type)
url = e.url
if url[-1] != "/":
url += "/"
self._endpoints[service_type] = url
return url
def get_project_id(self):
if self._project_id:
return self._project_id
try:
pid = self._session.get_project_id()
self._project_id = pid
return pid
except getattr(requests.exceptions, 'RequestException') as inst:
self.module.fail_json(msg=inst.message)
def _validate(self):
if not HAS_REQUESTS:
self.module.fail_json(msg=missing_required_lib('requests'),
exception=REQUESTS_IMP_ERR)
if not HAS_THIRD_LIBRARIES:
self.module.fail_json(
msg=missing_required_lib('keystoneauth1'),
exception=THIRD_LIBRARIES_IMP_ERR)
def _credentials(self):
auth = v3.Password(
auth_url=self.module.params['identity_endpoint'],
password=self.module.params['password'],
username=self.module.params['user'],
user_domain_name=self.module.params['domain'],
project_name=self.module.params['project'],
reauthenticate=True
)
return session.Session(auth=auth)
def _headers(self):
return {
'User-Agent': "Huawei-Ansible-MM-%s" % self.product,
'Accept': 'application/json',
}
class HwcModule(AnsibleModule):
def __init__(self, *args, **kwargs):
arg_spec = kwargs.setdefault('argument_spec', {})
arg_spec.update(
dict(
identity_endpoint=dict(
required=True, type='str',
fallback=(env_fallback, ['ANSIBLE_HWC_IDENTITY_ENDPOINT']),
),
user=dict(
required=True, type='str',
fallback=(env_fallback, ['ANSIBLE_HWC_USER']),
),
password=dict(
required=True, type='str', no_log=True,
fallback=(env_fallback, ['ANSIBLE_HWC_PASSWORD']),
),
domain=dict(
required=True, type='str',
fallback=(env_fallback, ['ANSIBLE_HWC_DOMAIN']),
),
project=dict(
required=True, type='str',
fallback=(env_fallback, ['ANSIBLE_HWC_PROJECT']),
),
region=dict(
required=True, type='str',
fallback=(env_fallback, ['ANSIBLE_HWC_REGION']),
),
timeouts=dict(type='dict', options=dict(
create=dict(default='10m', type='str'),
update=dict(default='10m', type='str'),
delete=dict(default='10m', type='str'),
), default={}),
id=dict(type='str')
)
)
super(HwcModule, self).__init__(*args, **kwargs)
class DictComparison(object):
''' This class takes in two dictionaries `a` and `b`.
These are dictionaries of arbitrary depth, but made up of standard
Python types only.
This differ will compare all values in `a` to those in `b`.
Note: Only keys in `a` will be compared. Extra keys in `b` will be ignored.
Note: On all lists, order does matter.
'''
def __init__(self, request):
self.request = request
def __eq__(self, other):
return self._compare_dicts(self.request, other.request)
def __ne__(self, other):
return not self.__eq__(other)
def _compare_dicts(self, dict1, dict2):
if len(dict1.keys()) != len(dict2.keys()):
return False
return all([
self._compare_value(dict1.get(k), dict2.get(k)) for k in dict1
])
def _compare_lists(self, list1, list2):
"""Takes in two lists and compares them."""
if len(list1) != len(list2):
return False
difference = []
for index in range(len(list1)):
value1 = list1[index]
if index < len(list2):
value2 = list2[index]
difference.append(self._compare_value(value1, value2))
return all(difference)
def _compare_value(self, value1, value2):
"""
return: True: value1 is same as value2, otherwise False.
"""
if not (value1 and value2):
return (not value1) and (not value2)
# Can assume non-None types at this point.
if isinstance(value1, list):
return self._compare_lists(value1, value2)
elif isinstance(value1, dict):
return self._compare_dicts(value1, value2)
# Always use to_text values to avoid unicode issues.
else:
return (to_text(value1, errors='surrogate_or_strict')
== to_text(value2, errors='surrogate_or_strict'))
class _DictClean(object):
def __init__(self, obj, func):
self.obj = obj
self.keep_it = func
def __call__(self):
return self._clean_dict(self.obj)
def _clean_dict(self, obj):
r = {}
for k, v in obj.items():
v1 = v
if isinstance(v, dict):
v1 = self._clean_dict(v)
elif isinstance(v, list):
v1 = self._clean_list(v)
if self.keep_it(v1):
r[k] = v1
return r
def _clean_list(self, obj):
r = []
for v in obj:
v1 = v
if isinstance(v, dict):
v1 = self._clean_dict(v)
elif isinstance(v, list):
v1 = self._clean_list(v)
if self.keep_it(v1):
r.append(v1)
return r

0
lib/ansible/modules/cloud/huawei/__init__.py Normal file
View file

500
lib/ansible/modules/cloud/huawei/hwc_network_vpc.py Normal file
View file

@ -0,0 +1,500 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2018 Huawei
# GNU General Public License v3.0+ (see COPYING or
# https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
###############################################################################
# Documentation
###############################################################################
ANSIBLE_METADATA = {'metadata_version': '1.1',
'status': ["preview"],
'supported_by': 'community'}
DOCUMENTATION = '''
---
module: hwc_network_vpc
description:
- Represents an vpc resource.
short_description: Creates a Huawei Cloud VPC
version_added: '2.8'
author: Huawei Inc. (@huaweicloud)
requirements:
- requests >= 2.18.4
- keystoneauth1 >= 3.6.0
options:
state:
description:
- Whether the given object should exist in vpc
type: str
choices: ['present', 'absent']
default: 'present'
name:
description:
- the name of vpc.
type: str
required: true
cidr:
description:
- the range of available subnets in the vpc.
type: str
required: true
extends_documentation_fragment: hwc
'''
EXAMPLES = '''
- name: create a vpc
hwc_network_vpc:
identity_endpoint: "{{ identity_endpoint }}"
user: "{{ user }}"
password: "{{ password }}"
domain: "{{ domain }}"
project: "{{ project }}"
region: "{{ region }}"
name: "vpc_1"
cidr: "192.168.100.0/24"
state: present
'''
RETURN = '''
id:
description:
- the id of vpc.
type: str
returned: success
name:
description:
- the name of vpc.
type: str
returned: success
cidr:
description:
- the range of available subnets in the vpc.
type: str
returned: success
status:
description:
- the status of vpc.
type: str
returned: success
routes:
description:
- the route information.
type: complex
returned: success
contains:
destination:
description:
- the destination network segment of a route.
type: str
returned: success
next_hop:
description:
- the next hop of a route.
type: str
returned: success
enable_shared_snat:
description:
- show whether the shared snat is enabled.
type: bool
returned: success
'''
###############################################################################
# Imports
###############################################################################
from ansible.module_utils.hwc_utils import (HwcSession, HwcModule,
DictComparison, navigate_hash,
remove_nones_from_dict,
remove_empty_from_dict,
are_dicts_different)
import json
import re
import time
###############################################################################
# Main
###############################################################################
def main():
"""Main function"""
module = HwcModule(
argument_spec=dict(
state=dict(default='present', choices=['present', 'absent'], type='str'),
name=dict(required=True, type='str'),
cidr=dict(required=True, type='str')
),
supports_check_mode=True,
)
session = HwcSession(module, 'network')
state = module.params['state']
if (not module.params.get("id")) and module.params.get("name"):
module.params['id'] = get_id_by_name(session)
fetch = None
link = self_link(session)
# the link will include Nones if required format parameters are missed
if not re.search('/None/|/None$', link):
fetch = fetch_resource(session, link)
if fetch:
fetch = fetch.get('vpc')
changed = False
if fetch:
if state == 'present':
expect = _get_editable_properties(module)
current_state = response_to_hash(module, fetch)
if are_dicts_different(expect, current_state):
if not module.check_mode:
fetch = update(session, self_link(session), [200])
fetch = response_to_hash(module, fetch.get('vpc'))
changed = True
else:
fetch = current_state
else:
if not module.check_mode:
delete(session, self_link(session))
fetch = {}
changed = True
else:
if state == 'present':
if not module.check_mode:
fetch = create(session, collection(session), [200])
fetch = response_to_hash(module, fetch.get('vpc'))
changed = True
else:
fetch = {}
fetch.update({'changed': changed})
module.exit_json(**fetch)
def create(session, link, success_codes=None):
if not success_codes:
success_codes = [201, 202]
module = session.module
r = return_if_object(module, session.post(link, resource_to_create(module)), success_codes)
wait_done = wait_for_operation(session, 'create', r)
url = resource_get_url(session, wait_done)
return fetch_resource(session, url)
def update(session, link, success_codes=None):
if not success_codes:
success_codes = [201, 202]
module = session.module
r = return_if_object(module, session.put(link, resource_to_update(module)), success_codes)
wait_done = wait_for_operation(session, 'update', r)
url = resource_get_url(session, wait_done)
return fetch_resource(session, url)
def delete(session, link, success_codes=None):
if not success_codes:
success_codes = [202, 204]
return_if_object(session.module, session.delete(link), success_codes, False)
wait_for_delete(session, link)
def fetch_resource(session, link, success_codes=None):
if not success_codes:
success_codes = [200]
return return_if_object(session.module, session.get(link), success_codes)
def link_wrapper(f):
def _wrapper(module, *args, **kwargs):
try:
return f(module, *args, **kwargs)
except KeyError as ex:
module.fail_json(
msg="Mapping keys(%s) are not found in generating link." % ex)
return _wrapper
def get_id_by_name(session):
module = session.module
name = module.params.get("name")
link = list_link(session, {'limit': 10, 'marker': '{marker}'})
not_format_keys = re.findall("={marker}", link)
none_values = re.findall("=None", link)
if not (not_format_keys or none_values):
r = fetch_resource(session, link)
if r is None:
return ""
r = r.get('vpcs', [])
ids = [
i.get('id') for i in r if i.get('name', '') == name
]
if not ids:
return ""
elif len(ids) == 1:
return ids[0]
else:
module.fail_json(msg="Multiple resources with same name are found.")
elif none_values:
module.fail_json(
msg="Can not find id by name because url includes None.")
else:
p = {'marker': ''}
ids = set()
while True:
r = fetch_resource(session, link.format(**p))
if r is None:
break
r = r.get('vpcs', [])
if r == []:
break
for i in r:
if i.get('name') == name:
ids.add(i.get('id'))
if len(ids) >= 2:
module.fail_json(
msg="Multiple resources with same name are found.")
p['marker'] = r[-1].get('id')
return ids.pop() if ids else ""
@link_wrapper
def list_link(session, extra_data=None):
url = "{endpoint}vpcs?limit={limit}&marker={marker}"
combined = session.module.params.copy()
if extra_data:
combined.update(extra_data)
combined['endpoint'] = session.get_service_endpoint('vpc')
return url.format(**combined)
@link_wrapper
def self_link(session):
url = "{endpoint}vpcs/{id}"
combined = session.module.params.copy()
combined['endpoint'] = session.get_service_endpoint('vpc')
return url.format(**combined)
@link_wrapper
def collection(session):
url = "{endpoint}vpcs"
combined = session.module.params.copy()
combined['endpoint'] = session.get_service_endpoint('vpc')
return url.format(**combined)
def return_if_object(module, response, success_codes, has_content=True):
code = response.status_code
# If not found, return nothing.
if code == 404:
return None
success_codes = [200, 201, 202, 203, 204, 205, 206, 207, 208, 226]
# If no content, return nothing.
if code in success_codes and not has_content:
return None
result = None
try:
result = response.json()
except getattr(json.decoder, 'JSONDecodeError', ValueError) as inst:
module.fail_json(msg="Invalid JSON response with error: %s" % inst)
if code not in success_codes:
msg = navigate_hash(result, ['message'])
if msg:
module.fail_json(msg=msg)
else:
module.fail_json(msg="operation failed, return code=%d" % code)
return result
def resource_to_create(module):
request = remove_empty_from_dict({
u'name': module.params.get('name'),
u'cidr': module.params.get('cidr')
})
return {'vpc': request}
def resource_to_update(module):
request = remove_nones_from_dict({
u'name': module.params.get('name'),
u'cidr': module.params.get('cidr')
})
return {'vpc': request}
def _get_editable_properties(module):
request = remove_nones_from_dict({
"name": module.params.get("name"),
"cidr": module.params.get("cidr"),
})
return request
def response_to_hash(module, response):
""" Remove unnecessary properties from the response.
This is for doing comparisons with Ansible's current parameters.
"""
return {
u'id': response.get(u'id'),
u'name': response.get(u'name'),
u'cidr': response.get(u'cidr'),
u'status': response.get(u'status'),
u'routes': VpcRoutesArray(response.get(u'routes', []), module).from_response(),
u'enable_shared_snat': response.get(u'enable_shared_snat')
}
@link_wrapper
def resource_get_url(session, wait_done):
combined = session.module.params.copy()
combined['op_id'] = navigate_hash(wait_done, ['vpc', 'id'])
url = 'vpcs/{op_id}'.format(**combined)
endpoint = session.get_service_endpoint('vpc')
return endpoint + url
@link_wrapper
def async_op_url(session, extra_data=None):
url = "{endpoint}vpcs/{op_id}"
combined = session.module.params.copy()
if extra_data:
combined.update(extra_data)
combined['endpoint'] = session.get_service_endpoint('vpc')
return url.format(**combined)
def wait_for_operation(session, op_type, op_result):
op_id = navigate_hash(op_result, ['vpc', 'id'])
url = async_op_url(session, {'op_id': op_id})
timeout = 60 * int(session.module.params['timeouts'][op_type].rstrip('m'))
states = {
'create': {
'allowed': ['CREATING', 'DONW', 'OK'],
'complete': ['OK'],
},
'update': {
'allowed': ['PENDING_UPDATE', 'DONW', 'OK'],
'complete': ['OK'],
}
}
return wait_for_completion(url, timeout, states[op_type]['allowed'],
states[op_type]['complete'], session)
def wait_for_completion(op_uri, timeout, allowed_states,
complete_states, session):
module = session.module
end = time.time() + timeout
while time.time() <= end:
try:
op_result = fetch_resource(session, op_uri)
except Exception:
time.sleep(1.0)
continue
raise_if_errors(op_result, module)
status = navigate_hash(op_result, ['vpc', 'status'])
if status not in allowed_states:
module.fail_json(msg="Invalid async operation status %s" % status)
if status in complete_states:
return op_result
time.sleep(1.0)
module.fail_json(msg="Timeout to wait completion.")
def raise_if_errors(response, module):
errors = navigate_hash(response, [])
if errors:
module.fail_json(msg=navigate_hash(response, []))
def wait_for_delete(session, link):
end = time.time() + 60 * int(
session.module.params['timeouts']['delete'].rstrip('m'))
while time.time() <= end:
try:
resp = session.get(link)
if resp.status_code == 404:
return
except Exception:
pass
time.sleep(1.0)
session.module.fail_json(msg="Timeout to wait for deletion to be complete.")
class VpcRoutesArray(object):
def __init__(self, request, module):
self.module = module
if request:
self.request = request
else:
self.request = []
def to_request(self):
items = []
for item in self.request:
items.append(self._request_for_item(item))
return items
def from_response(self):
items = []
for item in self.request:
items.append(self._response_from_item(item))
return items
def _request_for_item(self, item):
return {
u'destination': item.get('destination'),
u'nexthop': item.get('next_hop')
}
def _response_from_item(self, item):
return {
u'destination': item.get(u'destination'),
u'next_hop': item.get(u'nexthop')
}
if __name__ == '__main__':
main()

82
lib/ansible/plugins/doc_fragments/hwc.py Normal file
View file

@ -0,0 +1,82 @@
# Copyright: (c) 2018, Huawei Inc.
# GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
class ModuleDocFragment(object):
# HWC doc fragment.
DOCUMENTATION = '''
options:
identity_endpoint:
description:
- The Identity authentication URL.
type: str
required: true
user:
description:
- The user name to login with (currently only user names are
supported, and not user IDs).
type: str
required: true
password:
description:
- The password to login with.
type: str
required: true
domain:
description:
- The name of the Domain to scope to (Identity v3, currently only
domain names are supported, and not domain IDs).
type: str
required: true
project:
description:
- The name of the Tenant (Identity v2) or Project (Identity v3).
(currently only project names are supported, and not
project IDs).
type: str
required: true
region:
description:
- The region to which the project belongs.
type: str
required: true
timeouts:
description:
- The timeouts for create/update/delete operation.
type: dict
suboptions:
create:
description:
- The timeouts for create operation.
type: str
default: '10m'
update:
description:
- The timeouts for update operation.
type: str
default: '10m'
delete:
description:
- The timeouts for delete operation.
type: str
id:
description:
- The id of resource to be managed.
type: str
notes:
- For authentication, you can set identity_endpoint using the
C(ANSIBLE_HWC_IDENTITY_ENDPOINT) env variable.
- For authentication, you can set user using the
C(ANSIBLE_HWC_USER) env variable.
- For authentication, you can set password using the C(ANSIBLE_HWC_PASSWORD) env
variable.
- For authentication, you can set domain using the C(ANSIBLE_HWC_DOMAIN) env
variable.
- For authentication, you can set project using the C(ANSIBLE_HWC_PROJECT) env
variable.
- For authentication, you can set region using the C(ANSIBLE_HWC_REGION) env variable.
- Environment variables values will only be used if the playbook values are
not set.
'''

10
test/integration/cloud-config-hwc.yml.template Normal file
View file

@ -0,0 +1,10 @@
# This is the configuration template for ansible-test HWC integration tests.
# Please fill in the @VAR placeholders below and save this file without the
# .template extension.
identity_endpoint: @identity_endpoint
user: @user
password: @password
domain: @domain
project: @project
region: @region

1
test/integration/targets/hwc_network_vpc/aliases Normal file
View file

@ -0,0 +1 @@
unsupported

96
test/integration/targets/hwc_network_vpc/tasks/main.yml Normal file
View file

@ -0,0 +1,96 @@
---
# ----------------------------------------------------------------------------
#
# *** AUTO GENERATED CODE *** AUTO GENERATED CODE ***
#
# ----------------------------------------------------------------------------
#
# This file is automatically generated by Magic Modules and manual
# changes will be clobbered when the file is regenerated.
#
# Please read more about how to change this file at
# https://www.github.com/huaweicloud/magic-modules
#
# ----------------------------------------------------------------------------
# Pre-test setup
- name: delete a vpc
hwc_network_vpc:
identity_endpoint: "{{ identity_endpoint }}"
user: "{{ user }}"
password: "{{ password }}"
domain: "{{ domain }}"
project: "{{ project }}"
region: "{{ region }}"
name: "vpc_1"
cidr: "192.168.100.0/24"
state: absent
#----------------------------------------------------------
- name: create a vpc
hwc_network_vpc:
identity_endpoint: "{{ identity_endpoint }}"
user: "{{ user }}"
password: "{{ password }}"
domain: "{{ domain }}"
project: "{{ project }}"
region: "{{ region }}"
name: "vpc_1"
cidr: "192.168.100.0/24"
state: present
register: result
- name: assert changed is true
assert:
that:
- result is changed
# ----------------------------------------------------------------------------
- name: create a vpc that already exists
hwc_network_vpc:
identity_endpoint: "{{ identity_endpoint }}"
user: "{{ user }}"
password: "{{ password }}"
domain: "{{ domain }}"
project: "{{ project }}"
region: "{{ region }}"
name: "vpc_1"
cidr: "192.168.100.0/24"
state: present
register: result
- name: assert changed is false
assert:
that:
- result.failed == 0
- result.changed == false
#----------------------------------------------------------
- name: delete a vpc
hwc_network_vpc:
identity_endpoint: "{{ identity_endpoint }}"
user: "{{ user }}"
password: "{{ password }}"
domain: "{{ domain }}"
project: "{{ project }}"
region: "{{ region }}"
name: "vpc_1"
cidr: "192.168.100.0/24"
state: absent
register: result
- name: assert changed is true
assert:
that:
- result is changed
# ----------------------------------------------------------------------------
- name: delete a vpc that does not exist
hwc_network_vpc:
identity_endpoint: "{{ identity_endpoint }}"
user: "{{ user }}"
password: "{{ password }}"
domain: "{{ domain }}"
project: "{{ project }}"
region: "{{ region }}"
name: "vpc_1"
cidr: "192.168.100.0/24"
state: absent
register: result
- name: assert changed is false
assert:
that:
- result.failed == 0
- result.changed == false

192
test/units/module_utils/hwc/test_dict_comparison.py Normal file
View file

@ -0,0 +1,192 @@
# -*- coding: utf-8 -*-
# 2018.07.26 --- use DictComparison instead of GcpRequest
#
# (c) 2016, Tom Melendez <tom@supertom.com>
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
import os
import sys
from units.compat import unittest
from ansible.module_utils.hwc_utils import DictComparison
class HwcDictComparisonTestCase(unittest.TestCase):
def test_simple_no_difference(self):
value1 = {
'foo': 'bar',
'test': 'original'
}
d = DictComparison(value1)
d_ = d
self.assertTrue(d == d_)
def test_simple_different(self):
value1 = {
'foo': 'bar',
'test': 'original'
}
value2 = {
'foo': 'bar',
'test': 'different'
}
value3 = {
'test': 'original'
}
dict1 = DictComparison(value1)
dict2 = DictComparison(value2)
dict3 = DictComparison(value3)
self.assertFalse(dict1 == dict2)
self.assertFalse(dict1 == dict3)
self.assertFalse(dict2 == dict3)
def test_nested_dictionaries_no_difference(self):
value1 = {
'foo': {
'quiet': {
'tree': 'test'
},
'bar': 'baz'
},
'test': 'original'
}
d = DictComparison(value1)
d_ = d
self.assertTrue(d == d_)
def test_nested_dictionaries_with_difference(self):
value1 = {
'foo': {
'quiet': {
'tree': 'test'
},
'bar': 'baz'
},
'test': 'original'
}
value2 = {
'foo': {
'quiet': {
'tree': 'baz'
},
'bar': 'hello'
},
'test': 'original'
}
value3 = {
'foo': {
'quiet': {
'tree': 'test'
},
'bar': 'baz'
}
}
dict1 = DictComparison(value1)
dict2 = DictComparison(value2)
dict3 = DictComparison(value3)
self.assertFalse(dict1 == dict2)
self.assertFalse(dict1 == dict3)
self.assertFalse(dict2 == dict3)
def test_arrays_strings_no_difference(self):
value1 = {
'foo': [
'baz',
'bar'
]
}
d = DictComparison(value1)
d_ = d
self.assertTrue(d == d_)
def test_arrays_strings_with_difference(self):
value1 = {
'foo': [
'baz',
'bar',
]
}
value2 = {
'foo': [
'baz',
'hello'
]
}
value3 = {
'foo': [
'bar',
]
}
dict1 = DictComparison(value1)
dict2 = DictComparison(value2)
dict3 = DictComparison(value3)
self.assertFalse(dict1 == dict2)
self.assertFalse(dict1 == dict3)
self.assertFalse(dict2 == dict3)
def test_arrays_dicts_with_no_difference(self):
value1 = {
'foo': [
{
'test': 'value',
'foo': 'bar'
},
{
'different': 'dict'
}
]
}
d = DictComparison(value1)
d_ = d
self.assertTrue(d == d_)
def test_arrays_dicts_with_difference(self):
value1 = {
'foo': [
{
'test': 'value',
'foo': 'bar'
},
{
'different': 'dict'
}
]
}
value2 = {
'foo': [
{
'test': 'value2',
'foo': 'bar2'
},
]
}
value3 = {
'foo': [
{
'test': 'value',
'foo': 'bar'
}
]
}
dict1 = DictComparison(value1)
dict2 = DictComparison(value2)
dict3 = DictComparison(value3)
self.assertFalse(dict1 == dict2)
self.assertFalse(dict1 == dict3)
self.assertFalse(dict2 == dict3)

109
test/units/module_utils/hwc/test_hwc_utils.py Normal file
View file

@ -0,0 +1,109 @@
# -*- coding: utf-8 -*-
import os
import sys
from units.compat import unittest
from ansible.module_utils.hwc_utils import (navigate_hash,
remove_empty_from_dict,
remove_nones_from_dict,
replace_resource_dict)
class HwcUtilsTestCase(unittest.TestCase):
def test_navigate_hash(self):
value = {
'foo': {
'quiet': {
'tree': 'test'
},
}
}
self.assertEquals(navigate_hash(value, ["foo", "quiet", "tree"]),
"test")
self.assertEquals(navigate_hash(value, ["foo", "q", "tree"], 123),
123)
self.assertIsNone(navigate_hash(value, [], 123))
def test_remove_empty_from_dict(self):
value = {
'foo': {
'quiet': {
'tree': 'test',
'tree1': [
None,
{},
[],
'test'
],
'tree2': {},
'tree3': []
},
},
'foo1': [],
'foo2': {},
'foo3': None,
}
expect = {
'foo': {
'quiet': {
'tree': 'test',
'tree1': [
'test'
]
},
},
}
self.assertEqual(remove_empty_from_dict(value), expect)
def test_remove_nones_from_dict(self):
value = {
'foo': {
'quiet': {
'tree': 'test',
'tree1': [
None,
{},
[],
'test'
],
'tree2': {},
'tree3': []
},
},
'foo1': [],
'foo2': {},
'foo3': None,
}
expect = {
'foo': {
'quiet': {
'tree': 'test',
'tree1': [
{},
[],
'test'
],
'tree2': {},
'tree3': []
},
},
'foo1': [],
'foo2': {},
}
self.assertEqual(remove_nones_from_dict(value), expect)
def test_replace_resource_dict(self):
self.assertEqual(replace_resource_dict({'foo': 'quiet'}, 'foo'), 'quiet')
self.assertEqual(replace_resource_dict({}, 'foo'), {})
self.assertEqual(replace_resource_dict([[{'foo': 'quiet'}]], 'foo'),
[['quiet']])