diff --git a/.github/workflows/integration_source.yml b/.github/workflows/integration_source.yml index a30e412f..ccd63079 100644 --- a/.github/workflows/integration_source.yml +++ b/.github/workflows/integration_source.yml @@ -179,3 +179,7 @@ jobs: - name: Run configuration_item_batch integration tests run: ansible-test integration configuration_item_batch working-directory: ${{ steps.identify.outputs.collection_path }} + + - name: Run configuration_item_relations integration tests + run: ansible-test integration configuration_item_relations + working-directory: ${{ steps.identify.outputs.collection_path }} diff --git a/changelogs/fragments/ci_relations.yml b/changelogs/fragments/ci_relations.yml new file mode 100644 index 00000000..5f1c1fb3 --- /dev/null +++ b/changelogs/fragments/ci_relations.yml @@ -0,0 +1,3 @@ +--- +minor_changes: + - configuration_item_relations - add module to add and remove relations between configuration items. diff --git a/changelogs/fragments/ci_relations_info.yml b/changelogs/fragments/ci_relations_info.yml new file mode 100644 index 00000000..59c6f85c --- /dev/null +++ b/changelogs/fragments/ci_relations_info.yml @@ -0,0 +1,3 @@ +--- +minor_changes: + - configuration_item_relations_info - add module retrieve relations of a configuration item. diff --git a/plugins/module_utils/client.py b/plugins/module_utils/client.py index 8e903cf6..7b29365f 100644 --- a/plugins/module_utils/client.py +++ b/plugins/module_utils/client.py @@ -220,5 +220,6 @@ def put(self, path, data, query=None): def delete(self, path, query=None): resp = self.request("DELETE", path, query=query) - if resp.status != 204: - raise UnexpectedAPIResponse(resp.status, resp.data) + if resp.status in (200, 204): + return resp + raise UnexpectedAPIResponse(resp.status, resp.data) diff --git a/plugins/module_utils/cmdb_relation.py b/plugins/module_utils/cmdb_relation.py new file mode 100644 index 00000000..e119b905 --- /dev/null +++ b/plugins/module_utils/cmdb_relation.py @@ -0,0 +1,196 @@ +# -*- coding: utf-8 -*- +# Copyright: (c) 2024, Red Hat +# +# 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 + + +OUTBOUND_KEY = "outbound_relations" +INBOUND_KEY = "inbound_relations" +OUTBOUND = "outbound" +INBOUND = "inbound" + + +class CmdbRelation(object): + """ + CmdbRelation is a representation of the relation from CMDB Instance API. + #!/reference/api/utah/rest/cmdb-instance-api#cmdb-POST-instance-relation + Please refer to: https://developer.servicenow.com/dev.do + """ + + def __init__(self, value): + if "sys_id" not in value: + raise ValueError("Relation has no sys_id") + if "type" not in value or not isinstance(value["type"], dict): + raise ValueError("Relation has no type or type is not a dictionary") + if "target" not in value or not isinstance(value["target"], dict): + raise ValueError("Relation has no target or target is not a dictionary") + + self.sys_id = value["sys_id"] + self.type_name = value["type"]["display_value"] + self.type_id = value["type"]["value"] + self.target_id = value["target"]["value"] + self.target_name = value["target"]["display_value"] + + def __eq__(self, o): + # new relations don't have a sys_id yet + if o.sys_id and self.sys_id: + return o.sys_id == self.sys_id + return o.target_id == self.target_id and o.type_id == self.type_id + + def to_payload(self): + return dict( + type=self.type_id, + target=self.target_id, + ) + + def to_json(self): + return dict( + sys_id=self.sys_id, + target=dict(value=self.target_id, display_value=self.target_name), + type=dict(value=self.type_id, display_value=self.type_name), + ) + + @classmethod + def from_values(cls, type_sys_id, type_name, target_sys_id, target_name): + d = dict( + sys_id=None, + type=dict(value=type_sys_id, display_value=type_name), + target=dict(value=target_sys_id, display_value=target_name), + ) + return cls(d) + + +class CmdbItemRelations(object): + """CmdbItemRelations manage the relations of a configuration item.""" + + def __init__(self, configuration_item=None): + self.relations = [] + # holds the tainted(added/deleted) relations as a tuple (direction, action, relation) + self.tainted = [] + + if configuration_item: + self.configuration_item = configuration_item + self.__read(configuration_item) + + def __iter__(self): + for dir, relation in self.relations: + yield dir, relation + + def clone(self): + c = CmdbItemRelations() + c.relations = self.relations[:] + return c + + def get(self, direction, target_id): + """Get returns a relation based on direction and target_id""" + for dir, relation in self: + if relation.target_id == target_id and dir == direction: + return relation + return None + + def add(self, direction, relation): + """Add adds a new relation. + User must call update to actually make the request. + """ + for dir, action, r in self.tainted: + if dir == direction and r == relation and action == "add": + return + self.tainted.append((direction, "add", relation)) + + def remove(self, direction, relation): + """Remove removes a relation. + User must call update to actually make the request. + """ + for dir, action, r in self.tainted: + if dir == direction and r == relation and action == "remove": + return + self.tainted.append((direction, "remove", relation)) + + def update(self, api_path, generic_client, check_mode=False): + """ + Update updates the configuration item with the tainted relations. + Due to the behaviour of the caller, this method is either called for adding + or for removing relations but not for both. + """ + if len(self.tainted) == 0: + return + + payload = self.__create_payload(check_mode) + if payload: + if check_mode: + return CmdbItemRelations(payload) + result = generic_client.create_record( + api_path, payload, check_mode=False, query=None + ) + # return a new instance of the ci + return CmdbItemRelations(result) + + # remove relations by calling DELETE endpoint + # SNow does not returned any response following a succesfull DELETE op + # So, we just remove the relation from a clone and return it. + clone = self.clone() + for dir, action, rel in self.tainted: + if action == "add": + continue + if not check_mode: + generic_client.delete_record_by_sys_id(api_path, rel.sys_id) + for idx, r in enumerate(clone): + if r[1].sys_id == rel.sys_id: + clone.relations.pop(idx) + break + return clone + + def to_json(self): + result = dict(outbound_relations=[], inbound_relations=[]) + for dir, rel in self: + if dir == OUTBOUND: + result[OUTBOUND_KEY].append(rel.to_json()) + if dir == INBOUND: + result[INBOUND_KEY].append(rel.to_json()) + return result + + def __create_payload(self, check_mode=False): + """ + Create payload for added relations + Return: payload if there're relation to be added. None otherwise + """ + payload = dict( + source="ServiceNow", + ) + + for dir, action, rel in self.tainted: + if action == "remove": + continue + if dir == OUTBOUND: + # it seems that SNow complains for empty list in payload. + # so we add the key only if we have to (inbound or outbound). + if OUTBOUND_KEY not in payload: + payload[OUTBOUND_KEY] = [] + if check_mode: + payload.get(OUTBOUND_KEY).append(rel.to_json()) + else: + payload.get(OUTBOUND_KEY).append(rel.to_payload()) + elif dir == INBOUND: + if INBOUND_KEY not in payload: + payload[INBOUND_KEY] = [] + if check_mode: + payload.get(INBOUND_KEY).append(rel.to_json()) + else: + payload.get(INBOUND_KEY).append(rel.to_payload()) + + if len(payload.get(INBOUND_KEY, [])) + len(payload.get(OUTBOUND_KEY, [])) > 0: + return payload + + return None + + def __read(self, configuration_item): + if OUTBOUND_KEY in configuration_item: + for r in configuration_item[OUTBOUND_KEY]: + self.relations.append((OUTBOUND, CmdbRelation(r))) + if INBOUND_KEY in configuration_item: + for r in configuration_item[INBOUND_KEY]: + self.relations.append((INBOUND, CmdbRelation(r))) diff --git a/plugins/modules/configuration_item_relations.py b/plugins/modules/configuration_item_relations.py new file mode 100644 index 00000000..5167bb91 --- /dev/null +++ b/plugins/modules/configuration_item_relations.py @@ -0,0 +1,342 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +# Copyright: (c) 2024, Red Hat +# +# 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 = r""" +module: configuration_item_relations + +author: + - Cosmin Tupangiu (@tupyy) + +short_description: Manage ServiceNow relations between configuration items + +description: + - Add and remove ServiceNow relations between configuration items. + +version_added: 2.5.0 + +extends_documentation_fragment: + - servicenow.itsm.instance + - servicenow.itsm.sysparm_display_value + +seealso: + - module: servicenow.itsm.configuration_item + - module: servicenow.itsm.configuration_item_info + +options: + state: + description: + - State of the relation. + type: str + choices: [ present, absent ] + default: present + name: + description: + - The name of the relation. + - Mutually exclusive with C(sys_id). + type: str + direction: + description: + - Direction of the relation. + type: str + choices: [ inbound, outbound ] + default: outbound + parent_sys_id: + description: + - The sys_id of the configuration item who own the relation. + type: str + required: true + parent_classname: + description: + - The class of the configuration item. + type: str + required: true + targets: + description: + - List of configuration items to be associated with the parent. + type: list + elements: dict + suboptions: + name: + description: + - Name of the configuration item + type: str + required: true + sys_id: + description: + - Sys_id of the configuration item + type: str + required: true + required: true +""" + +EXAMPLES = r""" +- name: Create relation between two ci + servicenow.itsm.configuration_item_relations: + name: Depends_On + direction: outbound + state: present + parent_sys_id: "{{ parent_sys_id }}" + parent_classname: cmdb_ci_linux_server + targets: + - name: target1 + sys_id: target1_id + +- name: Remove relation + servicenow.itsm.configuration_item_relations: + direction: outbound + state: absent + parent_sys_id: "{{ parent_sys_id }}" + parent_classname: cmdb_ci_linux_server + targets: + - name: target1 + sys_id: target1_id + +- name: Update relation by adding one more target + servicenow.itsm.configuration_item_relations: + name: Depends_On + direction: outbound + state: present + parent_sys_id: "{{ owner_sys_id }}" + parent_classname: cmdb_ci_linux_server + targets: + - name: target1 + sys_id: target1_id +""" + +RETURN = r""" +record: + description: + - The relations of the configuration item. + returned: success + type: dict + sample: + "inbound_relations": "" + "outbound_relations": + - "sys_id": "06d7f70697514210d8a379100153af3d" + "target": + "display_value": "PS LinuxApp01" + "value": "3a290cc60a0a0bb400000bdb386af1cf" + "type": + "display_value": "Cools::Cooled By" + "value": "015633570a0a0bc70029121512d46ede" +""" + +from ..module_utils.utils import get_mapper +from ..module_utils.configuration_item import PAYLOAD_FIELDS_MAPPING +from ..module_utils import cmdb_relation as cmdb +from ..module_utils import arguments, client, errors, generic +from ansible.module_utils.basic import AnsibleModule + +CMDB_INSTANCE_BASE_API_PATH = "api/now/cmdb/instance" +CMDB_RELATION_TYPE_API_PATH = "/api/now/table/cmdb_rel_type" + + +def ensure_present(module, generic_client): + mapper = get_mapper( + module, + "configuration_item_mapping", + PAYLOAD_FIELDS_MAPPING, + sysparm_display_value=module.params["sysparm_display_value"], + ) + + # get the relation type sys_id + relation_records = generic_client.list_records( + CMDB_RELATION_TYPE_API_PATH, dict(sys_name=module.params["name"]) + ) + if len(relation_records) == 0: + raise errors.ServiceNowError( + "Error finding relation by sys_name {0}".format(module.params["name"]) + ) + + relation_type_sys_id = relation_records[0]["sys_id"] + + parent_ci = mapper.to_ansible( + generic_client.get_by_sys_id( + "/".join([CMDB_INSTANCE_BASE_API_PATH, module.params["parent_classname"]]), + module.params["parent_sys_id"], + True, + ) + ) + + relations = cmdb.CmdbItemRelations(parent_ci) + + changed = False + for target in module.params["targets"]: + existing_relation = relations.get(module.params["direction"], target["sys_id"]) + if not existing_relation: + relations.add( + module.params["direction"], + cmdb.CmdbRelation.from_values( + relation_type_sys_id, + module.params["name"], + target["sys_id"], + target["name"], + ), + ) + changed = True + + if changed: + new_relations = relations.update( + "/".join( + [ + CMDB_INSTANCE_BASE_API_PATH, + module.params["parent_classname"], + module.params["parent_sys_id"], + "relation", + ] + ), + generic_client, + module.check_mode, + ) + + return ( + True, + new_relations.to_json(), + dict(before=relations.to_json(), after=new_relations.to_json()), + ) + + return ( + False, + relations.to_json(), + dict( + before=dict(record=relations.to_json()), + after=dict(record=relations.to_json()), + ), + ) + + +def ensure_absent(module, generic_client): + mapper = get_mapper( + module, + "configuration_item_mapping", + PAYLOAD_FIELDS_MAPPING, + sysparm_display_value=module.params["sysparm_display_value"], + ) + + parent_ci = mapper.to_ansible( + generic_client.get_by_sys_id( + "/".join([CMDB_INSTANCE_BASE_API_PATH, module.params["parent_classname"]]), + module.params["parent_sys_id"], + True, + ) + ) + + relations = cmdb.CmdbItemRelations(parent_ci) + + changed = False + for target in module.params["targets"]: + existing_relation = relations.get(module.params["direction"], target["sys_id"]) + if existing_relation: + relations.remove(module.params["direction"], existing_relation) + changed = True + + if changed: + new_relations = relations.update( + "/".join( + [ + CMDB_INSTANCE_BASE_API_PATH, + module.params["parent_classname"], + module.params["parent_sys_id"], + "relation", + ] + ), + generic_client, + module.check_mode, + ) + return ( + True, + new_relations.to_json(), + dict(before=relations.to_json(), after=new_relations.to_json()), + ) + + return ( + False, + relations.to_json(), + dict( + before=relations.to_json(), + after=relations.to_json(), + ), + ) + + +def run(module, generic_client): + if module.params["state"] == "absent": + return ensure_absent(module, generic_client) + return ensure_present(module, generic_client) + + +def main(): + module_args = dict( + arguments.get_spec( + "instance", + "sysparm_display_value", + ), + state=dict( + type="str", + choices=[ + "present", + "absent", + ], + default="present", + ), + name=dict( + type="str", + ), + direction=dict( + type="str", + choices=[ + "inbound", + "outbound", + ], + default="outbound", + ), + parent_sys_id=dict( + type="str", + required=True, + ), + parent_classname=dict( + type="str", + required=True, + ), + targets=dict( + type="list", + elements="dict", + options=dict( + name=dict( + type="str", + required=True, + ), + sys_id=dict( + type="str", + required=True, + ), + ), + required=True, + ), + ) + + module = AnsibleModule( + argument_spec=module_args, + supports_check_mode=True, + required_if=[("state", "present", ("name",))], + ) + + try: + snow_client = client.Client(**module.params["instance"]) + generic_client = generic.GenericClient(snow_client) + changed, record, diff = run(module, generic_client) + module.exit_json(changed=changed, record=record, diff=diff) + except errors.ServiceNowError as e: + module.fail_json(msg=str(e)) + + +if __name__ == "__main__": + main() diff --git a/plugins/modules/configuration_item_relations_info.py b/plugins/modules/configuration_item_relations_info.py new file mode 100644 index 00000000..5b2646c7 --- /dev/null +++ b/plugins/modules/configuration_item_relations_info.py @@ -0,0 +1,122 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +# Copyright: (c) 2024, Red Hat +# +# 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 = r""" +module: configuration_item_relations_info + +author: + - Cosmin Tupangiu (@tupyy) + +short_description: Retreive ServiceNow relations of configuration items + +description: + - Retreive configuration items relations + +version_added: 2.5.0 + +extends_documentation_fragment: + - servicenow.itsm.instance + - servicenow.itsm.sys_id.info + - servicenow.itsm.sysparm_display_value + +options: + classname: + description: + - The class of the configuration item. + type: str + required: true +seealso: + - module: servicenow.itsm.configuration_item_relations + +""" + +EXAMPLES = r""" +- name: Retreive relations of a configuration item + servicenow.itsm.configuration_item_relations_info: + sys_id: "{{ configuration_item_sys_id }}" + classname: cmdb_ci_linux_server +""" + +RETURN = r""" +record: + description: + - The relations of the configuration item. + returned: success + type: dict + sample: + "inbound_relations": "" + "outbound_relations": + - "sys_id": "06d7f70697514210d8a379100153af3d" + "target": + "display_value": "PS LinuxApp01" + "value": "3a290cc60a0a0bb400000bdb386af1cf" + "type": + "display_value": "Cools::Cooled By" + "value": "015633570a0a0bc70029121512d46ede" +""" + +from ansible.module_utils.basic import AnsibleModule +from ..module_utils import arguments, client, errors, generic +from ..module_utils import cmdb_relation as cmdb +from ..module_utils.configuration_item import PAYLOAD_FIELDS_MAPPING +from ..module_utils.utils import get_mapper + +CMDB_INSTANCE_BASE_API_PATH = "api/now/cmdb/instance" + + +def run(module, generic_client): + mapper = get_mapper( + module, + "configuration_item_mapping", + PAYLOAD_FIELDS_MAPPING, + sysparm_display_value=module.params["sysparm_display_value"], + ) + + ci = mapper.to_ansible( + generic_client.get_by_sys_id( + "/".join([CMDB_INSTANCE_BASE_API_PATH, module.params["classname"]]), + module.params["sys_id"], + True, + ) + ) + + return cmdb.CmdbItemRelations(ci).to_json() + + +def main(): + module_args = dict( + arguments.get_spec( + "instance", + "sys_id", + "sysparm_display_value", + ), + classname=dict( + type="str", + required=True, + ), + ) + + module = AnsibleModule( + argument_spec=module_args, + supports_check_mode=True, + ) + + try: + snow_client = client.Client(**module.params["instance"]) + generic_client = generic.GenericClient(snow_client) + records = run(module, generic_client) + module.exit_json(changed=False, record=records) + except errors.ServiceNowError as e: + module.fail_json(msg=str(e)) + + +if __name__ == "__main__": + main() diff --git a/tests/integration/targets/configuration_item_relations/tasks/main.yml b/tests/integration/targets/configuration_item_relations/tasks/main.yml new file mode 100644 index 00000000..05c03d75 --- /dev/null +++ b/tests/integration/targets/configuration_item_relations/tasks/main.yml @@ -0,0 +1,117 @@ +--- +- environment: + SN_HOST: "{{ sn_host }}" + SN_USERNAME: "{{ sn_username }}" + SN_PASSWORD: "{{ sn_password }}" + + block: + - name: Create 3 servers + servicenow.itsm.api: + resource: cmdb_ci_linux_server + action: post + data: + name: "{{ 'linux_' + lookup('password', '/dev/null chars=ascii_letters,digit length=8') | lower }}" + loop: "{{ range(0,3,1)|list }}" + register: servers + + - ansible.builtin.assert: + that: + - servers.results | length == 3 + + - name: Add relation between the first server and the rest -- check mode -- + servicenow.itsm.configuration_item_relations: &create-relations-data + parent_sys_id: "{{ servers.results[0].record.sys_id }}" + parent_classname: cmdb_ci_linux_server + state: present + name: Cools::Cooled By + targets: + - name: "{{ servers.results[1].record.name }}" + sys_id: "{{ servers.results[1].record.sys_id }}" + - name: "{{ servers.results[2].record.name }}" + sys_id: "{{ servers.results[2].record.sys_id }}" + register: relations + check_mode: true + + - ansible.builtin.assert: + that: + - relations.record.outbound_relations | length == 2 + + - name: Make sure the relations don't exist + servicenow.itsm.configuration_item_relations_info: + sys_id: "{{ servers.results[0].record.sys_id }}" + classname: cmdb_ci_linux_server + register: relations + + - ansible.builtin.assert: + that: + - relations.record.outbound_relations | length == 0 + + - name: Add relation between the first server and the rest + servicenow.itsm.configuration_item_relations: *create-relations-data + register: relations + + + - name: Retrieve relations for the first server + servicenow.itsm.configuration_item_relations_info: + sys_id: "{{ servers.results[0].record.sys_id }}" + classname: cmdb_ci_linux_server + register: relations + + - debug: var=relations + + - ansible.builtin.assert: + that: + - relations.record.outbound_relations | length == 2 + - relations.record.inbound_relations | length == 0 + + - name: Remove relation -- check mode -- + servicenow.itsm.configuration_item_relations: &delete-relations-data + parent_sys_id: "{{ servers.results[0].record.sys_id }}" + parent_classname: cmdb_ci_linux_server + state: absent + targets: + - name: "{{ servers.results[1].record.name }}" + sys_id: "{{ servers.results[1].record.sys_id }}" + - name: "{{ servers.results[2].record.name }}" + sys_id: "{{ servers.results[2].record.sys_id }}" + register: deleted_relation + check_mode: true + + - ansible.builtin.assert: + that: + - deleted_relation.record.outbound_relations | length == 0 + + - name: Make sure the relations are still present + servicenow.itsm.configuration_item_relations_info: + sys_id: "{{ servers.results[0].record.sys_id }}" + classname: cmdb_ci_linux_server + register: relations + + - ansible.builtin.assert: + that: + - relations.record.outbound_relations | length == 2 + - relations.record.inbound_relations | length == 0 + + - name: Remove relation + servicenow.itsm.configuration_item_relations: *delete-relations-data + + - name: Retrieve relations for the first server + servicenow.itsm.configuration_item_relations_info: + sys_id: "{{ servers.results[0].record.sys_id }}" + classname: cmdb_ci_linux_server + register: relations + + - ansible.builtin.assert: + that: + - relations.record.outbound_relations | length == 0 + - relations.record.inbound_relations | length == 0 + + - name: Remove servers + servicenow.itsm.api: + resource: cmdb_ci_linux_server + action: delete + sys_id: "{{ item }}" + loop: + - "{{ servers.results[0].record.sys_id }}" + - "{{ servers.results[1].record.sys_id }}" + - "{{ servers.results[2].record.sys_id }}" diff --git a/tests/unit/plugins/conftest.py b/tests/unit/plugins/conftest.py index 85f00d79..5ae951c7 100644 --- a/tests/unit/plugins/conftest.py +++ b/tests/unit/plugins/conftest.py @@ -20,6 +20,9 @@ ProblemClient, ) from ansible_collections.servicenow.itsm.plugins.module_utils.table import TableClient +from ansible_collections.servicenow.itsm.plugins.module_utils.generic import ( + GenericClient, +) @pytest.fixture @@ -34,6 +37,11 @@ def table_client(mocker): return mocker.Mock(spec=TableClient) +@pytest.fixture +def generic_client(mocker): + return mocker.Mock(spec=GenericClient) + + @pytest.fixture def attachment_client(mocker): return mocker.Mock(spec=AttachmentClient) diff --git a/tests/unit/plugins/module_utils/test_cmdb_relations.py b/tests/unit/plugins/module_utils/test_cmdb_relations.py new file mode 100644 index 00000000..fee4f2de --- /dev/null +++ b/tests/unit/plugins/module_utils/test_cmdb_relations.py @@ -0,0 +1,420 @@ +# -*- coding: utf-8 -*- +# Copyright: (c) 2024, Red Hat +# +# 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 + +import pytest +import sys + +from ansible_collections.servicenow.itsm.plugins.module_utils import cmdb_relation + +pytestmark = pytest.mark.skipif( + sys.version_info < (2, 7), reason="requires python2.7 or higher" +) + + +class TestCmdbRelation: + def test_valid_ci(self): + input = dict( + sys_id="relation_id", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ) + + relation = cmdb_relation.CmdbRelation(input) + + assert relation.sys_id == "relation_id" + assert relation.type_id == "relation_sys_id" + assert relation.type_name == "relation_name" + assert relation.target_id == "target_id" + assert relation.target_name == "target_name" + + @pytest.mark.parametrize( + "input", + [ + dict( + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + dict( + sys_id="relation_id", + target=dict(value="target_id", display_value="target_name"), + ), + dict( + sys_id="relation_id", + type=dict(value="relation_sys_id", display_value="relation_name"), + ), + dict( + sys_id="relation_id", + type=[], + target=dict(value="target_id", display_value="target_name"), + ), + dict( + sys_id="relation_id", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=[], + ), + ], + ) + def test_invalid_ci(self, input): + with pytest.raises(ValueError): + cmdb_relation.CmdbRelation(input) + + def test_equal(self): + input = dict( + sys_id="relation_id", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ) + + relation1 = cmdb_relation.CmdbRelation(input) + relation2 = cmdb_relation.CmdbRelation(input) + + assert relation1 == relation2 + + @pytest.mark.parametrize( + "input1,input2", + [ + ( + dict( + sys_id="relation_id", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + dict( + sys_id="relation_id1", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + ), + ( + dict( + sys_id=None, + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + dict( + sys_id=None, + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="another_target_id", display_value="target_name"), + ), + ), + ( + dict( + sys_id=None, + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + dict( + sys_id=None, + type=dict( + value="another_relation_sys_id", display_value="relation_name" + ), + target=dict(value="target_id", display_value="target_name"), + ), + ), + ], + ) + def test_not_equal(self, input1, input2): + + relation1 = cmdb_relation.CmdbRelation(input1) + relation2 = cmdb_relation.CmdbRelation(input2) + + assert relation1 != relation2 + + def test_to_json(self): + input = dict( + sys_id="relation_id", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ) + json_dict = dict( + sys_id="relation_id", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ) + + r = cmdb_relation.CmdbRelation(input) + + assert json_dict == r.to_json() + + def test_to_payload(self): + input = dict( + sys_id="relation_id", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ) + payload = dict( + type="relation_sys_id", + target="target_id", + ) + + r = cmdb_relation.CmdbRelation(input) + + assert payload == r.to_payload() + + +class TestCmdbRelations: + def test_init(self): + ci = dict( + sys_id="relation_id", + outbound_relations=[ + dict( + sys_id="relation_1", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + dict( + sys_id="relation_2", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + ], + ) + + ci_relations = cmdb_relation.CmdbItemRelations(ci) + + assert len(ci_relations.relations) == 2 + for dir, r in ci_relations: + assert dir == "outbound" + assert r.sys_id in ("relation_1", "relation_2") + + def test_init2(self): + ci = dict( + sys_id="relation_id", + outbound_relations=[ + dict( + sys_id="relation_1", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + dict( + sys_id="relation_2", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + ], + inbound_relations=[ + dict( + sys_id="inbound_relation_1", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ) + ], + ) + + ci_relations = cmdb_relation.CmdbItemRelations(ci) + + assert len(ci_relations.relations) == 3 + + # count outbound and inbound relations + count = [0, 0] + for dir, r in ci_relations: + if dir == "outbound": + count[1] += 1 + elif dir == "inbound": + count[0] += 1 + assert count == [1, 2] + + def test_get(self): + ci_relations = cmdb_relation.CmdbItemRelations(get_configuration_item()) + + rel = ci_relations.get("outbound", "target_id") + assert rel is not None + assert rel.sys_id == "relation_1" + + rel1 = ci_relations.get("outbound", "unknown_id") + assert rel1 is None + + rel2 = ci_relations.get("inbound", "inbound_target_id") + assert rel2 is not None + assert rel2.sys_id == "inbound_relation_1" + + def test_add(self): + ci_relations = cmdb_relation.CmdbItemRelations(get_configuration_item()) + + assert len(ci_relations.tainted) == 0 + + ci_relations.add( + "outbound", + dict( + sys_id="inbound_relation_2", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + ) + + assert len(ci_relations.tainted) == 1 + + # add the same relation one more time. + ci_relations.add( + "outbound", + dict( + sys_id="inbound_relation_2", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + ) + + assert len(ci_relations.tainted) == 1 + + def test_remove(self): + ci_relations = cmdb_relation.CmdbItemRelations(get_configuration_item()) + + assert len(ci_relations.tainted) == 0 + + ci_relations.remove( + "outbound", + dict( + sys_id="inbound_relation_2", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + ) + + assert len(ci_relations.tainted) == 1 + + # add the same relation one more time. + ci_relations.remove( + "outbound", + dict( + sys_id="inbound_relation_2", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + ) + + assert len(ci_relations.tainted) == 1 + + def test_clone(self): + original = cmdb_relation.CmdbItemRelations(get_configuration_item()) + + clone = original.clone() + + assert len(clone.relations) == 3 + + for dir, rel in clone: + found = False + for d, r in original: + if d == dir and r == rel: + found = True + assert found + + def test_update(self): + ci_relations = cmdb_relation.CmdbItemRelations(dict(sys_id="ci_1")) + + relation_to_add = cmdb_relation.CmdbRelation( + dict( + sys_id="outbound_relation_2", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ) + ) + + ci_relations.add("outbound", relation_to_add) + assert len(ci_relations.tainted) == 1 + + client_mock = GenericClientMock( + dict( + sys_id="ci_1", + outbound_relations=[ + dict( + sys_id="outbound_relation_2", + type=dict( + value="relation_sys_id", display_value="relation_name" + ), + target=dict(value="target_id", display_value="target_name"), + ) + ], + ) + ) + + updated_ci = ci_relations.update("test_path", client_mock) + + assert updated_ci is not None + assert "outbound_relations" in client_mock.payload + assert client_mock.payload["outbound_relations"] == [ + dict(type="relation_sys_id", target="target_id") + ] + + def test_update2(self): + ci = dict( + sys_id="relation_id", + outbound_relations=[ + dict( + sys_id="relation_1", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + dict( + sys_id="relation_2", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + ], + ) + + ci_relations = cmdb_relation.CmdbItemRelations(ci) + ci_relations.remove( + "oubound", + cmdb_relation.CmdbRelation( + dict( + sys_id="relation_1", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ) + ), + ) + + client_mock = GenericClientMock(None) + updated_ci = ci_relations.update("test_api", client_mock) + + assert updated_ci is not None + assert client_mock.sys_id == "relation_1" + assert len(updated_ci.relations) == 1 + + +def get_configuration_item(): + return dict( + sys_id="relation_id", + outbound_relations=[ + dict( + sys_id="relation_1", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + dict( + sys_id="relation_2", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id", display_value="target_name"), + ), + ], + inbound_relations=[ + dict( + sys_id="inbound_relation_1", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="inbound_target_id", display_value="target_name"), + ) + ], + ) + + +class GenericClientMock: + def __init__(self, response_ci): + self.response_ci = response_ci + self.payload = None + self.api_path = "" + self.sys_id = "" + + def create_record(self, api_path, payload, check_mode, query=None): + self.payload = payload + self.api_path = api_path + return self.response_ci + + def delete_record_by_sys_id(self, api_path, sys_id): + self.api_path = api_path + self.sys_id = sys_id diff --git a/tests/unit/plugins/modules/test_configuraiton_item_relations_info.py b/tests/unit/plugins/modules/test_configuraiton_item_relations_info.py new file mode 100644 index 00000000..31d59ac8 --- /dev/null +++ b/tests/unit/plugins/modules/test_configuraiton_item_relations_info.py @@ -0,0 +1,57 @@ +# -*- coding: utf-8 -*- +# # Copyright: (c) 2024, Red Hat +# +# 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 + +import sys + +import pytest +from ansible_collections.servicenow.itsm.plugins.modules import ( + configuration_item_relations_info, +) + +pytestmark = pytest.mark.skipif( + sys.version_info < (2, 7), reason="requires python2.7 or higher" +) + + +class TestConfigurationInfoRelationsInfo: + def test_add_relations(self, create_module, generic_client): + module = create_module( + params=dict( + instance=dict( + host="https://my.host.name", username="user", password="pass" + ), + sysparm_display_value="true", + sys_id="parent_id", + classname="cmdb_ci_linux_server", + ) + ) + + generic_client.get_by_sys_id.return_value = dict( + inbound_relations=[], + outbound_relations=[ + dict( + sys_id="relation_1", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id_1", display_value="target_name"), + ), + ], + ) + + result = configuration_item_relations_info.run(module, generic_client) + + assert result == dict( + inbound_relations=[], + outbound_relations=[ + dict( + sys_id="relation_1", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id_1", display_value="target_name"), + ), + ], + ) diff --git a/tests/unit/plugins/modules/test_configuration_item_relations.py b/tests/unit/plugins/modules/test_configuration_item_relations.py new file mode 100644 index 00000000..18df4a0e --- /dev/null +++ b/tests/unit/plugins/modules/test_configuration_item_relations.py @@ -0,0 +1,194 @@ +# -*- coding: utf-8 -*- +# # Copyright: (c) 2024, Red Hat +# +# 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 + +import sys + +import pytest +from ansible_collections.servicenow.itsm.plugins.modules import ( + configuration_item_relations, +) + +pytestmark = pytest.mark.skipif( + sys.version_info < (2, 7), reason="requires python2.7 or higher" +) + + +class TestEnsureAbsent: + def test_add_relations(self, create_module, generic_client): + module = create_module( + params=dict( + instance=dict( + host="https://my.host.name", username="user", password="pass" + ), + state="absent", + sysparm_display_value="true", + parent_sys_id="parent_id", + parent_classname="cmdb_ci_linux_server", + name="Cools:Cooled by", + direction="outbound", + targets=[ + dict(sys_id="target_id_1", name="target") + ], + ) + ) + + generic_client.list_records.return_value = [dict(sys_id="relation_1")] + + generic_client.get_by_sys_id.return_value = dict( + sys_class_name="cmdb_ci_linux_server", + sys_id="parent_id", + outbound_relations=[ + dict( + sys_id="relation_1", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id_1", display_value="target_name"), + ), + dict( + sys_id="relation_2", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id_2", display_value="target_name"), + ), + ], + ) + + generic_client.delete_record_by_sys_id.return_value = None + + result = configuration_item_relations.ensure_absent(module, generic_client) + + assert result == ( + True, + dict( + inbound_relations=[], + outbound_relations=[ + dict( + sys_id="relation_2", + type=dict( + value="relation_sys_id", display_value="relation_name" + ), + target=dict(value="target_id_2", display_value="target_name"), + ), + ], + ), + dict( + before=dict( + inbound_relations=[], + outbound_relations=[ + dict( + sys_id="relation_1", + type=dict( + value="relation_sys_id", display_value="relation_name" + ), + target=dict( + value="target_id_1", display_value="target_name" + ), + ), + dict( + sys_id="relation_2", + type=dict( + value="relation_sys_id", display_value="relation_name" + ), + target=dict( + value="target_id_2", display_value="target_name" + ), + ), + ], + ), + after=dict( + inbound_relations=[], + outbound_relations=[ + dict( + sys_id="relation_2", + type=dict( + value="relation_sys_id", display_value="relation_name" + ), + target=dict( + value="target_id_2", display_value="target_name" + ), + ), + ], + ), + ), + ) + + +class TestEnsurePresent: + def test_add_relations(self, create_module, generic_client): + module = create_module( + params=dict( + instance=dict( + host="https://my.host.name", username="user", password="pass" + ), + state="present", + sysparm_display_value="true", + parent_sys_id="parent_id", + parent_classname="cmdb_ci_linux_server", + name="Cools:Cooled by", + direction="outbound", + targets=[ + dict(sys_id="target_id_1", name="target") + ], + ) + ) + + generic_client.list_records.return_value = [dict(sys_id="relation_1")] + + generic_client.get_by_sys_id.return_value = dict( + sys_class_name="cmdb_ci_linux_server", + sys_id="parent_id", + outbound_relations=[], + ) + + generic_client.create_record.return_value = dict( + inbound_relations=[], + outbound_relations=[ + dict( + sys_id="relation_1", + type=dict(value="relation_sys_id", display_value="relation_name"), + target=dict(value="target_id_1", display_value="target_name"), + ), + ], + ) + + result = configuration_item_relations.ensure_present(module, generic_client) + + assert result == ( + True, + dict( + inbound_relations=[], + outbound_relations=[ + dict( + sys_id="relation_1", + type=dict( + value="relation_sys_id", display_value="relation_name" + ), + target=dict(value="target_id_1", display_value="target_name"), + ), + ], + ), + dict( + before=dict( + inbound_relations=[], + outbound_relations=[], + ), + after=dict( + inbound_relations=[], + outbound_relations=[ + dict( + sys_id="relation_1", + type=dict( + value="relation_sys_id", display_value="relation_name" + ), + target=dict( + value="target_id_1", display_value="target_name" + ), + ), + ], + ), + ), + )