Add ipcut filter plugin (#291)

* Add ipcut filter plugin

* Add testcases

* Add testcases

* Fix sanity

* Fix sanity

* Fix sanity

* Fix ansible-lint

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Update ipcut.py

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Fix sanity

* Fix sanity

* iFix sanity

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Ashwini Mhatre <amhatre@amhatre-thinkpadt14sgen2i.pnq.csb>

.github/release-drafter.yml

@@ -1,2 +1,3 @@
+---
 # see https://github.com/ansible-community/devtools
 _extends: ansible-community/devtools

.github/workflows/ack.yml

@@ -1,3 +1,4 @@
+---
 # See https://github.com/ansible-community/devtools/blob/main/.github/workflows/ack.yml
 name: ack
 
@@ -5,7 +6,7 @@ concurrency:
   group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
   cancel-in-progress: true
 
-on:
+on:  # yamllint disable-line rule:truthy
   pull_request_target:
     types: [opened, labeled, unlabeled, synchronize]
 

.github/workflows/codecoverage.yml

@@ -1,7 +1,7 @@
 ---
 name: code_coverage
 
-on:
+on:  # yamllint disable-line rule:truthy
   push:
   pull_request:
     branches: [ main ]

.github/workflows/push.yml

@@ -1,3 +1,4 @@
+---
 # push workflow is shared and expected to perform actions after a merge happens
 # on a maintenance branch (default or release). For example updating the
 # draft release-notes.
@@ -9,7 +10,7 @@ concurrency:
   group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
   cancel-in-progress: true
 
-on:
+on:  # yamllint disable-line rule:truthy
   push:
     # branches to consider in the event; optional, defaults to all
     branches:

.github/workflows/release.yml

@@ -1,6 +1,6 @@
 ---
 name: release
-on:
+on:  # yamllint disable-line rule:truthy
   release:
     types: [published]
 

.github/workflows/test.yml

@@ -1,3 +1,4 @@
+---
 name: tests
 
 concurrency:

.github/workflows/token_refresh.yml

@@ -1,5 +1,6 @@
+---
 name: refresh_automation_hub_token
-on:
+on:  # yamllint disable-line rule:truthy
   schedule:
     - cron: '0 0 * * *'
   workflow_dispatch:

README.md

@@ -31,6 +31,7 @@ Name | Description
 [ansible.utils.index_of](https://github.com/ansible-collections/ansible.utils/blob/main/docs/ansible.utils.index_of_filter.rst)|Find the indices of items in a list matching some criteria
 [ansible.utils.ip4_hex](https://github.com/ansible-collections/ansible.utils/blob/main/docs/ansible.utils.ip4_hex_filter.rst)|This filter is designed to convert IPv4 address to Hexadecimal notation with optional delimiter.
 [ansible.utils.ipaddr](https://github.com/ansible-collections/ansible.utils/blob/main/docs/ansible.utils.ipaddr_filter.rst)|This filter is designed to return the input value if a query is True, else False.
+[ansible.utils.ipcut](https://github.com/ansible-collections/ansible.utils/blob/main/docs/ansible.utils.ipcut_filter.rst)|This filter is designed to get 1st or last few bits of IP address.
 [ansible.utils.ipmath](https://github.com/ansible-collections/ansible.utils/blob/main/docs/ansible.utils.ipmath_filter.rst)|This filter is designed to do simple IP math/arithmetic.
 [ansible.utils.ipsubnet](https://github.com/ansible-collections/ansible.utils/blob/main/docs/ansible.utils.ipsubnet_filter.rst)|This filter can be used to manipulate network subnets in several ways.
 [ansible.utils.ipv4](https://github.com/ansible-collections/ansible.utils/blob/main/docs/ansible.utils.ipv4_filter.rst)|To filter only Ipv4 addresses Ipv4 filter is used.

changelogs/fragments/ipcut.yaml

@@ -0,0 +1,3 @@
+---
+minor_changes:
+  - Add ipcut filter plugin.(https://github.com/ansible-collections/ansible.utils/issues/251)

docs/ansible.utils.ipcut_filter.rst

@@ -0,0 +1,150 @@
+.. _ansible.utils.ipcut_filter:
+
+
+*******************
+ansible.utils.ipcut
+*******************
+
+**This filter is designed to get 1st or last few bits of IP address.**
+
+
+Version added: 2.11.0
+
+.. contents::
+   :local:
+   :depth: 1
+
+
+Synopsis
+--------
+- This filter is designed to fetch 1st or last few bits of Ip address.
+
+
+
+
+Parameters
+----------
+
+.. raw:: html
+
+    <table  border=0 cellpadding=0 class="documentation-table">
+        <tr>
+            <th colspan="1">Parameter</th>
+            <th>Choices/<font color="blue">Defaults</font></th>
+                <th>Configuration</th>
+            <th width="100%">Comments</th>
+        </tr>
+            <tr>
+                <td colspan="1">
+                    <div class="ansibleOptionAnchor" id="parameter-"></div>
+                    <b>amount</b>
+                    <a class="ansibleOptionLink" href="#parameter-" title="Permalink to this option"></a>
+                    <div style="font-size: small">
+                        <span style="color: purple">integer</span>
+                    </div>
+                </td>
+                <td>
+                </td>
+                    <td>
+                    </td>
+                <td>
+                        <div>integer for arithmetic. Example -1,2,3</div>
+                </td>
+            </tr>
+            <tr>
+                <td colspan="1">
+                    <div class="ansibleOptionAnchor" id="parameter-"></div>
+                    <b>value</b>
+                    <a class="ansibleOptionLink" href="#parameter-" title="Permalink to this option"></a>
+                    <div style="font-size: small">
+                        <span style="color: purple">string</span>
+                         / <span style="color: red">required</span>
+                    </div>
+                </td>
+                <td>
+                </td>
+                    <td>
+                    </td>
+                <td>
+                        <div>list of subnets or individual address or any other values input for ip_cut plugin</div>
+                </td>
+            </tr>
+    </table>
+    <br/>
+
+
+
+
+Examples
+--------
+
+.. code-block:: yaml
+
+    #### examples
+    - name: Get first 64 bits of Ipv6 address
+      debug:
+        msg: "{{ '1234:4321:abcd:dcba::17' | ansible.utils.ipcut(64) }}"
+
+    - name: Get last 80 bits of Ipv6 address
+      debug:
+        msg: "{{ '1234:4321:abcd:dcba::17' | ansible.utils.ipcut(-80) }}"
+    # PLAY [IPCUT filter plugin examples] ************************************************************************************************
+
+    # TASK [Get first X bits of Ipv6 address] ********************************************************************************************
+    # ok: [localhost] => {
+    #     "msg": "1234:4321:abcd:dcba"
+    # }
+
+    # TASK [Get last X bits of Ipv6 address] *********************************************************************************************
+    # ok: [localhost] => {
+    #     "msg": "dcba:0:0:0:17"
+    # }
+
+    # PLAY RECAP *************************************************************************************************************************
+    # localhost                  : ok=2    changed=0    unreachable=0    failed=0    skipped=0    rescued=0    ignored=0
+
+
+
+Return Values
+-------------
+Common return values are documented `here <https://docs.ansible.com/ansible/latest/reference_appendices/common_return_values.html#common-return-values>`_, the following are the fields unique to this filter:
+
+.. raw:: html
+
+    <table border=0 cellpadding=0 class="documentation-table">
+        <tr>
+            <th colspan="1">Key</th>
+            <th>Returned</th>
+            <th width="100%">Description</th>
+        </tr>
+            <tr>
+                <td colspan="1">
+                    <div class="ansibleOptionAnchor" id="return-"></div>
+                    <b>data</b>
+                    <a class="ansibleOptionLink" href="#return-" title="Permalink to this return value"></a>
+                    <div style="font-size: small">
+                      <span style="color: purple">string</span>
+                    </div>
+                </td>
+                <td></td>
+                <td>
+                            <div>Returns result of portion of IP.</div>
+                    <br/>
+                </td>
+            </tr>
+    </table>
+    <br/><br/>
+
+
+Status
+------
+
+
+Authors
+~~~~~~~
+
+- Ashwini Mhatre (@amhatre)
+
+
+.. hint::
+    Configuration entries for each entry type have a low to high priority order. For example, a variable that is lower in the list will override a variable that is higher up.

plugins/filter/ipcut.py

@@ -0,0 +1,152 @@
+# -*- coding: utf-8 -*-
+# Copyright 2023 Red Hat
+# GNU General Public License v3.0+
+# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+"""
+filter plugin file for ipaddr filters: ip_cut
+"""
+from __future__ import absolute_import, division, print_function
+
+from functools import partial
+
+from ansible.errors import AnsibleFilterError
+
+from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import (
+    AnsibleArgSpecValidator,
+)
+from ansible_collections.ansible.utils.plugins.plugin_utils.base.ipaddr_utils import _need_netaddr
+
+
+__metaclass__ = type
+
+
+try:
+    from jinja2.filters import pass_environment
+except ImportError:
+    from jinja2.filters import environmentfilter as pass_environment
+
+try:
+    import netaddr
+
+    HAS_NETADDR = True
+except ImportError:
+    # in this case, we'll make the filters return error messages (see bottom)
+    HAS_NETADDR = False
+else:
+
+    class mac_linux(netaddr.mac_unix):
+        pass
+
+    mac_linux.word_fmt = "%.2x"
+
+DOCUMENTATION = """
+    name: ipcut
+    author: Ashwini Mhatre (@amhatre)
+    version_added: "2.11.0"
+    short_description: This filter is designed to get 1st or last few bits of IP address.
+    description:
+        - This filter is designed to fetch 1st or last few bits of Ip address.
+    options:
+        value:
+            description:
+            - list of subnets or individual address or any other values input for ip_cut plugin
+            type: str
+            required: True
+        amount:
+            type: int
+            description: integer for arithmetic. Example -1,2,3
+"""
+
+EXAMPLES = r"""
+#### examples
+- name: Get first 64 bits of Ipv6 address
+  debug:
+    msg: "{{ '1234:4321:abcd:dcba::17' | ansible.utils.ipcut(64) }}"
+
+- name: Get last 80 bits of Ipv6 address
+  debug:
+    msg: "{{ '1234:4321:abcd:dcba::17' | ansible.utils.ipcut(-80) }}"
+# PLAY [IPCUT filter plugin examples] ************************************************************************************************
+
+# TASK [Get first X bits of Ipv6 address] ********************************************************************************************
+# ok: [localhost] => {
+#     "msg": "1234:4321:abcd:dcba"
+# }
+
+# TASK [Get last X bits of Ipv6 address] *********************************************************************************************
+# ok: [localhost] => {
+#     "msg": "dcba:0:0:0:17"
+# }
+
+# PLAY RECAP *************************************************************************************************************************
+# localhost                  : ok=2    changed=0    unreachable=0    failed=0    skipped=0    rescued=0    ignored=0
+"""
+
+RETURN = """
+  data:
+    type: str
+    description:
+      - Returns result of portion of IP.
+"""
+
+
+@pass_environment
+def _ipcut(*args, **kwargs):
+    """Fetch first or last bits of IPV6 address"""
+    keys = ["value", "amount"]
+    data = dict(zip(keys, args[1:]))
+    data.update(kwargs)
+    aav = AnsibleArgSpecValidator(data=data, schema=DOCUMENTATION, name="ipmath")
+    valid, errors, updated_data = aav.validate()
+    if not valid:
+        raise AnsibleFilterError(errors)
+    return ipcut(**updated_data)
+
+
+def ipcut(value, amount):
+    ipv6_oct = []
+    try:
+        ip = netaddr.IPAddress(value)
+        ipv6address = ip.bits().replace(":", "")
+    except (netaddr.AddrFormatError, ValueError):
+        msg = "You must pass a valid IP address; {0} is invalid".format(value)
+        raise AnsibleFilterError(msg)
+
+    if not isinstance(amount, int):
+        msg = ("You must pass an integer for arithmetic; " "{0} is not a valid integer").format(
+            amount,
+        )
+        raise AnsibleFilterError(msg)
+    else:
+        if amount < 0:
+            ipsub = ipv6address[amount:]
+        else:
+            ipsub = ipv6address[0:amount]
+
+    ipsubfinal = []
+
+    for i in range(0, len(ipsub), 16):
+        oct_sub = i + 16
+        ipsubfinal.append(ipsub[i:oct_sub])
+
+    for i in ipsubfinal:
+        x = hex(int(i, 2))
+        ipv6_oct.append(x.replace("0x", ""))
+    return str(":".join(ipv6_oct))
+
+
+class FilterModule(object):
+    """IP address and network manipulation filters"""
+
+    filter_map = {
+        # This filter is designed to fetch first or last bits of IPV6 address
+        "ipcut": _ipcut,
+    }
+
+    def filters(self):
+        """ipcut filter"""
+        if HAS_NETADDR:
+            return self.filter_map
+        else:
+            return dict((f, partial(_need_netaddr, f)) for f in self.filter_map)

plugins/filter/next_nth_usable.py

@@ -129,7 +129,7 @@ def next_nth_usable(value, offset):
     except Exception:
         return False
 
-    if type(offset) != int:
+    if type(offset) is not int:
         raise AnsibleFilterError("Must pass in an integer")
     if v.size > 1:
         first_usable, last_usable = _first_last(v)

plugins/filter/previous_nth_usable.py

@@ -128,7 +128,7 @@ def previous_nth_usable(value, offset):
     except Exception:
         return False
 
-    if type(offset) != int:
+    if type(offset) is not int:
         raise AnsibleFilterError("Must pass in an integer")
     if v.size > 1:
         first_usable, last_usable = _first_last(v)

tests/integration/targets/utils_ipaddr_filter/tasks/ipcut.yaml

@@ -0,0 +1,16 @@
+---
+- name: Get first X bits of Ipv6 address
+  ansible.builtin.set_fact:
+    result1: "{{ '1234:4321:abcd:dcba::17'|ansible.utils.ipcut(64) }}"
+
+- name: Assert result for ipcut.
+  ansible.builtin.assert:
+    that: "{{ result1 == '1234:4321:abcd:dcba' }}"
+
+- name: Get last X bits of Ipv6 address
+  ansible.builtin.set_fact:
+    result1: "{{ '1234:4321:abcd:dcba::17'|ansible.utils.ipcut(-80) }}"
+
+- name: Assert result for ipcut.
+  ansible.builtin.assert:
+    that: "{{ result1 == 'dcba:0:0:0:17' }}"

tests/unit/mock/loader.py

@@ -31,7 +31,7 @@ from ansible.parsing.dataloader import DataLoader
 class DictDataLoader(DataLoader):
     def __init__(self, file_mapping=None):
         file_mapping = {} if file_mapping is None else file_mapping
-        assert type(file_mapping) == dict
+        assert type(file_mapping) is dict
 
         super(DictDataLoader, self).__init__()
 

tests/unit/plugins/filter/test_ipcut.py

@@ -0,0 +1,36 @@
+# -*- coding: utf-8 -*-
+# Copyright 2023 Red Hat
+# GNU General Public License v3.0+
+# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+"""
+Unit test file for ipcut filter plugin
+"""
+
+from __future__ import absolute_import, division, print_function
+
+
+__metaclass__ = type
+
+import unittest
+
+from ansible_collections.ansible.utils.plugins.filter.ipcut import _ipcut
+
+
+class TestIpCut(unittest.TestCase):
+    def setUp(self):
+        pass
+
+    def test_get_last_X_bits(self):
+        """Get last X bits of Ipv6 address"""
+
+        args = ["", "1234:4321:abcd:dcba::17", -80]
+        result = _ipcut(*args)
+        self.assertEqual(result, "dcba:0:0:0:17")
+
+    def test_get_first_X_bits(self):
+        """Get first X bits of Ipv6 address"""
+
+        args = ["", "1234:4321:abcd:dcba::17", 64]
+        result = _ipcut(*args)
+        self.assertEqual(result, "1234:4321:abcd:dcba")