This commit is contained in:
mrwho 2026-05-08 23:14:42 +03:00
parent f9712b45b0
commit d2b3f8d612
100 changed files with 4216 additions and 5001 deletions

2
.gitignore vendored
View file

@ -1,4 +1,4 @@
# ---> Ansible # ---> Ansible
*.retry *.retry
.ansible/ .ansible/
collections/

View file

@ -3,10 +3,12 @@ roles_path = roles
collections_paths = collections collections_paths = collections
filter_plugins = filter_plugins filter_plugins = filter_plugins
retry_files_enabled = False retry_files_enabled = False
stdout_callback = yaml
interpreter_python = auto_silent interpreter_python = auto_silent
host_key_checking = False host_key_checking = False
deprecation_warnings=False result_format = yaml
stdout_callback = default
callback_result_format = yaml
deprecation_warnings = False
[ssh_connection] [ssh_connection]
ssh_args = -o ControlMaster=auto -o ControlPersist=60s -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null ssh_args = -o ControlMaster=auto -o ControlPersist=60s -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null

View file

@ -2,61 +2,28 @@
name: "CI" name: "CI"
concurrency: concurrency:
group: ${{ github.head_ref || github.run_id }}-${{ github.event_name }} group: ${{ github.head_ref || github.run_id }}
cancel-in-progress: true cancel-in-progress: true
on: # yamllint disable-line rule:truthy on: # yamllint disable-line rule:truthy
push: pull_request:
branches: [main]
pull_request_target:
branches: [main] branches: [main]
workflow_dispatch: workflow_dispatch:
schedule: schedule:
- cron: '0 0 * * *' - cron: '0 0 * * *'
jobs: jobs:
sonar:
name: SonarCloud
uses: ansible/ansible-content-actions/.github/workflows/sonarcloud.yaml@main
with:
python_version: "3.12"
test_command: "pytest tests/unit -v --cov-report xml --cov=./ -o 'addopts='"
secrets:
SONAR_TOKEN: ${{ secrets.ANSIBLE_COLLECTIONS_ORG_SONAR_TOKEN_CICD_BOT }}
changelog: changelog:
uses: ansible/ansible-content-actions/.github/workflows/changelog.yaml@main uses: ansible/ansible-content-actions/.github/workflows/changelog.yaml@main
if: github.event_name == 'pull_request_target' if: github.event_name == 'pull_request'
ansible-lint: ansible-lint:
uses: ansible/ansible-content-actions/.github/workflows/ansible_lint.yaml@main uses: ansible/ansible-content-actions/.github/workflows/ansible_lint.yaml@main
sanity: sanity:
uses: ansible/ansible-content-actions/.github/workflows/sanity.yaml@main uses: ansible/ansible-content-actions/.github/workflows/sanity.yaml@main
unit-galaxy: unit-galaxy:
uses: ansible/ansible-content-actions/.github/workflows/unit.yaml@main uses: ansible/ansible-content-actions/.github/workflows/unit.yaml@main
integration: integration:
uses: ansible/ansible-content-actions/.github/workflows/integration.yaml@main uses: ansible/ansible-content-actions/.github/workflows/integration.yaml@main
report-status:
if: ${{ always() && github.event_name == 'schedule' }}
needs:
- changelog
- ansible-lint
- sanity
- unit-galaxy
- integration
uses: ansible/ansible-content-actions/.github/workflows/upload_upstream_results.yaml@main
with:
job_context: ${{ toJSON(needs) }}
component_name: ${{ github.event.repository.name }}
UPLOAD_USER: ${{ vars.PDE_ORG_RESULTS_AGGREGATOR_UPLOAD_USER }}
UPLOAD_URL: ${{ vars.PDE_ORG_RESULTS_AGGREGATOR_UPLOAD_URL }}
secrets:
UPLOAD_PASSWORD: ${{ secrets.PDE_ORG_RESULTS_UPLOAD_PASSWORD }}
all_green: all_green:
if: ${{ always() }} if: ${{ always() }}
needs: needs:
@ -65,8 +32,6 @@ jobs:
- unit-galaxy - unit-galaxy
- ansible-lint - ansible-lint
- integration - integration
- sonar
- report-status
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- run: >- - run: >-
@ -76,7 +41,5 @@ jobs:
'${{ needs.integration.result }}', '${{ needs.integration.result }}',
'${{ needs.sanity.result }}', '${{ needs.sanity.result }}',
'${{ needs.unit-galaxy.result }}', '${{ needs.unit-galaxy.result }}',
'${{ needs.ansible-lint.result }}', '${{ needs.ansible-lint.result }}'
'${{ needs.sonar.result }}',
'${{ needs.report-status.result }}'
])" ])"

View file

@ -135,5 +135,4 @@ dmypy.json
.pyre/ .pyre/
# ide # ide
.tests/integration/inventory .vscode
.vscode/

View file

@ -1,13 +1,13 @@
--- ---
repos: repos:
- repo: https://github.com/ansible-network/collection_prep - repo: https://github.com/ansible-network/collection_prep
rev: da78082ea59b03ad16cd7dbee267f53e8ff50bb5 rev: 1.1.1
hooks: hooks:
# - id: autoversion # removed as being handled by GHA push and release drafter # - id: autoversion # removed as being handled by GHA push and release drafter
- id: update-docs - id: update-docs
- repo: https://github.com/pre-commit/pre-commit-hooks - repo: https://github.com/pre-commit/pre-commit-hooks
rev: v6.0.0 rev: v4.5.0
hooks: hooks:
- id: check-merge-conflict - id: check-merge-conflict
- id: debug-statements - id: debug-statements
@ -16,7 +16,7 @@ repos:
- id: trailing-whitespace - id: trailing-whitespace
- repo: https://github.com/asottile/add-trailing-comma - repo: https://github.com/asottile/add-trailing-comma
rev: v3.2.0 rev: v3.1.0
hooks: hooks:
- id: add-trailing-comma - id: add-trailing-comma
@ -29,18 +29,18 @@ repos:
- prettier-plugin-toml - prettier-plugin-toml
- repo: https://github.com/PyCQA/isort - repo: https://github.com/PyCQA/isort
rev: 6.0.1 rev: 5.13.2
hooks: hooks:
- id: isort - id: isort
name: Sort import statements using isort name: Sort import statements using isort
args: ["--filter-files"] args: ["--filter-files"]
- repo: https://github.com/psf/black - repo: https://github.com/psf/black
rev: 25.1.0 rev: 23.12.1
hooks: hooks:
- id: black - id: black
- repo: https://github.com/pycqa/flake8 - repo: https://github.com/pycqa/flake8
rev: 7.3.0 rev: 7.0.0
hooks: hooks:
- id: flake8 - id: flake8

View file

@ -4,90 +4,6 @@ Ansible Utils Collection Release Notes
.. contents:: Topics .. contents:: Topics
v6.0.2
======
Bugfixes
--------
- cidr_merge - Fix filter failing when used inside a Jinja2 macro called with ``with context`` by unwrapping Ansible lazy template lists before validation.
- cli_parse - Honor ttp_results.results flat_list in TTP parser so output is a single-level list instead of double-wrapped (https://github.com/ansible-collections/ansible.utils/issues/402).
- ipaddress_utils - Support Python 3.14+ by using the public ``version`` attribute instead of the removed private ``_version`` on ``ipaddress`` network objects (bpo-118710).
- update_fact - Use task_vars at top-level instead of the deprecated ``vars`` key for compatibility with ansible-core 2.24 (ansible/ansible issue
v6.0.1
======
Bugfixes
--------
- Add a cleanup step that removes empty {} and [] values from lists in keep_keys_from_dict_n_list()
Documentation Changes
---------------------
- Fix the description of the reduce_on_network filter.
- Fix the module name in ipmath filter.
v6.0.0
======
Release Summary
---------------
With this release, the minimum required version of `ansible-core` for this collection is `2.16.0`. The last version known to be compatible with `ansible-core` versions below `2.16` is v5.1.2.
Major Changes
-------------
- Bumping `requires_ansible` to `>=2.16.0`, since previous ansible-core versions are EoL now.
v5.1.2
======
Bugfixes
--------
- keep_keys - Fixes keep_keys filter to retain the entire node when a key match occurs, rather than just the leaf node values.
v5.1.1
======
Bugfixes
--------
- keep_keys - Fixes issue where all keys are removed when data is passed in as a dict.
v5.1.0
======
Minor Changes
-------------
- Allows the cli_parse module to find parser.template_path inside roles or collections when a path relative to the role/collection directory is provided.
- Fix cli_parse module to require a connection.
- Previously, the ansible.utils.ipcut filter only supported IPv6 addresses, leading to confusing error messages when used with IPv4 addresses. This fix ensures that the filter now appropriately handles both IPv4 and IPv6 addresses.
- Removed conditional check for deprecated ansible.netcommon.cli_parse from ansible.utils.cli_parse
- The from_xml filter returns a python dictionary instead of a json string.
Documentation Changes
---------------------
- Add a wildcard mask/hostmask documentation to ipaddr filter doc page to obtain an IP address's wildcard mask/hostmask.
v5.0.0
======
Release Summary
---------------
With this release, the minimum required version of `ansible-core` for this collection is `2.15.0`. The last version known to be compatible with `ansible-core` versions below `2.15` is v4.1.0.
Major Changes
-------------
- Bumping `requires_ansible` to `>=2.15.0`, since previous ansible-core versions are EoL now.
v4.1.0 v4.1.0
====== ======

File diff suppressed because it is too large Load diff

View file

@ -2,7 +2,7 @@
"collection_info": { "collection_info": {
"namespace": "ansible", "namespace": "ansible",
"name": "utils", "name": "utils",
"version": "6.0.2", "version": "4.1.0",
"authors": [ "authors": [
"Ansible Community" "Ansible Community"
], ],
@ -30,7 +30,7 @@
"name": "FILES.json", "name": "FILES.json",
"ftype": "file", "ftype": "file",
"chksum_type": "sha256", "chksum_type": "sha256",
"chksum_sha256": "19dcf67b45414d5687c1e9811735ab46014a9a7fd8b1f04fbfd1ea26e2baf98f", "chksum_sha256": "364b8a7ea619abce47e2b659b41dd8442aaf031f027a38888c9f0c48ee30181c",
"format": 1 "format": 1
}, },
"format": 1 "format": 1

View file

@ -1,36 +1,16 @@
# Ansible Utilities Collection # Ansible Utilities Collection
[![Codecov](https://codecov.io/gh/ansible-collections/ansible.utils/branch/main/graph/badge.svg)](https://codecov.io/gh/ansible-collections/ansible.utils)
[![CI](https://github.com/ansible-collections/ansible.utils/actions/workflows/tests.yml/badge.svg?branch=main&event=schedule)](https://github.com/ansible-collections/ansible.utils/actions/workflows/tests.yml) [![CI](https://github.com/ansible-collections/ansible.utils/actions/workflows/tests.yml/badge.svg?branch=main&event=schedule)](https://github.com/ansible-collections/ansible.utils/actions/workflows/tests.yml)
The Ansible ``ansible.utils`` collection includes a variety of plugins that aid in the management, manipulation and visibility of data for the Ansible playbook developer. The Ansible ``ansible.utils`` collection includes a variety of plugins that aid in the management, manipulation and visibility of data for the Ansible playbook developer.
## Support
As a Red Hat Ansible [Certified Content](https://catalog.redhat.com/software/search?target_platforms=Red%20Hat%20Ansible%20Automation%20Platform), this collection is entitled to [support](https://access.redhat.com/support/) through [Ansible Automation Platform](https://www.redhat.com/en/technologies/management/ansible) (AAP).
If a support case cannot be opened with Red Hat and the collection has been obtained either from [Galaxy](https://galaxy.ansible.com/ui/) or [GitHub](https://github.com/ansible-collections/ansible.utils), there is community support available at no charge.
You can join us on [#network:ansible.com](https://matrix.to/#/#network:ansible.com) room or the [Ansible Forum Network Working Group](https://forum.ansible.com/g/network-wg).
For more information you can check the communication section below.
## Communication
* Join the Ansible forum:
* [Get Help](https://forum.ansible.com/c/help/6): get help or help others.
* [Posts tagged with 'network'](https://forum.ansible.com/tag/network): subscribe to participate in collection-related conversations.```
* [Ansible Network Automation Working Group](https://forum.ansible.com/g/network-wg/): by joining the team you will automatically get subscribed to the posts tagged with [network](https://forum.ansible.com/tags/network).
* [Social Spaces](https://forum.ansible.com/c/chat/4): gather and interact with fellow enthusiasts.
* [News & Announcements](https://forum.ansible.com/c/news/5): track project-wide announcements including social events.
* The Ansible [Bullhorn newsletter](https://docs.ansible.com/ansible/devel/community/communication.html#the-bullhorn): used to announce releases and important changes.
For more information about communication, see the [Ansible communication guide](https://docs.ansible.com/ansible/devel/community/communication.html).
<!--start requires_ansible--> <!--start requires_ansible-->
## Ansible version compatibility ## Ansible version compatibility
This collection has been tested against the following Ansible versions: **>=2.16.0**. This collection has been tested against following Ansible versions: **>=2.14.0**.
For collections that support Ansible 2.9, please ensure you update your `network_os` to use the
fully qualified collection name (for example, `cisco.ios.ios`).
Plugins and modules within a collection may be tested with only specific Ansible versions. Plugins and modules within a collection may be tested with only specific Ansible versions.
A collection may contain metadata that identifies these versions. A collection may contain metadata that identifies these versions.
PEP440 is the schema used to describe the versions of Ansible. PEP440 is the schema used to describe the versions of Ansible.
@ -172,7 +152,7 @@ Please read and familiarize yourself with this document.
## Release notes ## Release notes
<!--Add a link to a changelog.md file or an external docsite to cover this information. --> <!--Add a link to a changelog.md file or an external docsite to cover this information. -->
Release notes are available [here](CHANGELOG.rst) Release notes are available [here](https://github.com/ansible-collections/ansible.utils/blob/main/changelogs/CHANGELOG.rst)
For automated release announcements refer [here](https://twitter.com/AnsibleContent). For automated release announcements refer [here](https://twitter.com/AnsibleContent).

View file

@ -2,3 +2,5 @@
# see https://docs.openstack.org/infra/bindep/ for additional information. # see https://docs.openstack.org/infra/bindep/ for additional information.
gcc-c++ [doc test platform:rpm] gcc-c++ [doc test platform:rpm]
python3-devel [test platform:rpm]
python3 [test platform:rpm]

View file

@ -436,105 +436,3 @@ releases:
fragments: fragments:
- ipaddress_is_global_fallback.yaml - ipaddress_is_global_fallback.yaml
release_date: "2024-04-15" release_date: "2024-04-15"
5.0.0:
changes:
major_changes:
- Bumping `requires_ansible` to `>=2.15.0`, since previous ansible-core versions
are EoL now.
release_summary:
With this release, the minimum required version of `ansible-core`
for this collection is `2.15.0`. The last version known to be compatible with
`ansible-core` versions below `2.15` is v4.1.0.
fragments:
- bump_215.yaml
release_date: "2024-06-10"
5.1.0:
changes:
doc_changes:
- Add a wildcard mask/hostmask documentation to ipaddr filter doc page to obtain
an IP address's wildcard mask/hostmask.
minor_changes:
- Allows the cli_parse module to find parser.template_path inside roles or collections
when a path relative to the role/collection directory is provided.
- Fix cli_parse module to require a connection.
- Previously, the ansible.utils.ipcut filter only supported IPv6 addresses,
leading to confusing error messages when used with IPv4 addresses. This fix
ensures that the filter now appropriately handles both IPv4 and IPv6 addresses.
- Removed conditional check for deprecated ansible.netcommon.cli_parse from
ansible.utils.cli_parse
- The from_xml filter returns a python dictionary instead of a json string.
fragments:
- 200.yaml
- 203.yaml
- 204.yaml
- 324.yaml
- 358_ipcut.yaml
- add_template_path.yaml
- fix_cli_parse.yaml
- fix_from_xml.yaml
- todo_condition.yml
release_date: "2024-08-05"
5.1.1:
changes:
bugfixes:
- keep_keys - Fixes issue where all keys are removed when data is passed in
as a dict.
fragments:
- 0-readme.yml
- keep_keys.yaml
release_date: "2024-09-05"
5.1.2:
changes:
bugfixes:
- keep_keys - Fixes keep_keys filter to retain the entire node when a key match
occurs, rather than just the leaf node values.
fragments:
- keep_keys_greedy.yaml
release_date: "2024-09-30"
6.0.0:
changes:
major_changes:
- Bumping `requires_ansible` to `>=2.16.0`, since previous ansible-core versions
are EoL now.
release_summary:
With this release, the minimum required version of `ansible-core`
for this collection is `2.16.0`. The last version known to be compatible with
`ansible-core` versions below `2.16` is v5.1.2.
fragments:
- bump_216.yaml
- ipaddr.yaml
- test_ansibleundefined.yaml
release_date: "2025-04-08"
6.0.1:
changes:
bugfixes:
- Add a cleanup step that removes empty {} and [] values from lists in keep_keys_from_dict_n_list()
doc_changes:
- Fix the description of the reduce_on_network filter.
- Fix the module name in ipmath filter.
fragments:
- fix_data_tagging.yaml
- fix_description.yaml
- fix_keep_keys_list_retention.yaml
- fix_module_name.yaml
- fix_sanity.yaml
release_date: "2025-12-31"
6.0.2:
changes:
bugfixes:
- cidr_merge - Fix filter failing when used inside a Jinja2 macro called with
``with context`` by unwrapping Ansible lazy template lists before validation.
- cli_parse - Honor ttp_results.results flat_list in TTP parser so output is
a single-level list instead of double-wrapped (https://github.com/ansible-collections/ansible.utils/issues/402).
- ipaddress_utils - Support Python 3.14+ by using the public ``version`` attribute
instead of the removed private ``_version`` on ``ipaddress`` network objects
(bpo-118710).
- update_fact - Use task_vars at top-level instead of the deprecated ``vars``
key for compatibility with ansible-core 2.24 (ansible/ansible issue
fragments:
- 402-flat_list.yml
- 426-update_fact_task_vars.yml
- cidr_merge-macro_context.yml
- fix_deprecations.yaml
- ipaddress_utils-python314.yml
release_date: "2026-04-09"

View file

@ -19,16 +19,9 @@ Synopsis
-------- --------
- This filter is designed to return the input value if a query is True, and False if a query is False - This filter is designed to return the input value if a query is True, and False if a query is False
- This way it can be easily used in chained filters - This way it can be easily used in chained filters
- For more details on how to use this plugin, please refer to `<docsite/rst/filters_ipaddr.rst>`_
Requirements
------------
The below requirements are needed on the local Ansible controller node that executes this filter.
- netaddr>=0.10.1
Parameters Parameters
---------- ----------

View file

@ -84,32 +84,32 @@ Examples
# Ipmath filter plugin with different arthmetic. # Ipmath filter plugin with different arthmetic.
# Get the next fifth address based on an IP address # Get the next fifth address based on an IP address
- debug: - debug:
msg: "{{ '192.168.1.5' | ansible.utils.ipmath(5) }}" msg: "{{ '192.168.1.5' | ansible.netcommon.ipmath(5) }}"
# Get the tenth previous address based on an IP address # Get the tenth previous address based on an IP address
- debug: - debug:
msg: "{{ '192.168.1.5' | ansible.utils.ipmath(-10) }}" msg: "{{ '192.168.1.5' | ansible.netcommon.ipmath(-10) }}"
# Get the next fifth address using CIDR notation # Get the next fifth address using CIDR notation
- debug: - debug:
msg: "{{ '192.168.1.1/24' | ansible.utils.ipmath(5) }}" msg: "{{ '192.168.1.1/24' | ansible.netcommon.ipmath(5) }}"
# Get the previous fifth address using CIDR notation # Get the previous fifth address using CIDR notation
- debug: - debug:
msg: "{{ '192.168.1.6/24' | ansible.utils.ipmath(-5) }}" msg: "{{ '192.168.1.6/24' | ansible.netcommon.ipmath(-5) }}"
# Get the previous tenth address using cidr notation # Get the previous tenth address using cidr notation
# It returns a address of the previous network range # It returns a address of the previous network range
- debug: - debug:
msg: "{{ '192.168.2.6/24' | ansible.utils.ipmath(-10) }}" msg: "{{ '192.168.2.6/24' | ansible.netcommon.ipmath(-10) }}"
# Get the next tenth address in IPv6 # Get the next tenth address in IPv6
- debug: - debug:
msg: "{{ '2001::1' | ansible.utils.ipmath(10) }}" msg: "{{ '2001::1' | ansible.netcommon.ipmath(10) }}"
# Get the previous tenth address in IPv6 # Get the previous tenth address in IPv6
- debug: - debug:
msg: "{{ '2001::5' | ansible.utils.ipmath(-10) }}" msg: "{{ '2001::5' | ansible.netcommon.ipmath(-10) }}"
# TASK [debug] ********************************************************************************************************** # TASK [debug] **********************************************************************************************************
# ok: [localhost] => { # ok: [localhost] => {

View file

@ -120,7 +120,7 @@ Common return values are documented `here <https://docs.ansible.com/ansible/late
</td> </td>
<td></td> <td></td>
<td> <td>
<div>Returns the filtered list of addresses belonging to the network.</div> <div>Returns whether an address or a network passed as argument is in a network.</div>
<br/> <br/>
</td> </td>
</tr> </tr>

View file

@ -195,9 +195,9 @@ Examples
update_list: [] update_list: []
update: update:
- path: addresses[{{ idx }}].network - path: addresses[{{ idx }}].network
value: "{{ item['raw'] | ansible.utils.ipaddr('network') }}" value: "{{ item['raw'] | ansible.netcommon.ipaddr('network') }}"
- path: addresses[{{ idx }}].prefix - path: addresses[{{ idx }}].prefix
value: "{{ item['raw'] | ansible.utils.ipaddr('prefix') }}" value: "{{ item['raw'] | ansible.netcommon.ipaddr('prefix') }}"
- debug: - debug:
var: update_list var: update_list

View file

@ -340,15 +340,6 @@ If needed, you can extract subnet and prefix information from the 'host/prefix'
# {{ host_prefix | ansible.utils.ipaddr('host/prefix') | ansible.utils.ipaddr('prefix') }} # {{ host_prefix | ansible.utils.ipaddr('host/prefix') | ansible.utils.ipaddr('prefix') }}
[64, 24] [64, 24]
To get the wildcard mask from host_prefix
.. code-block:: jinja
wildcard {{ host_prefix | ansible.utils.ipaddr('hostmask')}}
# from host_prefix '192.0.2.0/24' following will be generated
wildcard 0.0.0.255
Converting subnet masks to CIDR notation Converting subnet masks to CIDR notation
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@ -857,3 +848,7 @@ the filter ``slaac()`` generates an IPv6 address for a given network and a MAC A
Playbook organization by roles Playbook organization by roles
:ref:`playbooks_best_practices` :ref:`playbooks_best_practices`
Tips and tricks for playbooks Tips and tricks for playbooks
`User Mailing List <https://groups.google.com/group/ansible-devel>`_
Have a question? Stop by the google group!
:ref:`communication_irc`
How to join Ansible chat channels

View file

@ -1,2 +1,2 @@
--- ---
requires_ansible: ">=2.16.0" requires_ansible: ">=2.14.0"

View file

@ -16,7 +16,7 @@ import json
from importlib import import_module from importlib import import_module
from ansible.errors import AnsibleActionFail from ansible.errors import AnsibleActionFail
from ansible.module_utils.common.text.converters import to_native, to_text from ansible.module_utils._text import to_native, to_text
from ansible.module_utils.connection import Connection from ansible.module_utils.connection import Connection
from ansible.module_utils.connection import ConnectionError as AnsibleConnectionError from ansible.module_utils.connection import ConnectionError as AnsibleConnectionError
from ansible.plugins.action import ActionBase from ansible.plugins.action import ActionBase
@ -37,7 +37,7 @@ ARGSPEC_CONDITIONALS = {
class ActionModule(ActionBase): class ActionModule(ActionBase):
"""action module""" """action module"""
_requires_connection = True _requires_connection = False
PARSER_CLS_NAME = "CliParser" PARSER_CLS_NAME = "CliParser"
@ -100,6 +100,19 @@ class ActionModule(ActionBase):
""" """
requested_parser = self._task.args.get("parser").get("name") requested_parser = self._task.args.get("parser").get("name")
cref = dict(zip(["corg", "cname", "plugin"], requested_parser.split("."))) cref = dict(zip(["corg", "cname", "plugin"], requested_parser.split(".")))
if cref["cname"] == "netcommon" and cref["plugin"] in [
"json",
"textfsm",
"ttp",
"xml",
]:
cref["cname"] = "utils"
msg = (
"Use 'ansible.utils.{plugin}' for parser name instead of '{requested_parser}'."
" This feature will be removed from 'ansible.netcommon' collection in a release"
" after 2022-11-01".format(plugin=cref["plugin"], requested_parser=requested_parser)
)
self._display.warning(msg)
parserlib = "ansible_collections.{corg}.{cname}.plugins.sub_plugins.cli_parser.{plugin}_parser".format( parserlib = "ansible_collections.{corg}.{cname}.plugins.sub_plugins.cli_parser.{plugin}_parser".format(
**cref, **cref,
@ -113,6 +126,33 @@ class ActionModule(ActionBase):
) )
return parser return parser
except Exception as exc: except Exception as exc:
# TODO: The condition is added to support old sub-plugin structure.
# Remove the if condition after ansible.netcommon.cli_parse module is removed
# from ansible.netcommon collection
if cref["cname"] == "netcommon" and cref["plugin"] in [
"native",
"content_templates",
"ntc",
"pyats",
]:
parserlib = (
"ansible_collections.{corg}.{cname}.plugins.cli_parsers.{plugin}_parser".format(
**cref,
)
)
try:
parsercls = getattr(import_module(parserlib), self.PARSER_CLS_NAME)
parser = parsercls(
task_args=self._task.args,
task_vars=task_vars,
debug=self._debug,
)
return parser
except Exception as exc:
self._result["failed"] = True
self._result["msg"] = "Error loading parser: {err}".format(err=to_native(exc))
return None
self._result["failed"] = True self._result["failed"] = True
self._result["msg"] = "Error loading parser: {err}".format(err=to_native(exc)) self._result["msg"] = "Error loading parser: {err}".format(err=to_native(exc))
return None return None
@ -165,9 +205,6 @@ class ActionModule(ActionBase):
cmd_as_fname = self._task.args.get("parser").get("command").replace(" ", "_") cmd_as_fname = self._task.args.get("parser").get("command").replace(" ", "_")
fname = "{os}_{cmd}.{ext}".format(os=oper_sys, cmd=cmd_as_fname, ext=template_extension) fname = "{os}_{cmd}.{ext}".format(os=oper_sys, cmd=cmd_as_fname, ext=template_extension)
source = self._find_needle("templates", fname) source = self._find_needle("templates", fname)
else:
source = self._task.args.get("parser").get("template_path")
source = self._find_needle("templates", source)
self._debug("template_path in task args updated to {source}".format(source=source)) self._debug("template_path in task args updated to {source}".format(source=source))
self._task.args["parser"]["template_path"] = source self._task.args["parser"]["template_path"] = source

View file

@ -12,7 +12,7 @@ import re
from importlib import import_module from importlib import import_module
from ansible.module_utils.common.text.converters import to_native from ansible.module_utils._text import to_native
from ansible.plugins.action import ActionBase from ansible.plugins.action import ActionBase
from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import ( from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import (

View file

@ -11,10 +11,9 @@ __metaclass__ = type
import ast import ast
import re import re
from collections.abc import MutableMapping, MutableSequence
from ansible.errors import AnsibleActionFail from ansible.errors import AnsibleActionFail
from ansible.module_utils.common.text.converters import to_native from ansible.module_utils._text import to_native
from ansible.module_utils.common._collections_compat import MutableMapping, MutableSequence
from ansible.plugins.action import ActionBase from ansible.plugins.action import ActionBase
from jinja2 import Template, TemplateSyntaxError from jinja2 import Template, TemplateSyntaxError
@ -163,30 +162,23 @@ class ActionModule(ActionBase):
self._result["changed"] = False self._result["changed"] = False
self._check_argspec() self._check_argspec()
results = set() results = set()
full_replaces = set() # keys that were fully replaced (no path)
self._ensure_valid_jinja() self._ensure_valid_jinja()
# Use task_vars (top-level) instead of task_vars["vars"] to avoid the
# deprecated internal "vars" dictionary (ansible-core 2.24, issue #426).
for entry in self._task.args["updates"]: for entry in self._task.args["updates"]:
parts = self._field_split(entry["path"]) parts = self._field_split(entry["path"])
obj, path = parts[0], parts[1:] obj, path = parts[0], parts[1:]
results.add(obj) results.add(obj)
if obj not in task_vars: if obj not in task_vars["vars"]:
msg = "'{obj}' was not found in the current facts.".format(obj=obj) msg = "'{obj}' was not found in the current facts.".format(obj=obj)
raise AnsibleActionFail(msg) raise AnsibleActionFail(msg)
retrieved = task_vars.get(obj) retrieved = task_vars["vars"].get(obj)
if path: if path:
self.set_value(retrieved, path, entry["value"]) self.set_value(retrieved, path, entry["value"])
else: else:
if retrieved != entry["value"]: if task_vars["vars"][obj] != entry["value"]:
self._result.setdefault("ansible_facts", {})[obj] = entry["value"] task_vars["vars"][obj] = entry["value"]
full_replaces.add(obj)
self._result["changed"] = True self._result["changed"] = True
for key in results: for key in results:
if key in full_replaces: value = task_vars["vars"].get(key)
value = self._result.get("ansible_facts", {}).get(key)
else:
value = task_vars.get(key)
self._result[key] = value self._result[key] = value
return self._result return self._result

View file

@ -12,7 +12,7 @@ from __future__ import absolute_import, division, print_function
__metaclass__ = type __metaclass__ = type
from ansible.errors import AnsibleActionFail, AnsibleError from ansible.errors import AnsibleActionFail, AnsibleError
from ansible.module_utils.common.text.converters import to_text from ansible.module_utils._text import to_text
from ansible.plugins.action import ActionBase from ansible.plugins.action import ActionBase
from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import ( from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import (

View file

@ -129,26 +129,12 @@ RETURN = """
""" """
def _unwrap_lazy_list(value):
"""
If value is an Ansible lazy template list, return a list of raw elements without
triggering templar.template() resolution. This avoids Template.__new__() errors
when the filter is used inside a Jinja2 macro called with ``with context``.
"""
if value is not None and hasattr(value, "_yield_non_lazy_list_items"):
return list(value._yield_non_lazy_list_items())
return value
@pass_environment @pass_environment
def _cidr_merge(*args, **kwargs): def _cidr_merge(*args, **kwargs):
"""Merge CIDRs; compatible with pipe syntax (value | cidr_merge).""" """Convert the given data from json to xml."""
keys = ["value", "action"] keys = ["value", "action"]
data = dict(zip(keys, args[1:])) data = dict(zip(keys, args[1:]))
data.update(kwargs) data.update(kwargs)
# Unwrap lazy list when present to avoid resolution bug in macro context (no template.j2 change).
if "value" in data:
data["value"] = _unwrap_lazy_list(data["value"])
aav = AnsibleArgSpecValidator(data=data, schema=DOCUMENTATION, name="cidr_merge") aav = AnsibleArgSpecValidator(data=data, schema=DOCUMENTATION, name="cidr_merge")
valid, errors, updated_data = aav.validate() valid, errors, updated_data = aav.validate()
if not valid: if not valid:

View file

@ -172,7 +172,7 @@ RETURN = """
- Returns diff between before and after facts. - Returns diff between before and after facts.
""" """
from ansible.errors import AnsibleFilterError from ansible.errors import AnsibleFilterError
from ansible.module_utils.common.text.converters import to_text from ansible.module_utils._text import to_text
from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import ( from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import (
AnsibleArgSpecValidator, AnsibleArgSpecValidator,

View file

@ -51,7 +51,6 @@ DOCUMENTATION = """
description: description:
- This filter is designed to return the input value if a query is True, and False if a query is False - This filter is designed to return the input value if a query is True, and False if a query is False
- This way it can be easily used in chained filters - This way it can be easily used in chained filters
- For more details on how to use this plugin, please refer to `<docsite/rst/filters_ipaddr.rst>`_
options: options:
value: value:
description: description:
@ -76,8 +75,6 @@ DOCUMENTATION = """
type: str type: str
description: type of filter. example ipaddr, ipv4, ipv6, ipwrap description: type of filter. example ipaddr, ipv4, ipv6, ipwrap
notes: notes:
requirements:
- netaddr>=0.10.1
""" """
EXAMPLES = r""" EXAMPLES = r"""

View file

@ -105,15 +105,10 @@ def _ipcut(*args, **kwargs):
def ipcut(value, amount): def ipcut(value, amount):
ipv6_oct = []
try: try:
ip = netaddr.IPAddress(value) ip = netaddr.IPAddress(value)
if ip.version == 6: ipv6address = ip.bits().replace(":", "")
ip_bits = ip.bits().replace(":", "")
elif ip.version == 4:
ip_bits = ip.bits().replace(".", "")
else:
msg = "Unknown IP Address Version: {0}".format(ip.version)
raise AnsibleFilterError(msg)
except (netaddr.AddrFormatError, ValueError): except (netaddr.AddrFormatError, ValueError):
msg = "You must pass a valid IP address; {0} is invalid".format(value) msg = "You must pass a valid IP address; {0} is invalid".format(value)
raise AnsibleFilterError(msg) raise AnsibleFilterError(msg)
@ -125,27 +120,20 @@ def ipcut(value, amount):
raise AnsibleFilterError(msg) raise AnsibleFilterError(msg)
else: else:
if amount < 0: if amount < 0:
ipsub = ip_bits[amount:] ipsub = ipv6address[amount:]
else: else:
ipsub = ip_bits[0:amount] ipsub = ipv6address[0:amount]
ipsubfinal = []
if ip.version == 6:
ipv4_oct = []
for i in range(0, len(ipsub), 16): for i in range(0, len(ipsub), 16):
oct_sub = i + 16 oct_sub = i + 16
ipv4_oct.append( ipsubfinal.append(ipsub[i:oct_sub])
hex(int(ipsub[i:oct_sub], 2)).replace("0x", ""),
) for i in ipsubfinal:
result = str(":".join(ipv4_oct)) x = hex(int(i, 2))
else: # ip.version == 4: ipv6_oct.append(x.replace("0x", ""))
ipv4_oct = [] return str(":".join(ipv6_oct))
for i in range(0, len(ipsub), 8):
oct_sub = i + 8
ipv4_oct.append(
str(int(ipsub[i:oct_sub], 2)),
)
result = str(".".join(ipv4_oct))
return result
class FilterModule(object): class FilterModule(object):

View file

@ -63,32 +63,32 @@ EXAMPLES = r"""
# Ipmath filter plugin with different arthmetic. # Ipmath filter plugin with different arthmetic.
# Get the next fifth address based on an IP address # Get the next fifth address based on an IP address
- debug: - debug:
msg: "{{ '192.168.1.5' | ansible.utils.ipmath(5) }}" msg: "{{ '192.168.1.5' | ansible.netcommon.ipmath(5) }}"
# Get the tenth previous address based on an IP address # Get the tenth previous address based on an IP address
- debug: - debug:
msg: "{{ '192.168.1.5' | ansible.utils.ipmath(-10) }}" msg: "{{ '192.168.1.5' | ansible.netcommon.ipmath(-10) }}"
# Get the next fifth address using CIDR notation # Get the next fifth address using CIDR notation
- debug: - debug:
msg: "{{ '192.168.1.1/24' | ansible.utils.ipmath(5) }}" msg: "{{ '192.168.1.1/24' | ansible.netcommon.ipmath(5) }}"
# Get the previous fifth address using CIDR notation # Get the previous fifth address using CIDR notation
- debug: - debug:
msg: "{{ '192.168.1.6/24' | ansible.utils.ipmath(-5) }}" msg: "{{ '192.168.1.6/24' | ansible.netcommon.ipmath(-5) }}"
# Get the previous tenth address using cidr notation # Get the previous tenth address using cidr notation
# It returns a address of the previous network range # It returns a address of the previous network range
- debug: - debug:
msg: "{{ '192.168.2.6/24' | ansible.utils.ipmath(-10) }}" msg: "{{ '192.168.2.6/24' | ansible.netcommon.ipmath(-10) }}"
# Get the next tenth address in IPv6 # Get the next tenth address in IPv6
- debug: - debug:
msg: "{{ '2001::1' | ansible.utils.ipmath(10) }}" msg: "{{ '2001::1' | ansible.netcommon.ipmath(10) }}"
# Get the previous tenth address in IPv6 # Get the previous tenth address in IPv6
- debug: - debug:
msg: "{{ '2001::5' | ansible.utils.ipmath(-10) }}" msg: "{{ '2001::5' | ansible.netcommon.ipmath(-10) }}"
# TASK [debug] ********************************************************************************************************** # TASK [debug] **********************************************************************************************************
# ok: [localhost] => { # ok: [localhost] => {

View file

@ -24,7 +24,7 @@ from ansible_collections.ansible.utils.plugins.plugin_utils.base.ipaddr_utils im
__metaclass__ = type __metaclass__ = type
from ansible.module_utils.common.text.converters import to_text from ansible.module_utils._text import to_text
try: try:

View file

@ -118,23 +118,17 @@ def next_nth_usable(value, offset):
Returns the next nth usable ip within a network described by value. Returns the next nth usable ip within a network described by value.
""" """
try: try:
v = None
vtype = ipaddr(value, "type") vtype = ipaddr(value, "type")
if vtype == "address": if vtype == "address":
v = ipaddr(value, "cidr") v = ipaddr(value, "cidr")
elif vtype == "network": elif vtype == "network":
v = ipaddr(value, "subnet") v = ipaddr(value, "subnet")
if v is not None:
v = netaddr.IPNetwork(v) v = netaddr.IPNetwork(v)
else:
return False
except Exception: except Exception:
return False return False
if not isinstance(offset, int): if not isinstance(offset, int):
raise AnsibleFilterError("Must pass in an integer") raise AnsibleFilterError("Must pass in an integer")
if v.size > 1: if v.size > 1:
first_usable, last_usable = _first_last(v) first_usable, last_usable = _first_last(v)
nth_ip = int(netaddr.IPAddress(int(v.ip) + offset)) nth_ip = int(netaddr.IPAddress(int(v.ip) + offset))

View file

@ -117,18 +117,13 @@ def previous_nth_usable(value, offset):
Returns the previous nth usable ip within a network described by value. Returns the previous nth usable ip within a network described by value.
""" """
try: try:
v = None
vtype = ipaddr(value, "type") vtype = ipaddr(value, "type")
if vtype == "address": if vtype == "address":
v = ipaddr(value, "cidr") v = ipaddr(value, "cidr")
elif vtype == "network": elif vtype == "network":
v = ipaddr(value, "subnet") v = ipaddr(value, "subnet")
if v is not None:
v = netaddr.IPNetwork(v) v = netaddr.IPNetwork(v)
else:
return False
except Exception: except Exception:
return False return False

View file

@ -85,7 +85,7 @@ RETURN = """
data: data:
type: bool type: bool
description: description:
- Returns the filtered list of addresses belonging to the network. - Returns whether an address or a network passed as argument is in a network.
""" """

View file

@ -20,10 +20,6 @@ from ansible_collections.ansible.utils.plugins.plugin_utils.base.utils import _v
__metaclass__ = type __metaclass__ = type
from ansible.errors import AnsibleFilterError
from ansible.module_utils.common.text.converters import to_text
DOCUMENTATION = """ DOCUMENTATION = """
name: usable_range name: usable_range
author: Priyam Sahoo (@priyamsahoo) author: Priyam Sahoo (@priyamsahoo)
@ -152,6 +148,10 @@ RETURN = """
- List of usable IP addresses under the key C(usable_ips) - List of usable IP addresses under the key C(usable_ips)
""" """
from ansible.errors import AnsibleFilterError
from ansible.module_utils.common.text.converters import to_text
from ansible.module_utils.six import ensure_text
@_need_ipaddress @_need_ipaddress
def _usable_range(ip): def _usable_range(ip):
@ -162,11 +162,11 @@ def _usable_range(ip):
try: try:
if ip_network(ip).version == 4: if ip_network(ip).version == 4:
ips = [to_text(usable_ips) for usable_ips in IPv4Network(to_text(ip))] ips = [to_text(usable_ips) for usable_ips in IPv4Network(ensure_text(ip))]
no_of_ips = IPv4Network(to_text(ip)).num_addresses no_of_ips = IPv4Network(ensure_text(ip)).num_addresses
if ip_network(ip).version == 6: if ip_network(ip).version == 6:
ips = [to_text(usable_ips) for usable_ips in IPv6Network(to_text(ip))] ips = [to_text(usable_ips) for usable_ips in IPv6Network(ensure_text(ip))]
no_of_ips = IPv6Network(to_text(ip)).num_addresses no_of_ips = IPv6Network(ensure_text(ip)).num_addresses
except Exception as e: except Exception as e:
raise AnsibleFilterError( raise AnsibleFilterError(

View file

@ -70,7 +70,7 @@ RETURN = """
""" """
from ansible.errors import AnsibleError, AnsibleFilterError from ansible.errors import AnsibleError, AnsibleFilterError
from ansible.module_utils.common.text.converters import to_text from ansible.module_utils._text import to_text
from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import ( from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import (
check_argspec, check_argspec,

View file

@ -82,7 +82,7 @@ RETURN = """
""" """
from ansible.errors import AnsibleError, AnsibleLookupError from ansible.errors import AnsibleError, AnsibleLookupError
from ansible.module_utils.common.text.converters import to_text from ansible.module_utils._text import to_text
from ansible.plugins.lookup import LookupBase from ansible.plugins.lookup import LookupBase
from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import ( from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import (

View file

@ -27,6 +27,7 @@ __metaclass__ = type
import re import re
from ansible.module_utils.basic import AnsibleModule from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.six import iteritems
from ansible_collections.ansible.utils.plugins.module_utils.common.utils import dict_merge from ansible_collections.ansible.utils.plugins.module_utils.common.utils import dict_merge
@ -174,7 +175,7 @@ class AnsibleArgSpecValidator:
:type temp_schema: dict :type temp_schema: dict
""" """
options_obj = doc_obj.get("options") options_obj = doc_obj.get("options")
for okey, ovalue in options_obj.items(): for okey, ovalue in iteritems(options_obj):
temp_schema[okey] = {} temp_schema[okey] = {}
for metakey in list(ovalue): for metakey in list(ovalue):
if metakey == "suboptions": if metakey == "suboptions":

View file

@ -14,7 +14,7 @@ __metaclass__ = type
import re import re
from collections.abc import Mapping, MutableMapping from ansible.module_utils.common._collections_compat import Mapping, MutableMapping
def to_paths(var, prepend, wantlist): def to_paths(var, prepend, wantlist):

View file

@ -8,9 +8,11 @@ from __future__ import absolute_import, division, print_function
__metaclass__ = type __metaclass__ = type
from collections.abc import Mapping
from copy import deepcopy from copy import deepcopy
from ansible.module_utils.common._collections_compat import Mapping
from ansible.module_utils.six import iteritems
def sort_list(val): def sort_list(val):
if isinstance(val, list): if isinstance(val, list):
@ -55,7 +57,7 @@ def dict_merge(base, other):
combined = dict() combined = dict()
for key, value in deepcopy(base).items(): for key, value in iteritems(deepcopy(base)):
if isinstance(value, dict): if isinstance(value, dict):
if key in other: if key in other:
item = other.get(key) item = other.get(key)

View file

@ -144,9 +144,9 @@ EXAMPLES = r"""
update_list: [] update_list: []
update: update:
- path: addresses[{{ idx }}].network - path: addresses[{{ idx }}].network
value: "{{ item['raw'] | ansible.utils.ipaddr('network') }}" value: "{{ item['raw'] | ansible.netcommon.ipaddr('network') }}"
- path: addresses[{{ idx }}].prefix - path: addresses[{{ idx }}].prefix
value: "{{ item['raw'] | ansible.utils.ipaddr('prefix') }}" value: "{{ item['raw'] | ansible.netcommon.ipaddr('prefix') }}"
- debug: - debug:
var: update_list var: update_list

View file

@ -1,7 +1,6 @@
""" """
The base class for cli_parsers The base class for cli_parsers
""" """
from __future__ import absolute_import, division, print_function from __future__ import absolute_import, division, print_function

View file

@ -17,7 +17,7 @@ from functools import wraps
from ansible import errors from ansible import errors
from ansible.errors import AnsibleError from ansible.errors import AnsibleError
from ansible.module_utils.basic import missing_required_lib from ansible.module_utils.basic import missing_required_lib
from ansible.module_utils.common.text.converters import to_text from ansible.module_utils.six import ensure_text
from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import ( from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import (
check_argspec, check_argspec,
@ -38,7 +38,7 @@ def ip_network(ip):
if not HAS_IPADDRESS: if not HAS_IPADDRESS:
raise AnsibleError(missing_required_lib("ipaddress")) raise AnsibleError(missing_required_lib("ipaddress"))
return ipaddress.ip_network(to_text(ip)) return ipaddress.ip_network(ensure_text(ip))
def ip_address(ip): def ip_address(ip):
@ -47,7 +47,7 @@ def ip_address(ip):
if not HAS_IPADDRESS: if not HAS_IPADDRESS:
raise AnsibleError(missing_required_lib("ipaddress")) raise AnsibleError(missing_required_lib("ipaddress"))
return ipaddress.ip_address(to_text(ip)) return ipaddress.ip_address(ensure_text(ip))
def _need_ipaddress(func): def _need_ipaddress(func):
@ -60,29 +60,15 @@ def _need_ipaddress(func):
return wrapper return wrapper
def _get_network_version(network):
"""
Python 3.14 changes the _version attribute to version, so we have to try both.
"""
if hasattr(network, "_version"):
return network._version
return network.version
def _is_subnet_of(network_a, network_b): def _is_subnet_of(network_a, network_b):
"""
Return True if network_a is a subnet of network_b (same logic as ipaddress).
Uses the public .version attribute for compatibility with Python 3.14+ where
the private _version was removed (see bpo-118710 / cpython@c530ce1).
"""
try: try:
if _get_network_version(network_a) != _get_network_version(network_b): if network_a._version != network_b._version:
return False return False
return ( return (
network_b.network_address <= network_a.network_address network_b.network_address <= network_a.network_address
and network_b.broadcast_address >= network_a.broadcast_address and network_b.broadcast_address >= network_a.broadcast_address
) )
except (AttributeError, TypeError): except Exception:
return False return False

View file

@ -1,7 +1,6 @@
""" """
The base class for validator The base class for validator
""" """
from __future__ import absolute_import, division, print_function from __future__ import absolute_import, division, print_function
@ -12,7 +11,8 @@ import os
from importlib import import_module from importlib import import_module
from ansible.errors import AnsibleError from ansible.errors import AnsibleError
from ansible.module_utils.common.text.converters import to_native, to_text from ansible.module_utils._text import to_native, to_text
from ansible.module_utils.six import iteritems
from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import ( from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import (
check_argspec, check_argspec,
@ -74,7 +74,7 @@ class ValidateBase(object):
if not options: if not options:
return None return None
for option_name, option_value in options.items(): for option_name, option_value in iteritems(options):
option_var_name_list = option_value.get("vars", []) option_var_name_list = option_value.get("vars", [])
option_env_name_list = option_value.get("env", []) option_env_name_list = option_value.get("env", [])
@ -181,10 +181,10 @@ def _load_validator(engine, data, criteria, plugin_vars=None, cls_name="Validate
return validator, result return validator, result
except Exception as exc: except Exception as exc:
result["failed"] = True result["failed"] = True
result["msg"] = ( result[
"For engine '{engine}' error loading the corresponding validate plugin: {err}".format( "msg"
] = "For engine '{engine}' error loading the corresponding validate plugin: {err}".format(
engine=engine, engine=engine,
err=to_native(exc), err=to_native(exc),
) )
)
return None, result return None, result

View file

@ -13,6 +13,8 @@ from __future__ import absolute_import, division, print_function
__metaclass__ = type __metaclass__ = type
import json
from ansible.errors import AnsibleFilterError from ansible.errors import AnsibleFilterError
@ -44,7 +46,7 @@ def from_xml(data, engine):
if not HAS_XMLTODICT: if not HAS_XMLTODICT:
_raise_error("Missing required library xmltodict") _raise_error("Missing required library xmltodict")
try: try:
res = xmltodict.parse(data, dict_constructor=dict) res = json.dumps(xmltodict.parse(data))
except Exception: except Exception:
_raise_error("Input Xml is not valid") _raise_error("Input Xml is not valid")
return res return res

View file

@ -14,7 +14,8 @@ __metaclass__ = type
import json import json
from ansible.module_utils.common.text.converters import to_native from ansible.module_utils._text import to_native
from ansible.module_utils.six import integer_types, string_types
from jinja2.exceptions import TemplateSyntaxError from jinja2.exceptions import TemplateSyntaxError
@ -157,7 +158,7 @@ def index_of(
if result: if result:
res.append(idx) res.append(idx)
elif isinstance(key, (str, int, bool)): elif isinstance(key, (string_types, integer_types, bool)):
if not all(isinstance(entry, dict) for entry in data): if not all(isinstance(entry, dict) for entry in data):
all_tipes = [type(_to_well_known_type(entry)).__name__ for entry in data] all_tipes = [type(_to_well_known_type(entry)).__name__ for entry in data]
msg = ( msg = (

View file

@ -1,3 +1,4 @@
#
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# Copyright 2021 Red Hat # Copyright 2021 Red Hat
# GNU General Public License v3.0+ # GNU General Public License v3.0+
@ -30,34 +31,37 @@ def _raise_error(msg):
def keep_keys_from_dict_n_list(data, target, matching_parameter): def keep_keys_from_dict_n_list(data, target, matching_parameter):
if isinstance(data, list): if isinstance(data, list):
list_data = [keep_keys_from_dict_n_list(each, target, matching_parameter) for each in data] list_data = [keep_keys_from_dict_n_list(each, target, matching_parameter) for each in data]
return [r for r in list_data if r not in ([], {}, None)] return list_data
if isinstance(data, dict): if isinstance(data, dict):
keep = {} keep = {}
for k, val in data.items(): for k, val in data.items():
key_matches = any( for key in target:
( match = None
k == t if not isinstance(val, (list, dict)):
if matching_parameter == "equality" if matching_parameter == "regex":
else ( match = re.match(key, k)
(re.match(t, k) is not None) if match:
if matching_parameter == "regex"
else (
k.startswith(t)
if matching_parameter == "starts_with"
else k.endswith(t)
)
)
)
for t in target
)
if key_matches:
keep[k] = val keep[k] = val
elif isinstance(val, (list, dict)): elif matching_parameter == "starts_with":
nested_keep = keep_keys_from_dict_n_list(val, target, matching_parameter) if k.startswith(key):
if nested_keep not in ([], {}, None): keep[k], match = val, True
keep[k] = nested_keep elif matching_parameter == "ends_with":
if k.endswith(key):
keep[k], match = val, True
else:
if k == key:
keep[k], match = val, True
else:
list_data = keep_keys_from_dict_n_list(val, target, matching_parameter)
if isinstance(list_data, list) and not match:
list_data = [r for r in list_data if r not in ([], {})]
if all(isinstance(s, str) for s in list_data):
continue
if list_data in ([], {}):
continue
keep[k] = list_data
return keep return keep
return None return data
def keep_keys(data, target, matching_parameter="equality"): def keep_keys(data, target, matching_parameter="equality"):

View file

@ -3,7 +3,6 @@ json parser
This is the json parser for use with the cli_parse module and action plugin This is the json parser for use with the cli_parse module and action plugin
""" """
from __future__ import absolute_import, division, print_function from __future__ import absolute_import, division, print_function
@ -38,7 +37,8 @@ EXAMPLES = r"""
import json import json
from ansible.module_utils.common.text.converters import to_native from ansible.module_utils._text import to_native
from ansible.module_utils.six import string_types
from ansible_collections.ansible.utils.plugins.plugin_utils.base.cli_parser import CliParserBase from ansible_collections.ansible.utils.plugins.plugin_utils.base.cli_parser import CliParserBase
@ -66,7 +66,7 @@ class CliParser(CliParserBase):
""" """
text = self._task_args.get("text") text = self._task_args.get("text")
try: try:
if not isinstance(text, str): if not isinstance(text, string_types):
text = json.dumps(text) text = json.dumps(text)
parsed = json.loads(text) parsed = json.loads(text)
except Exception as exc: except Exception as exc:

View file

@ -4,7 +4,6 @@ textfsm parser
This is the textfsm parser for use with the cli_parse module and action plugin This is the textfsm parser for use with the cli_parse module and action plugin
https://github.com/google/textfsm https://github.com/google/textfsm
""" """
from __future__ import absolute_import, division, print_function from __future__ import absolute_import, division, print_function
@ -40,8 +39,8 @@ EXAMPLES = r"""
import os import os
from ansible.module_utils._text import to_native
from ansible.module_utils.basic import missing_required_lib from ansible.module_utils.basic import missing_required_lib
from ansible.module_utils.common.text.converters import to_native
from ansible_collections.ansible.utils.plugins.plugin_utils.base.cli_parser import CliParserBase from ansible_collections.ansible.utils.plugins.plugin_utils.base.cli_parser import CliParserBase

View file

@ -4,7 +4,6 @@ ttp parser
This is the ttp parser for use with the cli_parse module and action plugin This is the ttp parser for use with the cli_parse module and action plugin
https://github.com/dmulyalin/ttp https://github.com/dmulyalin/ttp
""" """
from __future__ import absolute_import, division, print_function from __future__ import absolute_import, division, print_function
@ -40,8 +39,8 @@ EXAMPLES = r"""
import os import os
from ansible.module_utils._text import to_native
from ansible.module_utils.basic import missing_required_lib from ansible.module_utils.basic import missing_required_lib
from ansible.module_utils.common.text.converters import to_native
from ansible_collections.ansible.utils.plugins.plugin_utils.base.cli_parser import CliParserBase from ansible_collections.ansible.utils.plugins.plugin_utils.base.cli_parser import CliParserBase
@ -119,23 +118,7 @@ class CliParser(CliParserBase):
if parser_param.get("vars") if parser_param.get("vars")
else {} else {}
) )
# TTP's result() uses keyword "structure", but we accept "results" from users results = parser.result(**ttp_results)
result_kwargs = dict(ttp_results)
if "results" in result_kwargs:
result_kwargs["structure"] = result_kwargs.pop("results")
results = parser.result(**result_kwargs)
# When flat_list is requested, flatten one level if we got [[...]]
requested_flat = (
ttp_results.get("results") == "flat_list"
or ttp_results.get("structure") == "flat_list"
)
if (
requested_flat
and isinstance(results, list)
and len(results) == 1
and isinstance(results[0], list)
):
results = results[0]
except Exception as exc: except Exception as exc:
msg = "Template Text Parser returned an error while parsing. Error: {err}" msg = "Template Text Parser returned an error while parsing. Error: {err}"
return {"errors": [msg.format(err=to_native(exc))]} return {"errors": [msg.format(err=to_native(exc))]}

View file

@ -4,7 +4,6 @@ xml parser
This is the xml parser for use with the cli_parse module and action plugin This is the xml parser for use with the cli_parse module and action plugin
https://github.com/martinblech/xmltodict https://github.com/martinblech/xmltodict
""" """
from __future__ import absolute_import, division, print_function from __future__ import absolute_import, division, print_function
@ -38,8 +37,8 @@ EXAMPLES = r"""
register: nxos_xml_text register: nxos_xml_text
""" """
from ansible.module_utils._text import to_native
from ansible.module_utils.basic import missing_required_lib from ansible.module_utils.basic import missing_required_lib
from ansible.module_utils.common.text.converters import to_native
from ansible_collections.ansible.utils.plugins.plugin_utils.base.cli_parser import CliParserBase from ansible_collections.ansible.utils.plugins.plugin_utils.base.cli_parser import CliParserBase

View file

@ -9,7 +9,7 @@ from __future__ import absolute_import, division, print_function
__metaclass__ = type __metaclass__ = type
DOCUMENTATION = """ DOCUMENTATION = """
author: Katherine Case (@Qalthos) author: Nathaniel Case (@Qalthos)
name: config name: config
short_description: Define configurable options for configuration validate plugin short_description: Define configurable options for configuration validate plugin
description: description:
@ -51,7 +51,8 @@ import re
from io import StringIO from io import StringIO
from ansible.errors import AnsibleError from ansible.errors import AnsibleError
from ansible.module_utils.common.text.converters import to_text from ansible.module_utils._text import to_text
from ansible.module_utils.six import string_types
from ansible_collections.ansible.utils.plugins.module_utils.common.utils import to_list from ansible_collections.ansible.utils.plugins.module_utils.common.utils import to_list
from ansible_collections.ansible.utils.plugins.plugin_utils.base.validate import ValidateBase from ansible_collections.ansible.utils.plugins.plugin_utils.base.validate import ValidateBase
@ -87,7 +88,7 @@ class Validate(ValidateBase):
""" """
try: try:
if isinstance(self._criteria, str): if isinstance(self._criteria, string_types):
self._criteria = yaml.load(StringIO(self._criteria), Loader=SafeLoader) self._criteria = yaml.load(StringIO(self._criteria), Loader=SafeLoader)
except yaml.parser.ParserError as exc: except yaml.parser.ParserError as exc:
msg = ( msg = (

View file

@ -54,8 +54,9 @@ DOCUMENTATION = """
import json import json
from ansible.errors import AnsibleError from ansible.errors import AnsibleError
from ansible.module_utils._text import to_text
from ansible.module_utils.basic import missing_required_lib from ansible.module_utils.basic import missing_required_lib
from ansible.module_utils.common.text.converters import to_text from ansible.module_utils.six import string_types
from ansible.utils.display import Display from ansible.utils.display import Display
from ansible_collections.ansible.utils.plugins.module_utils.common.utils import to_list from ansible_collections.ansible.utils.plugins.module_utils.common.utils import to_list
@ -137,7 +138,7 @@ class Validate(ValidateBase):
:return: None: In case all arguments passed are valid :return: None: In case all arguments passed are valid
""" """
try: try:
if isinstance(self._data, str): if isinstance(self._data, string_types):
self._data = json.loads(self._data) self._data = json.loads(self._data)
else: else:
self._data = json.loads(json.dumps(self._data)) self._data = json.loads(json.dumps(self._data))
@ -154,7 +155,7 @@ class Validate(ValidateBase):
try: try:
criteria = [] criteria = []
for item in to_list(self._criteria): for item in to_list(self._criteria):
if isinstance(self._criteria, str): if isinstance(self._criteria, string_types):
criteria.append(json.loads(item)) criteria.append(json.loads(item))
else: else:
criteria.append(json.loads(json.dumps(item))) criteria.append(json.loads(json.dumps(item)))

View file

@ -74,7 +74,7 @@ RETURN = """
""" """
from ansible.errors import AnsibleError from ansible.errors import AnsibleError
from ansible.module_utils.common.text.converters import to_text from ansible.module_utils._text import to_text
from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import ( from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import (
check_argspec, check_argspec,

View file

@ -1,39 +0,0 @@
# Complete documentation with many more options at:
# https://docs.sonarqube.org/latest/analysis/analysis-parameters/
## The unique project identifier. This is mandatory.
# Do not duplicate or reuse!
# Available characters: [a-zA-Z0-9_:\.\-]
# Must have least one non-digit.
# Recommended format: <group>:<project>
sonar.projectKey=ansible-collections_ansible.utils
sonar.organization=ansible-collections
# Customize what paths to scan. Default is .
sonar.sources=plugins
sonar.inclusions=plugins/**/*.py
# Verbose name of project displayed in WUI. Default is set to the projectKey. This field is optional.
sonar.projectName=ansible.utils
# Version of project. This field is optional.
#sonar.projectVersion=1.0
# Exclude manifests directory as these are pulled in programmatically, or are partial files to be applied via kustomize
# Exclude cli from scanning until we can add tests
sonar.exclusions=tests/**
# Define test directories
sonar.tests=tests/unit
sonar.test.inclusions=tests/unit/**
# Tell sonar scanner where coverage files exist
# Note: coverage.xml is created by pytest-cov in Code Coverage workflow
sonar.python.coverage.reportPaths=coverage.xml
# Python versions used in CI (matching coverage workflow)
sonar.python.version=3.12
# Ignore code dupe for the migrations
# sonar.cpd.exclusions=

View file

@ -1,11 +1,9 @@
# Unit test runner # Unit test runner
pytest-ansible pytest-ansible
pytest-xdist pytest-xdist
typing-extensions
# For integration tests # For integration tests
pexpect pexpect
git+https://github.com/ansible-network/pytest-ansible-network-integration.git git+https://github.com/ansible-network/pytest-ansible-network-integration.git
ipaddress ; python_version < '3.0' ipaddress ; python_version < '3.0'
netaddr==0.10.1 netaddr==0.10.1
textfsm
ttp

View file

@ -1,3 +1,3 @@
--- ---
modules: modules:
python_requires: ">=3.10" python_requires: ">=3.9"

View file

@ -1,13 +0,0 @@
#
# Capture any deprecation warnings of e.g. ansible.module_utils for pytest
#
import warnings
import ansible.utils.display
ansible.utils.display.Display().deprecated = lambda msg, *args, **kwargs: warnings.warn(
msg,
DeprecationWarning,
stacklevel=3,
)

View file

@ -16,4 +16,4 @@
with_items: with_items:
- "{{ nxos_json_text['parsed'] is defined }}" - "{{ nxos_json_text['parsed'] is defined }}"
- "{{ nxos_json_text['parsed'][0][0][0]['admin_state'] is defined }}" - "{{ nxos_json_text['parsed'][0][0][0]['admin_state'] is defined }}"
# - "{{ nxos_json_text['parsed'] == nxos_json_text_parsed }}" - "{{ nxos_json_text['parsed'] == nxos_json_text_parsed }}"

View file

@ -1,11 +1,11 @@
--- ---
- name: Set fact - name: Set fact
ansible.builtin.set_fact: ansible.builtin.set_fact:
nxos_ttp_text_parsed: "{{ lookup('ansible.builtin.file', role_path ~ '/output/nxos_show_interface_ttp_parsed.json') | from_json }}" nxos_ttp_text_parsed: "{{ lookup('ansible.builtin.file', '{{ role_path }}/output/nxos_show_interface_ttp_parsed.json') | from_json }}"
- name: "Pass text and template_path {{ parser }}" - name: "Pass text and template_path {{ parser }}"
ansible.utils.cli_parse: ansible.utils.cli_parse:
text: "{{ lookup('ansible.builtin.file', role_path ~ '/files/nxos_show_interface.txt') }}" text: "{{ lookup('ansible.builtin.file', '{{ role_path }}/files/nxos_show_interface.txt') }}"
parser: parser:
name: ansible.utils.ttp name: ansible.utils.ttp
template_path: "{{ role_path }}/templates/nxos_show_interface.ttp" template_path: "{{ role_path }}/templates/nxos_show_interface.ttp"
@ -16,7 +16,7 @@
ansible.builtin.assert: ansible.builtin.assert:
that: that:
- "{{ POpqMQoJWTiDpEW is defined }}" - "{{ POpqMQoJWTiDpEW is defined }}"
- "{{ (nxos_ttp_text['parsed'][0][0] | selectattr('interface', 'search', 'mgmt0') | list | length) > 0 }}" - nxos_ttp_text['parsed'][0][0] | selectattr('interface', 'search', 'mgmt0') | list | length
- "{{ nxos_ttp_text['parsed'] == nxos_ttp_text_parsed }}" - "{{ nxos_ttp_text['parsed'] == nxos_ttp_text_parsed }}"
- name: "Pass text and custom variable {{ parser }}" - name: "Pass text and custom variable {{ parser }}"
@ -51,4 +51,4 @@
ansible.builtin.assert: ansible.builtin.assert:
that: item that: item
with_items: with_items:
- "{{ ((nxos_ttp_results['parsed'][0] | from_yaml)[0] | selectattr('interface', 'search', 'mgmt0') | list | length) > 0 }}" - "{{ (nxos_ttp_results['parsed'][0] | from_yaml)[0] | selectattr('interface', 'search', 'mgmt0') | list | length }}"

View file

@ -19,12 +19,10 @@
- name: Integration tests with and without default engine as xmltodict and - name: Integration tests with and without default engine as xmltodict and
ansible.builtin.assert: ansible.builtin.assert:
that: "{{ output == test_item.test }}" that: "{{ output == item.test }}"
loop: loop:
- test: "{{ data | ansible.utils.from_xml() }}" - test: "{{ data | ansible.utils.from_xml() }}"
- test: "{{ data | ansible.utils.from_xml('xmltodict') }}" - test: "{{ data | ansible.utils.from_xml('xmltodict') }}"
loop_control:
loop_var: test_item
- name: Setup invalid xml as input to ansible.utils.from_xml. - name: Setup invalid xml as input to ansible.utils.from_xml.
ansible.builtin.set_fact: ansible.builtin.set_fact:

View file

@ -52,15 +52,15 @@
# ok: [sw01] => (item=2) => # ok: [sw01] => (item=2) =>
# msg: 3 is > than 1 # msg: 3 is > than 1
# - name: Find numbers greater than 1, using with - name: Find numbers greater than 1, using with
# ansible.builtin.debug: ansible.builtin.debug:
# msg: "{{ data[item] }} is {{ params.test }} than {{ params.value }}" msg: "{{ data[item] }} is {{ params.test }} than {{ params.value }}"
# with_ansible.utils.index_of: "{{ params }}" with_ansible.utils.index_of: "{{ params }}"
# vars: vars:
# params: params:
# data: "{{ data }}" data: "{{ data }}"
# test: ">" test: ">"
# value: 1 value: 1
# TASK [Find numbers greater than 1, using with] ***************************** # TASK [Find numbers greater than 1, using with] *****************************
# ok: [nxos101] => (item=1) => # ok: [nxos101] => (item=1) =>

View file

@ -43,10 +43,10 @@
# - test: "{{ lookup('ansible.utils.index_of', complex.a, 'integer') }}" # - test: "{{ lookup('ansible.utils.index_of', complex.a, 'integer') }}"
# result: "3" # result: "3"
# - test: "{{ complex.b | ansible.utils.index_of('==', 1, 'b1') }}" - test: "{{ complex.b | ansible.utils.index_of('==', 1, 'b1') }}"
# result: 0 result: "0"
# - test: "{{ lookup('ansible.utils.index_of', complex.b, '==', 1, 'b1') }}" - test: "{{ lookup('ansible.utils.index_of', complex.b, '==', 1, 'b1') }}"
# result: 0 result: "0"
- test: "{{ complex.c.c1 | ansible.utils.index_of('!=', 'c') }}" - test: "{{ complex.c.c1 | ansible.utils.index_of('!=', 'c') }}"
result: [0, 1] result: [0, 1]
@ -75,15 +75,35 @@
ansible.builtin.assert: ansible.builtin.assert:
that: "{{ item.test == item.result }}" that: "{{ item.test == item.result }}"
loop: loop:
- test: "{{ complex.a.b.c.d | ansible.utils.index_of('eq', 'ansible', 'e1') }}"
result: "0"
- test: "{{ lookup('ansible.utils.index_of', complex.a.b.c.d, 'eq', 'ansible', 'e1') }}"
result: "0"
- test: "{{ complex.a.b.c.d | ansible.utils.index_of('eq', 'ansible', 'e1', wantlist=true) }}" - test: "{{ complex.a.b.c.d | ansible.utils.index_of('eq', 'ansible', 'e1', wantlist=true) }}"
result: [0] result: [0]
- test: "{{ lookup('ansible.utils.index_of', complex.a.b.c.d, 'eq', 'ansible', 'e1', wantlist=true) }}" - test: "{{ lookup('ansible.utils.index_of', complex.a.b.c.d, 'eq', 'ansible', 'e1', wantlist=true) }}"
result: [0] result: [0]
# - name: Test a missing key in the list of dictionaries
# ansible.builtin.assert: - name: Test a missing key in the list of dictionaries
# that: "{{ item.test == item.result }}" ansible.builtin.assert:
# loop: that: "{{ item.test == item.result }}"
# - test: "{{ complex.a.b.c.d | ansible.utils.index_of('eq', true, 'e2') }}" loop:
# result: 0 - test: "{{ complex.a.b.c.d | ansible.utils.index_of('eq', true, 'e2') }}"
# - test: "{{ lookup('ansible.utils.index_of', complex.a.b.c.d, 'eq', true, 'e2') }}" result: "0"
# result: 0 - test: "{{ lookup('ansible.utils.index_of', complex.a.b.c.d, 'eq', true, 'e2') }}"
result: "0"
- name: Test a missing key in the list of dictionaries, fail on missing
ansible.builtin.assert:
that: "{{ item.test == item.result }}"
loop:
- test: "{{ complex.a.b.c.d | ansible.utils.index_of('eq', true, 'e2', fail_on_missing=true) }}"
result: "0"
- test: "{{ lookup('ansible.utils.index_of', complex.a.b.c.d, 'eq', true, 'e2', fail_on_missing=true) }}"
result: "0"
ignore_errors: true
register: result
- name: Ensure the previous test failed
ansible.builtin.assert:
that: "{{ result.failed and 'not found in' in result.msg }}"

View file

@ -6,3 +6,11 @@
- name: Assert result for next_nth_usable. - name: Assert result for next_nth_usable.
ansible.builtin.assert: ansible.builtin.assert:
that: "{{ result1 == '192.168.122.3' }}" that: "{{ result1 == '192.168.122.3' }}"
- name: Next_nth_usable filter
ansible.builtin.set_fact:
result1: "{{ '192.168.122.254/24' | ansible.utils.next_nth_usable(2) }}"
- name: Assert result for ipv4.
ansible.builtin.assert:
that: "{{ result1 == '' }}"

View file

@ -6,3 +6,11 @@
- name: Assert result for previous_nth_usable. - name: Assert result for previous_nth_usable.
ansible.builtin.assert: ansible.builtin.assert:
that: "{{ result1 == '192.168.122.8' }}" that: "{{ result1 == '192.168.122.8' }}"
- name: Previous_nth_usable filter
ansible.builtin.set_fact:
result1: "{{ '192.168.122.1/24' | ansible.utils.previous_nth_usable(2) }}"
- name: Assert result for ipv4.
ansible.builtin.assert:
that: "{{ result1 == '' }}"

View file

@ -1,88 +0,0 @@
---
- name: Include expected output data
ansible.builtin.include_vars:
file: complex.yaml
- name: Test keep_keys
tags: keep_keys
block:
- name: Setup data as facts for complex keep_keys integration test l1
ansible.builtin.set_fact:
l1:
- { p1: a, p2: a, p3: a }
- { p1: b, p2: b, p3: b }
- { p1: c, p2: c, p3: c }
- name: Test keep_keys debug
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.keep_keys(target=['p1', 'p2']) }}"
register: result
- name: Test keep_keys assert
ansible.builtin.assert:
that: result['msg'] == keep['l1']
- name: Debug l1 for starts_with 'p'
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.keep_keys(target=['p1', 'p2'], matching_parameter='starts_with') }}"
register: result_l1_starts_with_p
- name: Assert result dicts l1 for starts_with 'p'
ansible.builtin.assert:
that:
- result_l1_starts_with_p['msg'] == keep['l1_starts_with_p']
- name: Debug l1 for ends_with
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.keep_keys(target=['p1', 'p2'], matching_parameter='ends_with') }}"
register: result_l1_ends_with
- name: Assert result dicts l1 for ends_with
ansible.builtin.assert:
that:
- result_l1_ends_with['msg'] == keep['l1_ends_with']
- name: Debug l1 for ends_with '2'
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.keep_keys(target=['2'], matching_parameter='ends_with') }}"
register: result_l1_ends_with_2
- name: Assert result dicts l1 for ends_with '2'
ansible.builtin.assert:
that:
- result_l1_ends_with_2['msg'] == keep['l1_ends_with_2']
- name: Test keep_keys regex
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.keep_keys(target=['p1', 'p2'], matching_parameter='regex') }}"
register: result_l1_regex
- name: Assert result dicts l1 for regex
ansible.builtin.assert:
that:
- result_l1_regex['msg'] == keep['l1_regex']
- name: Test map keep_keys
tags: map_keep_keys
block:
- name: Setup data as facts for complex keep integration test l2
ansible.builtin.set_fact:
l2:
- - { p1: a, p2: a, p3: a }
- { p1: b, p2: b, p3: b }
- { p1: c, p2: c, p3: c }
- - { p1: a, p2: a, p3: a }
- { p1: b, p2: b, p3: b }
- { p1: c, p2: c, p3: c }
- - { p1: a, p2: a, p3: a }
- { p1: b, p2: b, p3: b }
- { p1: c, p2: c, p3: c }
- name: Test map keep_keys debug
ansible.builtin.debug:
msg: "{{ l2 | map('ansible.utils.keep_keys', target=['p1', 'p2']) | list }}"
register: result
- name: Test map keep_keys assert
ansible.builtin.assert:
that: result['msg'] == keep['l2']

View file

@ -33,7 +33,7 @@
- name: Assert result dicts - name: Assert result dicts
ansible.builtin.assert: ansible.builtin.assert:
that: that:
- keep_values['starts_with'] == result['msg'] - keep['starts_with'] == result['msg']
- name: Setup data as facts for equivalent - name: Setup data as facts for equivalent
ansible.builtin.set_fact: ansible.builtin.set_fact:
@ -70,54 +70,3 @@
ansible.builtin.assert: ansible.builtin.assert:
that: that:
- keep_default['default'] == result['msg'] - keep_default['default'] == result['msg']
- name: Setup data for multiple keys in dict
ansible.builtin.set_fact:
tomcat_data:
tomcat:
tomcat1:
name: tomcat1
tomcat2:
name: tomcat2
tomcat3:
name: tomcat3
tomcat3:
name: tomcat3
tomcat4:
name: tomcat3
tomcat3:
tomcat3:
name: tomcattest
checktomcat3: stringinput
tomcats_block:
- tomcat1
- tomcat2
- name: Debug
ansible.builtin.debug:
msg: "{{ tomcat_data['tomcat'] | ansible.utils.keep_keys(target=['tomcats_block']) }}"
register: result
- name: Assert result dicts
ansible.builtin.assert:
that:
- keep_tomcat['tomcat'] == result['msg']
- name: Test to keep keys when initial node is matched
ansible.builtin.debug:
msg: "{{ tomcat_data['tomcat'] | ansible.utils.keep_keys(target=['tomcat1', 'tomcat2']) }}"
register: result
- name: Assert result dicts
ansible.builtin.assert:
that:
- keep_tomcat['greedy_values'] == result['msg']
- name: Test to keep keys to check if nested dict is matched
ansible.builtin.debug:
msg: "{{ tomcat_data['tomcat'] | ansible.utils.keep_keys(target=['tomcat3']) }}"
register: result
- name: Assert result dicts
ansible.builtin.assert:
that:
- keep_tomcat['nested_values'] == result['msg']

View file

@ -1,36 +0,0 @@
---
keep:
l1:
- { p1: a, p2: a }
- { p1: b, p2: b }
- { p1: c, p2: c }
l2:
- - { p1: a, p2: a }
- { p1: b, p2: b }
- { p1: c, p2: c }
- - { p1: a, p2: a }
- { p1: b, p2: b }
- { p1: c, p2: c }
- - { p1: a, p2: a }
- { p1: b, p2: b }
- { p1: c, p2: c }
l3:
- { p1_key_o1: a, p2_key_o2: a, p3_key_o3: a }
- { p1_key_o4: b, p2_key_o5: b, p3_key_o6: b }
- { p1_key_o7: c, p2_key_o8: c, p3_key_o9: c }
l1_starts_with_p:
- { p1: a, p2: a }
- { p1: b, p2: b }
- { p1: c, p2: c }
l1_ends_with:
- { p1: a, p2: a }
- { p1: b, p2: b }
- { p1: c, p2: c }
l1_ends_with_2:
- { p2: a }
- { p2: b }
- { p2: c }
l1_regex:
- { p1: a, p2: a }
- { p1: b, p2: b }
- { p1: c, p2: c }

View file

@ -1,5 +1,5 @@
--- ---
keep_values: keep:
starts_with: starts_with:
- interface_name: eth0 - interface_name: eth0
- interface_name: eth1 - interface_name: eth1
@ -22,27 +22,3 @@ keep_default:
is_enabled: true is_enabled: true
- interface_name: eth2 - interface_name: eth2
is_enabled: false is_enabled: false
keep_tomcat:
tomcat:
tomcats_block:
- tomcat1
- tomcat2
greedy_values:
tomcat1:
name: tomcat1
tomcat2:
name: tomcat2
tomcat3:
name: tomcat3
nested_values:
tomcat2:
tomcat3:
name: tomcat3
tomcat3:
name: tomcat3
tomcat4:
tomcat3:
tomcat3:
name: tomcattest
checktomcat3: stringinput

View file

@ -1,143 +0,0 @@
---
- name: Include expected output data
ansible.builtin.include_vars:
file: complex.yaml
- name: Test for remove_keys
vars:
target:
- { p3: a }
- { p3: b }
- { p3: c }
tags: remove_keys
block:
- name: Test remove_keys debug
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.remove_keys(target=['p1', 'p2']) }}"
register: result_l1
- name: Test remove_keys assert
ansible.builtin.assert:
that:
- result_l1["msg"] == target
- name: Test map remove_keys
vars:
target:
- - { p3: a }
- { p3: b }
- { p3: c }
- - { p3: a }
- { p3: b }
- { p3: c }
- - { p3: a }
- { p3: b }
- { p3: c }
tags: remove_keys
block:
- name: Test remove_keys map debug
ansible.builtin.debug:
msg: "{{ l2 | map('ansible.utils.remove_keys', target=['p1', 'p2']) | list }}"
register: result_l2
- name: Convert to yaml
ansible.builtin.debug:
var: result_l2|to_yaml
when: debug_test|d(false)|bool
- name: Test remove_keys map assert
ansible.builtin.assert:
that:
- result_l2["msg"] == target
- name: Test remove_keys starts_with
vars:
target:
- { p3: a }
- { p3: b }
- { p3: c }
tags: remove_keys_starts_with_2
block:
- name: Test remove_keys starts_with debug msg
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.remove_keys(target=['p1', 'p2'], matching_parameter='starts_with') }}"
register: result_3
- name: Test remove_keys starts_with convert to yaml
ansible.builtin.debug:
var: result_3|to_yaml
when: debug_test|d(false)|bool
- name: Test remove_keys assert
ansible.builtin.assert:
that: result_3["msg"] == target
- name: Test remove_keys starts_with 'p'
tags: remove_keys_starts_with_1
block:
- name: Test remove_keys starts_with 'p' debug
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.remove_keys(target=['p'], matching_parameter='starts_with') }}"
register: result_4
- name: Test remove_keys starts_with 'p' convert to yaml
ansible.builtin.debug:
var: result_4|to_yaml
when: debug_test|d(false)|bool
- name: Test remove_keys assert
ansible.builtin.assert:
that: result_4["msg"] == []
- name: Test remove_keys ends_with
vars:
target:
- { p3: a }
- { p3: b }
- { p3: c }
tags: remove_keys_ends_with_2
block:
- name: Test remove_keys ends_with
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.remove_keys(target=['p1', 'p2'], matching_parameter='ends_with') }}"
register: result_5
- name: Test remove_keys ends_with conver to yaml
ansible.builtin.debug:
var: result_5|to_yaml
when: debug_test|d(false)|bool
- name: Test remove_keys ends_with assert
ansible.builtin.assert:
that: result_5["msg"] == target
- name: Test remove_keys ends_with '2'
vars:
target:
- { p1: a, p3: a }
- { p1: b, p3: b }
- { p1: c, p3: c }
tags: remove_keys_ends_with_1
block:
- name: Test remove_keys ends_with '2' debug
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.remove_keys(target=['2'], matching_parameter='ends_with') }}"
register: result_6
- name: Test remove_keys ends_with '2' debug convert to yaml
ansible.builtin.debug:
var: result_6|to_yaml
when: debug_test|d(false)|bool
- name: Test remove_keys ends_with '2' assert
ansible.builtin.assert:
that: result_6["msg"] == target
- name: Test remove_keys regex
vars:
target:
- { p3: a }
- { p3: b }
- { p3: c }
tags: remove_keys_regex
block:
- name: Test remove_keys regex debug
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.remove_keys(target=['p1', 'p2'], matching_parameter='regex') }}"
register: result_7
- name: Test remove_keys regex debug conver to yaml
ansible.builtin.debug:
var: result_7|to_yaml
when: debug_test|d(false)|bool
- name: Test remove_keys regex assert
ansible.builtin.assert:
that: result_7["msg"] == target

View file

@ -1,19 +0,0 @@
---
l1:
- { p1: a, p2: a, p3: a }
- { p1: b, p2: b, p3: b }
- { p1: c, p2: c, p3: c }
l2:
- - { p1: a, p2: a, p3: a }
- { p1: b, p2: b, p3: b }
- { p1: c, p2: c, p3: c }
- - { p1: a, p2: a, p3: a }
- { p1: b, p2: b, p3: b }
- { p1: c, p2: c, p3: c }
- - { p1: a, p2: a, p3: a }
- { p1: b, p2: b, p3: b }
- { p1: c, p2: c, p3: c }
l3:
- { p1_key_o1: a, p2_key_o2: a, p3_key_o3: a }
- { p1_key_o4: b, p2_key_o5: b, p3_key_o6: b }
- { p1_key_o7: c, p2_key_o8: c, p3_key_o9: c }

View file

@ -1,127 +0,0 @@
---
- name: Include expected output data
ansible.builtin.include_vars:
file: complex.yaml
- name: Block for l1 tests
block:
- name: Setup data as facts for complex replace integration test l1
ansible.builtin.set_fact:
l1:
- { p1: a, p2: a, p3: a }
- { p1: b, p2: b, p3: b }
- { p1: c, p2: c, p3: c }
- name: Debug l1
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.replace_keys(target=[{'before': 'p1', 'after': 'x1'}, {'before': 'p2', 'after': 'x2'}]) }}"
register: result_l1
- name: Assert result dicts l1
ansible.builtin.assert:
that:
- result_l1['msg'] == replace['l1']
- name: Debug l1 for starts_with 'p'
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.replace_keys(target=[{'before': 'p', 'after': 'x'}], matching_parameter='starts_with') }}"
register: result_l1_starts_with_p
- name: Assert result dicts l1 for starts_with 'p'
ansible.builtin.assert:
that:
- result_l1_starts_with_p['msg'] == replace['l1_starts_with_p']
- name: Debug l1 for ends_with
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.replace_keys(target=[{'before': 'p1', 'after': 'x1'}, {'before': 'p2', 'after': 'x2'}], matching_parameter='ends_with') }}"
register: result_l1_ends_with
- name: Assert result dicts l1 for ends_with
ansible.builtin.assert:
that:
- result_l1_ends_with['msg'] == replace['l1_ends_with']
- name: Debug l1 for ends_with '2'
ansible.builtin.debug:
msg: "{{ l1 | ansible.utils.replace_keys(target=[{'before': '2', 'after': '9'}], matching_parameter='ends_with') }}"
register: result_l1_ends_with_2
- name: Assert result dicts l1 for ends_with '2'
ansible.builtin.assert:
that:
- result_l1_ends_with_2['msg'] == replace['l1_ends_with_2']
- name: Block for l2 tests
block:
- name: Setup data as facts for complex replace integration test l2
ansible.builtin.set_fact:
l2:
- - { p1: a, p2: a, p3: a }
- { p1: b, p2: b, p3: b }
- { p1: c, p2: c, p3: c }
- - { p1: a, p2: a, p3: a }
- { p1: b, p2: b, p3: b }
- { p1: c, p2: c, p3: c }
- - { p1: a, p2: a, p3: a }
- { p1: b, p2: b, p3: b }
- { p1: c, p2: c, p3: c }
- name: Debug l2
ansible.builtin.debug:
msg: "{{ l2 | map('ansible.utils.replace_keys', target=[{'before': 'p1', 'after': 'x1'}, {'before': 'p2', 'after': 'x2'}]) | list }}"
register: result_l2
- name: Assert result dicts l2
ansible.builtin.assert:
that:
- result_l2['msg'] == replace['l2']
- name: Block for l3 tests
block:
- name: Setup data as facts for replace integration test l3
ansible.builtin.set_fact:
l3:
- { p1_key_o1: a, p2_key_o2: a, p3_key_o3: a }
- { p1_key_o4: b, p2_key_o5: b, p3_key_o6: b }
- { p1_key_o7: c, p2_key_o8: c, p3_key_o9: c }
- name: Debug l3
ansible.builtin.debug:
msg: "{{ l3 | ansible.utils.replace_keys(target=[{'before': 'p1_key_o1', 'after': 'p1_key_oX1'}, {'before': 'p2_key_o2', 'after': 'p2_key_oX2'}]) }}"
register: result_l3
- name: Assert result dicts l3
ansible.builtin.assert:
that:
- result_l3['msg'] == replace['l3']
- name: Debug l3 for starts_with
ansible.builtin.debug:
msg: "{{ l3 | ansible.utils.replace_keys(target=[{'before': 'p1', 'after': 'x1'}, {'before': 'p2', 'after': 'x2'}], matching_parameter='starts_with') }}"
register: result_l3_starts_with
- name: Assert result dicts l3 for starts_with
ansible.builtin.assert:
that:
- result_l3_starts_with['msg'] == replace['l3_starts_with']
- name: Debug l3 for ends_with
ansible.builtin.debug:
msg: "{{ l3 | ansible.utils.replace_keys(target=[{'before': 'o1', 'after': 'x1'}, {'before': 'o2', 'after': 'x2'}], matching_parameter='ends_with') }}"
register: result_l3_ends_with
- name: Assert result dicts l3 for ends_with
ansible.builtin.assert:
that:
- result_l3_ends_with['msg'] == replace['l3_ends_with']
- name: Debug l3 for regex
ansible.builtin.debug:
msg: "{{ l3 | ansible.utils.replace_keys(target=[{'before': '^(.*)_key_o2$', 'after': 'XY_key_o2'}], matching_parameter='regex') }}"
register: result_l3_regex
- name: Assert result dicts l3 for regex
ansible.builtin.assert:
that:
- result_l3_regex['msg'] == replace['l3_regex']

View file

@ -25,10 +25,6 @@
mtu: 600 mtu: 600
enabled: false enabled: false
- name: Include expected output data
ansible.builtin.include_vars:
file: main.yaml
- name: Debug - name: Debug
ansible.builtin.debug: ansible.builtin.debug:
msg: "{{ data | ansible.utils.replace_keys(target=[{'before': 'interface_name', 'after': 'name'}]) }}" msg: "{{ data | ansible.utils.replace_keys(target=[{'before': 'interface_name', 'after': 'name'}]) }}"

View file

@ -1,44 +0,0 @@
---
replace:
l1:
- { "p3": "a", "x1": "a", "x2": "a" }
- { "p3": "b", "x1": "b", "x2": "b" }
- { "p3": "c", "x1": "c", "x2": "c" }
l2:
- - { "p3": "a", "x1": "a", "x2": "a" }
- { "p3": "b", "x1": "b", "x2": "b" }
- { "p3": "c", "x1": "c", "x2": "c" }
- - { "p3": "a", "x1": "a", "x2": "a" }
- { "p3": "b", "x1": "b", "x2": "b" }
- { "p3": "c", "x1": "c", "x2": "c" }
- - { "p3": "a", "x1": "a", "x2": "a" }
- { "p3": "b", "x1": "b", "x2": "b" }
- { "p3": "c", "x1": "c", "x2": "c" }
l3:
- { "p1_key_oX1": "a", "p2_key_oX2": "a", "p3_key_o3": "a" }
- { "p1_key_o4": "b", "p2_key_o5": "b", "p3_key_o6": "b" }
- { "p1_key_o7": "c", "p2_key_o8": "c", "p3_key_o9": "c" }
l3_starts_with:
- { "p3_key_o3": "a", "x1": "a", "x2": "a" }
- { "p3_key_o6": "b", "x1": "b", "x2": "b" }
- { "p3_key_o9": "c", "x1": "c", "x2": "c" }
l1_starts_with_p:
- { "x": "a" }
- { "x": "b" }
- { "x": "c" }
l1_ends_with:
- { "p3": "a", "x1": "a", "x2": "a" }
- { "p3": "b", "x1": "b", "x2": "b" }
- { "p3": "c", "x1": "c", "x2": "c" }
l3_ends_with:
- { "p3_key_o3": "a", "x1": "a", "x2": "a" }
- { "p1_key_o4": "b", "p2_key_o5": "b", "p3_key_o6": "b" }
- { "p1_key_o7": "c", "p2_key_o8": "c", "p3_key_o9": "c" }
l1_ends_with_2:
- { "9": "a", "p1": "a", "p3": "a" }
- { "9": "b", "p1": "b", "p3": "b" }
- { "9": "c", "p1": "c", "p3": "c" }
l3_regex:
- { "XY_key_o2": "a", "p1_key_o1": "a", "p3_key_o3": "a" }
- { "p1_key_o4": "b", "p2_key_o5": "b", "p3_key_o6": "b" }
- { "p1_key_o7": "c", "p2_key_o8": "c", "p3_key_o9": "c" }

View file

@ -63,11 +63,11 @@
- name: read data and criteria from file - name: read data and criteria from file
ansible.builtin.set_fact: ansible.builtin.set_fact:
data: "{{ lookup('ansible.builtin.file', 'data/show_interface.json') | from_json }}" data: "{{ lookup('ansible.builtin.file', 'data/show_interface.json') }}"
oper_status_up_criteria: "{{ lookup('ansible.builtin.file', 'criteria/oper_status_up.json') | from_json }}" oper_status_up_criteria: "{{ lookup('ansible.builtin.file', 'criteria/oper_status_up.json') }}"
enabled_check_criteria: "{{ lookup('ansible.builtin.file', 'criteria/enabled_check.json') | from_json }}" enabled_check_criteria: "{{ lookup('ansible.builtin.file', 'criteria/enabled_check.json') }}"
crc_error_check_criteria: "{{ lookup('ansible.builtin.file', 'criteria/crc_error_check.json') | from_json }}" crc_error_check_criteria: "{{ lookup('ansible.builtin.file', 'criteria/crc_error_check.json') }}"
in_rate_check_criteria: "{{ lookup('ansible.builtin.file', 'criteria/in_rate_check.json') | from_json }}" in_rate_check_criteria: "{{ lookup('ansible.builtin.file', 'criteria/in_rate_check.json') }}"
- name: validate data using jsonschema engine (invalid data read from file) - name: validate data using jsonschema engine (invalid data read from file)
ansible.builtin.set_fact: ansible.builtin.set_fact:

View file

@ -91,6 +91,12 @@
ansible.builtin.set_fact: ansible.builtin.set_fact:
data_criteria_checks: "{{ lookup('ansible.utils.validate', data, [oper_status_up_criteria, enabled_check_criteria, crc_error_check_criteria], engine='ansible.utils.jsonschema') }}" data_criteria_checks: "{{ lookup('ansible.utils.validate', data, [oper_status_up_criteria, enabled_check_criteria, crc_error_check_criteria], engine='ansible.utils.jsonschema') }}"
- ansible.builtin.assert:
that:
- "data_criteria_checks[0].data_path == 'GigabitEthernet0/0/0/0.oper_status'"
- "data_criteria_checks[1].data_path == 'GigabitEthernet0/0/0/1.enabled'"
- "data_criteria_checks[2].data_path == 'GigabitEthernet0/0/0/1.counters.in_crc_errors'"
- name: validate data using jsonschema engine (valid data read from file) - name: validate data using jsonschema engine (valid data read from file)
ansible.builtin.set_fact: ansible.builtin.set_fact:
data_criteria_checks: "{{ lookup('ansible.utils.validate', data, in_rate_check_criteria, engine='ansible.utils.jsonschema') }}" data_criteria_checks: "{{ lookup('ansible.utils.validate', data, in_rate_check_criteria, engine='ansible.utils.jsonschema') }}"

View file

@ -110,14 +110,13 @@
- ansible.builtin.assert: - ansible.builtin.assert:
that: that:
- "'errors' in result" - "'errors' in result"
- "'GigabitEthernet0/0/0/0.oper_status' in result['errors'][0]" - "result['errors'][0].data_path == 'GigabitEthernet0/0/0/0.oper_status'"
- "'GigabitEthernet0/0/0/1.enabled' in result['errors'][1]" - "result['errors'][1].data_path == 'GigabitEthernet0/0/0/1.enabled'"
- "'GigabitEthernet0/0/0/1.counters.in_crc_errors' in result['errors'][2]" - "result['errors'][2].data_path == 'GigabitEthernet0/0/0/1.counters.in_crc_errors'"
- "'Validation errors were found' in result.msg" - "'Validation errors were found' in result.msg"
- "'patternProperties.^.*.properties.oper_status.pattern' in result.msg" - "'patternProperties.^.*.properties.oper_status.pattern' in result.msg"
- "'patternProperties.^.*.properties.enabled.enum' in result.msg" - "'patternProperties.^.*.properties.enabled.enum' in result.msg"
- "'patternProperties.^.*.properties.counters.properties.in_crc_errors.maximum' in result.msg" - "'patternProperties.^.*.properties.counters.properties.in_crc_errors.maximum' in result.msg"
ignore_errors: true
- name: validate data using jsonschema engine (valid data read from file) - name: validate data using jsonschema engine (valid data read from file)
ansible.utils.validate: ansible.utils.validate:

View file

@ -89,14 +89,13 @@
- name: read data and criteria from file - name: read data and criteria from file
ansible.builtin.set_fact: ansible.builtin.set_fact:
data: "{{ lookup('ansible.builtin.file', 'data/test_format_checker.json') | from_json }}" data: "{{ lookup('ansible.builtin.file', 'data/test_format_checker.json') }}"
criteria_1: "{{ lookup('ansible.builtin.file', 'criteria/format_checker.json') | from_json }}" criteria_1: "{{ lookup('ansible.builtin.file', 'criteria/format_checker.json') }}"
- name: validate data using jsonschema engine - name: validate data using jsonschema engine (valid data read from file)
ansible.builtin.set_fact: ansible.builtin.set_fact:
is_data_valid: "{{ data is ansible.utils.validate(engine='ansible.utils.jsonschema', criteria=[criteria_1], draft='draft7') }}" is_data_valid: "{{ data is ansible.utils.validate(engine='ansible.utils.jsonschema', criteria=[criteria_1], draft='draft7') }}"
- name: assert the data is invalid - ansible.builtin.assert:
ansible.builtin.assert:
that: that:
- "is_data_valid == true" - "is_data_valid == true"

View file

@ -24,7 +24,7 @@ __metaclass__ = type
import os import os
from ansible.errors import AnsibleParserError from ansible.errors import AnsibleParserError
from ansible.module_utils.common.text.converters import to_bytes, to_text from ansible.module_utils._text import to_bytes, to_text
from ansible.parsing.dataloader import DataLoader from ansible.parsing.dataloader import DataLoader

View file

@ -29,7 +29,8 @@ from contextlib import contextmanager
from io import BytesIO, StringIO from io import BytesIO, StringIO
from unittest import TestCase from unittest import TestCase
from ansible.module_utils.common.text.converters import to_bytes from ansible.module_utils._text import to_bytes
from ansible.module_utils.six import PY3
@contextmanager @contextmanager
@ -40,8 +41,11 @@ def swap_stdin_and_argv(stdin_data="", argv_data=tuple()):
real_stdin = sys.stdin real_stdin = sys.stdin
real_argv = sys.argv real_argv = sys.argv
if PY3:
fake_stream = StringIO(stdin_data) fake_stream = StringIO(stdin_data)
fake_stream.buffer = BytesIO(to_bytes(stdin_data)) fake_stream.buffer = BytesIO(to_bytes(stdin_data))
else:
fake_stream = BytesIO(to_bytes(stdin_data))
try: try:
sys.stdin = fake_stream sys.stdin = fake_stream
@ -60,7 +64,10 @@ def swap_stdout():
""" """
old_stdout = sys.stdout old_stdout = sys.stdout
if PY3:
fake_stream = StringIO() fake_stream = StringIO()
else:
fake_stream = BytesIO()
try: try:
sys.stdout = fake_stream sys.stdout = fake_stream

View file

@ -17,7 +17,7 @@ from __future__ import absolute_import, division, print_function
__metaclass__ = type __metaclass__ = type
from ansible.module_utils.common.text.converters import to_bytes from ansible.module_utils._text import to_bytes
from ansible.parsing.vault import VaultSecret from ansible.parsing.vault import VaultSecret

View file

@ -6,6 +6,7 @@ import io
import yaml import yaml
from ansible.module_utils.six import PY3
from ansible.parsing.yaml.dumper import AnsibleDumper from ansible.parsing.yaml.dumper import AnsibleDumper
from ansible.parsing.yaml.loader import AnsibleLoader from ansible.parsing.yaml.loader import AnsibleLoader
@ -21,11 +22,17 @@ class YamlTestUtils(object):
def _dump_stream(self, obj, stream, dumper=None): def _dump_stream(self, obj, stream, dumper=None):
"""Dump to a py2-unicode or py3-string stream.""" """Dump to a py2-unicode or py3-string stream."""
if PY3:
return yaml.dump(obj, stream, Dumper=dumper) return yaml.dump(obj, stream, Dumper=dumper)
else:
return yaml.dump(obj, stream, Dumper=dumper, encoding=None)
def _dump_string(self, obj, dumper=None): def _dump_string(self, obj, dumper=None):
"""Dump to a py2-unicode or py3-string""" """Dump to a py2-unicode or py3-string"""
if PY3:
return yaml.dump(obj, Dumper=dumper) return yaml.dump(obj, Dumper=dumper)
else:
return yaml.dump(obj, Dumper=dumper, encoding=None)
def _dump_load_cycle(self, obj): def _dump_load_cycle(self, obj):
# Each pass though a dump or load revs the 'generation' # Each pass though a dump or load revs the 'generation'
@ -82,8 +89,22 @@ class YamlTestUtils(object):
stream_obj_from_stream = io.StringIO() stream_obj_from_stream = io.StringIO()
stream_obj_from_string = io.StringIO() stream_obj_from_string = io.StringIO()
if PY3:
yaml.dump(obj_from_stream, stream_obj_from_stream, Dumper=AnsibleDumper) yaml.dump(obj_from_stream, stream_obj_from_stream, Dumper=AnsibleDumper)
yaml.dump(obj_from_stream, stream_obj_from_string, Dumper=AnsibleDumper) yaml.dump(obj_from_stream, stream_obj_from_string, Dumper=AnsibleDumper)
else:
yaml.dump(
obj_from_stream,
stream_obj_from_stream,
Dumper=AnsibleDumper,
encoding=None,
)
yaml.dump(
obj_from_stream,
stream_obj_from_string,
Dumper=AnsibleDumper,
encoding=None,
)
yaml_string_stream_obj_from_stream = stream_obj_from_stream.getvalue() yaml_string_stream_obj_from_stream = stream_obj_from_stream.getvalue()
yaml_string_stream_obj_from_string = stream_obj_from_string.getvalue() yaml_string_stream_obj_from_string = stream_obj_from_string.getvalue()
@ -91,8 +112,20 @@ class YamlTestUtils(object):
stream_obj_from_stream.seek(0) stream_obj_from_stream.seek(0)
stream_obj_from_string.seek(0) stream_obj_from_string.seek(0)
if PY3:
yaml_string_obj_from_stream = yaml.dump(obj_from_stream, Dumper=AnsibleDumper) yaml_string_obj_from_stream = yaml.dump(obj_from_stream, Dumper=AnsibleDumper)
yaml_string_obj_from_string = yaml.dump(obj_from_string, Dumper=AnsibleDumper) yaml_string_obj_from_string = yaml.dump(obj_from_string, Dumper=AnsibleDumper)
else:
yaml_string_obj_from_stream = yaml.dump(
obj_from_stream,
Dumper=AnsibleDumper,
encoding=None,
)
yaml_string_obj_from_string = yaml.dump(
obj_from_string,
Dumper=AnsibleDumper,
encoding=None,
)
assert yaml_string == yaml_string_obj_from_stream assert yaml_string == yaml_string_obj_from_stream
assert yaml_string == yaml_string_obj_from_stream == yaml_string_obj_from_string assert yaml_string == yaml_string_obj_from_stream == yaml_string_obj_from_string

View file

@ -3,6 +3,7 @@
# GNU General Public License v3.0+ # GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) # (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function from __future__ import absolute_import, division, print_function
@ -10,14 +11,14 @@ __metaclass__ = type
from unittest import TestCase from unittest import TestCase
from jinja2 import Environment from ansible.template import Templar
from ansible_collections.ansible.utils.plugins.module_utils.common.get_path import get_path from ansible_collections.ansible.utils.plugins.module_utils.common.get_path import get_path
class TestGetPath(TestCase): class TestGetPath(TestCase):
def setUp(self): def setUp(self):
self._environment = Environment() self._environment = Templar(loader=None).environment
def test_get_path_pass(self): def test_get_path_pass(self):
var = {"a": {"b": {"c": {"d": [0, 1]}}}} var = {"a": {"b": {"c": {"d": [0, 1]}}}}
@ -37,7 +38,6 @@ class TestGetPath(TestCase):
var = {"a": {"b": {"c": {"d": [0, 1]}}}} var = {"a": {"b": {"c": {"d": [0, 1]}}}}
path = "a.b.e" path = "a.b.e"
expected = "dict object' has no attribute 'e'" expected = "dict object' has no attribute 'e'"
try: with self.assertRaises(Exception) as exc:
get_path(var, path, environment=self._environment, wantlist=False) get_path(var, path, environment=self._environment, wantlist=False)
except Exception as exc: self.assertIn(expected, str(exc.exception))
self.assertIn(expected, str(exc))

View file

@ -15,7 +15,7 @@ import os
from unittest import TestCase from unittest import TestCase
from jinja2 import Environment from ansible.template import Templar
from ansible_collections.ansible.utils.plugins.module_utils.common.get_path import get_path from ansible_collections.ansible.utils.plugins.module_utils.common.get_path import get_path
from ansible_collections.ansible.utils.plugins.module_utils.common.to_paths import to_paths from ansible_collections.ansible.utils.plugins.module_utils.common.to_paths import to_paths
@ -23,7 +23,7 @@ from ansible_collections.ansible.utils.plugins.module_utils.common.to_paths impo
class TestToPaths(TestCase): class TestToPaths(TestCase):
def setUp(self): def setUp(self):
self._environment = Environment() self._environment = Templar(loader=None).environment
def test_to_paths(self): def test_to_paths(self):
var = {"a": {"b": {"c": {"d": [0, 1]}}}} var = {"a": {"b": {"c": {"d": [0, 1]}}}}

View file

@ -137,23 +137,24 @@ class TestUpdate_Fact(TestCase):
"""Check for a missing fact""" """Check for a missing fact"""
self._plugin._task.args = {"updates": [{"path": "a.b.c", "value": 5}]} self._plugin._task.args = {"updates": [{"path": "a.b.c", "value": 5}]}
with self.assertRaises(Exception) as error: with self.assertRaises(Exception) as error:
self._plugin.run(task_vars={}) self._plugin.run(task_vars={"vars": {}})
self.assertIn("'a' was not found in the current facts.", str(error.exception)) self.assertIn("'a' was not found in the current facts.", str(error.exception))
def test_run_simple(self): def test_run_simple(self):
"""Confirm a valid argspec passes""" """Confirm a valid argspec passes"""
task_vars = {"a": {"b": [1, 2, 3]}} task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars) expected = copy.deepcopy(task_vars["vars"])
expected["a"]["b"] = 5 expected["a"]["b"] = 5
expected["changed"] = True expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": "a.b", "value": 5}]} self._plugin._task.args = {"updates": [{"path": "a.b", "value": 5}]}
result = self._plugin.run(task_vars=task_vars) result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected) self.assertEqual(result, expected)
def test_run_multiple(self): def test_run_multiple(self):
"""Confirm multiple paths passes""" """Confirm multiple paths passes"""
task_vars = {"a": {"b1": [1, 2, 3], "b2": {"c": "123", "d": False}}} task_vars = {"vars": {"a": {"b1": [1, 2, 3], "b2": {"c": "123", "d": False}}}}
expected = {"a": {"b1": [1, 2, 3, 4], "b2": {"c": 456, "d": True}}, "changed": True} expected = {"a": {"b1": [1, 2, 3, 4], "b2": {"c": 456, "d": True}}}
expected.update({"changed": True})
self._plugin._task.args = { self._plugin._task.args = {
"updates": [ "updates": [
{"path": "a.b1.3", "value": 4}, {"path": "a.b1.3", "value": 4},
@ -166,60 +167,60 @@ class TestUpdate_Fact(TestCase):
def test_run_replace_in_list(self): def test_run_replace_in_list(self):
"""Replace in list""" """Replace in list"""
task_vars = {"a": {"b": [1, 2, 3]}} task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars) expected = copy.deepcopy(task_vars["vars"])
expected["a"]["b"][1] = 5 expected["a"]["b"][1] = 5
expected["changed"] = True expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": "a.b.1", "value": 5}]} self._plugin._task.args = {"updates": [{"path": "a.b.1", "value": 5}]}
result = self._plugin.run(task_vars=task_vars) result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected) self.assertEqual(result, expected)
def test_run_append_to_list(self): def test_run_append_to_list(self):
"""Append to list""" """Append to list"""
task_vars = {"a": {"b": [1, 2, 3]}} task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars) expected = copy.deepcopy(task_vars["vars"])
expected["a"]["b"].append(4) expected["a"]["b"].append(4)
expected["changed"] = True expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": "a.b.3", "value": 4}]} self._plugin._task.args = {"updates": [{"path": "a.b.3", "value": 4}]}
result = self._plugin.run(task_vars=task_vars) result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected) self.assertEqual(result, expected)
def test_run_bracket_single_quote(self): def test_run_bracket_single_quote(self):
"""Bracket notation sigle quote""" """Bracket notation sigle quote"""
task_vars = {"a": {"b": [1, 2, 3]}} task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars) expected = copy.deepcopy(task_vars["vars"])
expected["a"]["b"].append(4) expected["a"]["b"].append(4)
expected["changed"] = True expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": "a['b'][3]", "value": 4}]} self._plugin._task.args = {"updates": [{"path": "a['b'][3]", "value": 4}]}
result = self._plugin.run(task_vars=task_vars) result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected) self.assertEqual(result, expected)
def test_run_bracket_double_quote(self): def test_run_bracket_double_quote(self):
"""Bracket notation double quote""" """Bracket notation double quote"""
task_vars = {"a": {"b": [1, 2, 3]}} task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars) expected = copy.deepcopy(task_vars["vars"])
expected["a"]["b"].append(4) expected["a"]["b"].append(4)
expected["changed"] = True expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": 'a["b"][3]', "value": 4}]} self._plugin._task.args = {"updates": [{"path": 'a["b"][3]', "value": 4}]}
result = self._plugin.run(task_vars=task_vars) result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected) self.assertEqual(result, expected)
def test_run_int_dict_keys(self): def test_run_int_dict_keys(self):
"""Integer dict keys""" """Integer dict keys"""
task_vars = {"a": {0: [1, 2, 3]}} task_vars = {"vars": {"a": {0: [1, 2, 3]}}}
expected = copy.deepcopy(task_vars) expected = copy.deepcopy(task_vars["vars"])
expected["a"][0][0] = 0 expected["a"][0][0] = 0
expected["changed"] = True expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": "a.0.0", "value": 0}]} self._plugin._task.args = {"updates": [{"path": "a.0.0", "value": 0}]}
result = self._plugin.run(task_vars=task_vars) result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected) self.assertEqual(result, expected)
def test_run_int_as_string(self): def test_run_int_as_string(self):
"""Integer dict keys as string""" """Integer dict keys as string"""
task_vars = {"a": {"0": [1, 2, 3]}} task_vars = {"vars": {"a": {"0": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars) expected = copy.deepcopy(task_vars["vars"])
expected["a"]["0"][0] = 0 expected["a"]["0"][0] = 0
expected["changed"] = True expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": 'a["0"].0', "value": 0}]} self._plugin._task.args = {"updates": [{"path": 'a["0"].0', "value": 0}]}
result = self._plugin.run(task_vars=task_vars) result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected) self.assertEqual(result, expected)
@ -228,44 +229,46 @@ class TestUpdate_Fact(TestCase):
"""Invalid path format""" """Invalid path format"""
self._plugin._task.args = {"updates": [{"path": "a.'b'", "value": 0}]} self._plugin._task.args = {"updates": [{"path": "a.'b'", "value": 0}]}
with self.assertRaises(Exception) as error: with self.assertRaises(Exception) as error:
self._plugin.run(task_vars={}) self._plugin.run(task_vars={"vars": {}})
self.assertIn("malformed", str(error.exception)) self.assertIn("malformed", str(error.exception))
def test_run_invalid_path_bracket_after_dot(self): def test_run_invalid_path_bracket_after_dot(self):
"""Invalid path format""" """Invalid path format"""
self._plugin._task.args = {"updates": [{"path": "a.['b']", "value": 0}]} self._plugin._task.args = {"updates": [{"path": "a.['b']", "value": 0}]}
with self.assertRaises(Exception) as error: with self.assertRaises(Exception) as error:
self._plugin.run(task_vars={}) self._plugin.run(task_vars={"vars": {}})
self.assertIn("malformed", str(error.exception)) self.assertIn("malformed", str(error.exception))
def test_run_invalid_key_start_with_dot(self): def test_run_invalid_key_start_with_dot(self):
"""Invalid key format""" """Invalid key format"""
self._plugin._task.args = {"updates": [{"path": ".abc", "value": 0}]} self._plugin._task.args = {"updates": [{"path": ".abc", "value": 0}]}
with self.assertRaises(Exception) as error: with self.assertRaises(Exception) as error:
self._plugin.run(task_vars={}) self._plugin.run(task_vars={"vars": {}})
self.assertIn("malformed", str(error.exception)) self.assertIn("malformed", str(error.exception))
def test_run_no_update_list(self): def test_run_no_update_list(self):
"""Confirm no change when same""" """Confirm no change when same"""
task_vars = {"a": {"b": [1, 2, 3]}} task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars) expected = copy.deepcopy(task_vars["vars"])
expected["changed"] = False expected["a"]["b"] = [1, 2, 3]
expected.update({"changed": False})
self._plugin._task.args = {"updates": [{"path": "a.b.0", "value": 1}]} self._plugin._task.args = {"updates": [{"path": "a.b.0", "value": 1}]}
result = self._plugin.run(task_vars=task_vars) result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected) self.assertEqual(result, expected)
def test_run_no_update_dict(self): def test_run_no_update_dict(self):
"""Confirm no change when same""" """Confirm no change when same"""
task_vars = {"a": {"b": [1, 2, 3]}} task_vars = {"vars": {"a": {"b": [1, 2, 3]}}}
expected = copy.deepcopy(task_vars) expected = copy.deepcopy(task_vars["vars"])
expected["changed"] = False expected["a"]["b"] = [1, 2, 3]
expected.update({"changed": False})
self._plugin._task.args = {"updates": [{"path": "a.b", "value": [1, 2, 3]}]} self._plugin._task.args = {"updates": [{"path": "a.b", "value": [1, 2, 3]}]}
result = self._plugin.run(task_vars=task_vars) result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected) self.assertEqual(result, expected)
def test_run_missing_key(self): def test_run_missing_key(self):
"""Confirm error when key not found""" """Confirm error when key not found"""
task_vars = {"a": {"b": 1}} task_vars = {"vars": {"a": {"b": 1}}}
self._plugin._task.args = {"updates": [{"path": "a.c.d", "value": 1}]} self._plugin._task.args = {"updates": [{"path": "a.c.d", "value": 1}]}
with self.assertRaises(Exception) as error: with self.assertRaises(Exception) as error:
self._plugin.run(task_vars=task_vars) self._plugin.run(task_vars=task_vars)
@ -273,7 +276,7 @@ class TestUpdate_Fact(TestCase):
def test_run_list_not_int(self): def test_run_list_not_int(self):
"""Confirm error when key not found""" """Confirm error when key not found"""
task_vars = {"a": {"b": [1]}} task_vars = {"vars": {"a": {"b": [1]}}}
self._plugin._task.args = {"updates": [{"path": "a.b['0']", "value": 2}]} self._plugin._task.args = {"updates": [{"path": "a.b['0']", "value": 2}]}
with self.assertRaises(Exception) as error: with self.assertRaises(Exception) as error:
self._plugin.run(task_vars=task_vars) self._plugin.run(task_vars=task_vars)
@ -281,7 +284,7 @@ class TestUpdate_Fact(TestCase):
def test_run_list_not_long(self): def test_run_list_not_long(self):
"""List not long enough""" """List not long enough"""
task_vars = {"a": {"b": [0]}} task_vars = {"vars": {"a": {"b": [0]}}}
self._plugin._task.args = {"updates": [{"path": "a.b.2", "value": 2}]} self._plugin._task.args = {"updates": [{"path": "a.b.2", "value": 2}]}
with self.assertRaises(Exception) as error: with self.assertRaises(Exception) as error:
self._plugin.run(task_vars=task_vars) self._plugin.run(task_vars=task_vars)
@ -300,23 +303,27 @@ class TestUpdate_Fact(TestCase):
def test_run_not_dotted_success_one(self): def test_run_not_dotted_success_one(self):
"""Test with a not dotted key""" """Test with a not dotted key"""
task_vars = {"a": 0} task_vars = {"vars": {"a": 0}}
expected = {"a": 1, "ansible_facts": {"a": 1}, "changed": True} expected = copy.deepcopy(task_vars["vars"])
expected["a"] = 1
expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": "a", "value": 1}]} self._plugin._task.args = {"updates": [{"path": "a", "value": 1}]}
result = self._plugin.run(task_vars=task_vars) result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected) self.assertEqual(result, expected)
def test_run_not_dotted_success_three(self): def test_run_not_dotted_success_three(self):
"""Test with a not dotted key longer""" """Test with a not dotted key longer"""
task_vars = {"abc": 0} task_vars = {"vars": {"abc": 0}}
expected = {"abc": 1, "ansible_facts": {"abc": 1}, "changed": True} expected = copy.deepcopy(task_vars["vars"])
expected["abc"] = 1
expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": "abc", "value": 1}]} self._plugin._task.args = {"updates": [{"path": "abc", "value": 1}]}
result = self._plugin.run(task_vars=task_vars) result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected) self.assertEqual(result, expected)
def test_run_not_dotted_fail_missing(self): def test_run_not_dotted_fail_missing(self):
"""Test with a not dotted key, missing""" """Test with a not dotted key, missing"""
task_vars = {"abc": 0} task_vars = {"vars": {"abc": 0}}
self._plugin._task.args = {"updates": [{"path": "123", "value": 1}]} self._plugin._task.args = {"updates": [{"path": "123", "value": 1}]}
with self.assertRaises(Exception) as error: with self.assertRaises(Exception) as error:
self._plugin.run(task_vars=task_vars) self._plugin.run(task_vars=task_vars)
@ -324,19 +331,19 @@ class TestUpdate_Fact(TestCase):
def test_run_not_dotted_success_same(self): def test_run_not_dotted_success_same(self):
"""Test with a not dotted key, no change""" """Test with a not dotted key, no change"""
task_vars = {"a": 0} task_vars = {"vars": {"a": 0}}
expected = copy.deepcopy(task_vars) expected = copy.deepcopy(task_vars["vars"])
expected["changed"] = False expected.update({"changed": False})
self._plugin._task.args = {"updates": [{"path": "a", "value": 0}]} self._plugin._task.args = {"updates": [{"path": "a", "value": 0}]}
result = self._plugin.run(task_vars=task_vars) result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected) self.assertEqual(result, expected)
def test_run_looks_like_a_bool(self): def test_run_looks_like_a_bool(self):
"""Test with a key that looks like a bool""" """Test with a key that looks like a bool"""
task_vars = {"a": {"True": 0}} task_vars = {"vars": {"a": {"True": 0}}}
expected = copy.deepcopy(task_vars) expected = copy.deepcopy(task_vars["vars"])
expected["a"]["True"] = 1 expected["a"]["True"] = 1
expected["changed"] = True expected.update({"changed": True})
self._plugin._task.args = {"updates": [{"path": "a['True']", "value": 1}]} self._plugin._task.args = {"updates": [{"path": "a['True']", "value": 1}]}
result = self._plugin.run(task_vars=task_vars) result = self._plugin.run(task_vars=task_vars)
self.assertEqual(result, expected) self.assertEqual(result, expected)

View file

@ -104,7 +104,7 @@ VALID_DATA = {"name": "ansible", "email": "ansible@redhat.com"}
IN_VALID_DATA = {"name": "ansible", "email": "redhatcom"} IN_VALID_DATA = {"name": "ansible", "email": "redhatcom"}
CRITERIA_FORMAT_SUPPORT_CHECK = { CRITERIA_FORMAT_SUPPORT_CHECK = {
"$schema": "https://json-schema.org/draft/2020-12/schema", "$schema": "https://json-schema.org/schema#",
"type": "object", "type": "object",
"properties": {"name": {"type": "string"}, "email": {"format": "email"}}, "properties": {"name": {"type": "string"}, "email": {"format": "email"}},
"required": ["email"], "required": ["email"],

View file

@ -22,12 +22,8 @@ VALID_DATA = (
"<schemas><schema/></schemas></netconf-state>" "<schemas><schema/></schemas></netconf-state>"
) )
OUTPUT = { OUTPUT = """{"netconf-state": \
"netconf-state": { {"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-netconf-monitoring", "schemas": {"schema": null}}}"""
"@xmlns": "urn:ietf:params:xml:ns:yang:ietf-netconf-monitoring",
"schemas": {"schema": None},
},
}
class TestFromXml(TestCase): class TestFromXml(TestCase):

View file

@ -16,9 +16,11 @@ from unittest import TestCase
import pytest import pytest
from ansible.errors import AnsibleFilterError from ansible.errors import AnsibleFilterError
from ansible.template import AnsibleUndefined
from ansible_collections.ansible.utils.plugins.filter.cidr_merge import cidr_merge from ansible_collections.ansible.utils.plugins.filter.cidr_merge import cidr_merge
from ansible_collections.ansible.utils.plugins.filter.ip4_hex import ip4_hex from ansible_collections.ansible.utils.plugins.filter.ip4_hex import ip4_hex
from ansible_collections.ansible.utils.plugins.filter.ipaddr import _ipaddr
from ansible_collections.ansible.utils.plugins.filter.ipmath import ipmath from ansible_collections.ansible.utils.plugins.filter.ipmath import ipmath
from ansible_collections.ansible.utils.plugins.filter.ipsubnet import ipsubnet from ansible_collections.ansible.utils.plugins.filter.ipsubnet import ipsubnet
from ansible_collections.ansible.utils.plugins.filter.network_in_network import network_in_network from ansible_collections.ansible.utils.plugins.filter.network_in_network import network_in_network
@ -63,6 +65,15 @@ class TestIpFilter(TestCase):
self.assertEqual(cidr_merge(subnets), ["1.12.1.1/32", "1.12.1.255/32"]) self.assertEqual(cidr_merge(subnets), ["1.12.1.1/32", "1.12.1.255/32"])
self.assertEqual(cidr_merge(subnets, "span"), "1.12.1.0/24") self.assertEqual(cidr_merge(subnets, "span"), "1.12.1.0/24")
def test_ipaddr_undefined_value(self):
"""Check ipaddr filter undefined value"""
args = ["", AnsibleUndefined(name="my_ip"), ""]
with pytest.raises(
AnsibleFilterError,
match="Unrecognized type <<class 'ansible.template.AnsibleUndefined'>> for ipaddr filter <value>",
):
_ipaddr(*args)
def test_ipaddr_empty_query(self): def test_ipaddr_empty_query(self):
self.assertEqual(ipaddr("192.0.2.230"), "192.0.2.230") self.assertEqual(ipaddr("192.0.2.230"), "192.0.2.230")
self.assertEqual(ipaddr("192.0.2.230/30"), "192.0.2.230/30") self.assertEqual(ipaddr("192.0.2.230/30"), "192.0.2.230/30")

View file

@ -21,30 +21,16 @@ class TestIpCut(TestCase):
def setUp(self): def setUp(self):
pass pass
def test_get_last_X_bits_ipv6(self): def test_get_last_X_bits(self):
"""Get last X bits of Ipv6 address""" """Get last X bits of Ipv6 address"""
args = ["", "1234:4321:abcd:dcba::17", -80] args = ["", "1234:4321:abcd:dcba::17", -80]
result = _ipcut(*args) result = _ipcut(*args)
self.assertEqual(result, "dcba:0:0:0:17") self.assertEqual(result, "dcba:0:0:0:17")
def test_get_first_X_bits_ipv6(self): def test_get_first_X_bits(self):
"""Get first X bits of Ipv6 address""" """Get first X bits of Ipv6 address"""
args = ["", "1234:4321:abcd:dcba::17", 64] args = ["", "1234:4321:abcd:dcba::17", 64]
result = _ipcut(*args) result = _ipcut(*args)
self.assertEqual(result, "1234:4321:abcd:dcba") self.assertEqual(result, "1234:4321:abcd:dcba")
def test_get_last_X_bits_ipv4(self):
"""Get last X bits of Ipv4 address"""
args = ["", "10.2.3.0", -16]
result = _ipcut(*args)
self.assertEqual(result, "3.0")
def test_get_first_X_bits_ipv4(self):
"""Get first X bits of Ipv4 address"""
args = ["", "10.2.3.0", 24]
result = _ipcut(*args)
self.assertEqual(result, "10.2.3")

View file

@ -14,6 +14,11 @@ __metaclass__ = type
from unittest import TestCase from unittest import TestCase
import pytest
from ansible.errors import AnsibleFilterError
from ansible.template import AnsibleUndefined
from ansible_collections.ansible.utils.plugins.filter.ipv4 import _ipv4 from ansible_collections.ansible.utils.plugins.filter.ipv4 import _ipv4
@ -40,6 +45,15 @@ class TestIp4(TestCase):
def setUp(self): def setUp(self):
pass pass
def test_ipv4_undefined_value(self):
"""Check ipv4 filter undefined value"""
args = ["", AnsibleUndefined(name="my_ip"), ""]
with pytest.raises(
AnsibleFilterError,
match="Unrecognized type <<class 'ansible.template.AnsibleUndefined'>> for ipv4 filter <value>",
):
_ipv4(*args)
def test_ipv4_filter_empty_query(self): def test_ipv4_filter_empty_query(self):
"""Check ipv4 filter empty query""" """Check ipv4 filter empty query"""
args = ["", VALID_DATA, ""] args = ["", VALID_DATA, ""]

View file

@ -14,6 +14,11 @@ __metaclass__ = type
from unittest import TestCase from unittest import TestCase
import pytest
from ansible.errors import AnsibleFilterError
from ansible.template import AnsibleUndefined
from ansible_collections.ansible.utils.plugins.filter.ipv6 import _ipv6 from ansible_collections.ansible.utils.plugins.filter.ipv6 import _ipv6
@ -43,6 +48,15 @@ class TestIp6(TestCase):
def setUp(self): def setUp(self):
pass pass
def test_ipv6_undefined_value(self):
"""Check ipv6 filter undefined value"""
args = ["", AnsibleUndefined(name="my_ip"), ""]
with pytest.raises(
AnsibleFilterError,
match="Unrecognized type <<class 'ansible.template.AnsibleUndefined'>> for ipv6 filter <value>",
):
_ipv6(*args)
def test_ipv6_filter_empty_query(self): def test_ipv6_filter_empty_query(self):
"""Check ipv6 filter empty query""" """Check ipv6 filter empty query"""
args = ["", VALID_DATA, ""] args = ["", VALID_DATA, ""]

View file

@ -14,6 +14,11 @@ __metaclass__ = type
from unittest import TestCase from unittest import TestCase
import pytest
from ansible.errors import AnsibleFilterError
from ansible.template import AnsibleUndefined
from ansible_collections.ansible.utils.plugins.filter.ipwrap import _ipwrap from ansible_collections.ansible.utils.plugins.filter.ipwrap import _ipwrap
@ -45,6 +50,15 @@ class TestIpWrap(TestCase):
def setUp(self): def setUp(self):
pass pass
def test_ipwrap_undefined_value(self):
"""Check ipwrap filter undefined value"""
args = ["", AnsibleUndefined(name="my_ip"), ""]
with pytest.raises(
AnsibleFilterError,
match="Unrecognized type <<class 'ansible.template.AnsibleUndefined'>> for ipwrap filter <value>",
):
_ipwrap(*args)
def test_valid_data_list(self): def test_valid_data_list(self):
"""Check passing valid argspec(list)""" """Check passing valid argspec(list)"""
args = ["", VALID_DATA, ""] args = ["", VALID_DATA, ""]

View file

@ -230,29 +230,3 @@ class TestKeepKeys(TestCase):
with self.assertRaises(AnsibleFilterError) as error: with self.assertRaises(AnsibleFilterError) as error:
_keep_keys(*args) _keep_keys(*args)
self.assertIn("Error when using plugin 'keep_keys'", str(error.exception)) self.assertIn("Error when using plugin 'keep_keys'", str(error.exception))
def test_keep_filter_excludes_list_values(self):
"""Test that keep_keys excludes keys with list values that are not in target."""
data = [
{
"k0": "A",
"k1": "B",
"k2": [1],
"k3": "foo",
},
{
"k0": "A",
"k1": "B",
"k2": [2],
"k3": "bar",
},
]
target = ["k0", "k1"]
output = [
{"k0": "A", "k1": "B"},
{"k0": "A", "k1": "B"},
]
args = ["", data, target]
result = _keep_keys(*args)
self.assertEqual(result, output)

View file

@ -9,29 +9,17 @@ from __future__ import absolute_import, division, print_function
__metaclass__ = type __metaclass__ = type
import re
from unittest import TestCase from unittest import TestCase
from jinja2 import Environment from ansible.template import Templar
from ansible_collections.ansible.utils.plugins.plugin_utils.index_of import index_of from ansible_collections.ansible.utils.plugins.plugin_utils.index_of import index_of
def jinja_match(value, pattern):
return re.match(pattern, value) is not None
def jinja_search(value, pattern):
return re.search(pattern, value) is not None
class TestIndexOfFilter(TestCase): class TestIndexOfFilter(TestCase):
def setUp(self): def setUp(self):
env = Environment() self._tests = Templar(loader=None).environment.tests
env.tests["match"] = jinja_match
env.tests["search"] = jinja_search
self._tests = env.tests
def test_fail_no_qualfier(self): def test_fail_no_qualfier(self):
obj, test, value = [1, 2], "@@", 1 obj, test, value = [1, 2], "@@", 1
@ -112,14 +100,14 @@ class TestIndexOfFilter(TestCase):
), ),
( (
[{"a": "abc"}, {"a": "def"}, {"a": "ghi"}, {"a": "jkl"}], [{"a": "abc"}, {"a": "def"}, {"a": "ghi"}, {"a": "jkl"}],
"match", "ansible.builtin.match",
"^a", "^a",
"a", "a",
0, 0,
), ),
( (
[{"a": "abc"}, {"a": "def"}, {"a": "ghi"}, {"a": "jkl"}], [{"a": "abc"}, {"a": "def"}, {"a": "ghi"}, {"a": "jkl"}],
"search", "ansible.builtin.search",
"e", "e",
"a", "a",
1, 1,

View file

@ -1,60 +0,0 @@
# -*- coding: utf-8 -*-
# Copyright 2026 Red Hat
# GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
"""
Unit tests for ipaddress_utils helpers.
"""
from __future__ import absolute_import, division, print_function
__metaclass__ = type
from unittest import TestCase
from ansible_collections.ansible.utils.plugins.plugin_utils.base.ipaddress_utils import (
_is_subnet_of,
ip_network,
)
class TestIsSubnetOf(TestCase):
def test_ipv4_subnet_true(self):
a = ip_network("10.1.1.0/24")
b = ip_network("10.0.0.0/8")
self.assertTrue(_is_subnet_of(a, b))
def test_ipv4_subnet_false_disjoint(self):
a = ip_network("192.168.1.0/24")
b = ip_network("10.0.0.0/8")
self.assertFalse(_is_subnet_of(a, b))
def test_ipv4_subnet_false_reverse(self):
a = ip_network("10.0.0.0/8")
b = ip_network("10.1.1.0/24")
self.assertFalse(_is_subnet_of(a, b))
def test_ipv4_same_network(self):
n = ip_network("10.0.0.0/8")
self.assertTrue(_is_subnet_of(n, n))
def test_ipv6_subnet_true(self):
a = ip_network("2001:db8:0:1::/64")
b = ip_network("2001:db8::/48")
self.assertTrue(_is_subnet_of(a, b))
def test_ipv6_subnet_false(self):
a = ip_network("2001:db8:2::/64")
b = ip_network("2001:db8:1::/48")
self.assertFalse(_is_subnet_of(a, b))
def test_mixed_ip_versions(self):
a = ip_network("10.0.0.0/8")
b = ip_network("2001:db8::/32")
self.assertFalse(_is_subnet_of(a, b))
def test_invalid_arguments_return_false(self):
self.assertFalse(_is_subnet_of(None, ip_network("10.0.0.0/8")))
self.assertFalse(_is_subnet_of(ip_network("10.0.0.0/8"), None))

View file

@ -52,40 +52,6 @@ class TestTextfsmParser(TestCase):
] ]
self.assertEqual(result["parsed"][0][0], parsed_output) self.assertEqual(result["parsed"][0][0], parsed_output)
def test_ttp_parser_flat_list(self):
"""With ttp_results.results: flat_list, output is a single-level list of items."""
nxos_cfg_path = os.path.join(os.path.dirname(__file__), "fixtures", "nxos_show_version.cfg")
nxos_template_path = os.path.join(
os.path.dirname(__file__),
"fixtures",
"nxos_show_version.ttp",
)
with open(nxos_cfg_path) as fhand:
nxos_show_version_output = fhand.read()
task_args = {
"text": nxos_show_version_output,
"parser": {
"name": "ansible.utils.ttp",
"command": "show version",
"template_path": nxos_template_path,
"vars": {
"ttp_results": {"results": "flat_list"},
},
},
}
parser = CliParser(task_args=task_args, task_vars=[], debug=False)
result = parser.parse()
self.assertNotIn("errors", result)
parsed = result["parsed"]
self.assertIsInstance(parsed, list)
self.assertGreaterEqual(len(parsed), 1)
self.assertIsInstance(parsed[0], dict)
self.assertIn("boot_image", parsed[0])
self.assertIn("os", parsed[0])
def test_textfsm_parser_invalid_parser(self): def test_textfsm_parser_invalid_parser(self):
fake_path = "/ /I hope this doesn't exist" fake_path = "/ /I hope this doesn't exist"
task_args = { task_args = {

View file

@ -9,7 +9,7 @@ __metaclass__ = type
import pytest import pytest
from ansible.errors import AnsibleError from ansible.errors import AnsibleError
from ansible.module_utils.common.text.converters import to_text from ansible.module_utils._text import to_text
from ansible_collections.ansible.utils.plugins.plugin_utils.base.validate import _load_validator from ansible_collections.ansible.utils.plugins.plugin_utils.base.validate import _load_validator

View file

@ -1,2 +1,9 @@
[tox] [ansible]
ignore_base_python_conflict = true skip =
py3.7
py3.8
2.9
2.10
2.11
2.12
2.13

View file

@ -1,12 +1,12 @@
--- ---
- name: Install and configure Certbot - name: Install and configure Certbot
hosts: vm01-balancer hosts: router
become: true become: true
gather_facts: true gather_facts: true
vars: vars:
certbot_install_method: package certbot_install_method: package
certbot_admin_email: admin@gigacoms.ru certbot_admin_email: mrwho@mrwho.ru
certbot_auto_renew: true certbot_auto_renew: true
certbot_auto_renew_user: "{{ ansible_user | default(lookup('env', 'USER')) }}" certbot_auto_renew_user: "{{ ansible_user | default(lookup('env', 'USER')) }}"
certbot_auto_renew_hour: "5" certbot_auto_renew_hour: "5"
@ -18,16 +18,17 @@
certbot_delete_certs: false certbot_delete_certs: false
certbot_certs: certbot_certs:
- domains: - domains:
- gigacoms.ru - local.mrwho.ru
- "*.gigacoms.ru" - "*.mrwho.ru"
- "*.local.mrwho.ru"
certbot_testmode: false certbot_testmode: false
# Generate key with: tsig-keygen -a HMAC-SHA512 CertbotKey # Generate key with: tsig-keygen -a HMAC-SHA512 CertbotKey
certbot_dns_plugin: rfc2136 certbot_dns_plugin: rfc2136
certbot_dns_target_server: 91.218.87.251 certbot_dns_target_server:
certbot_dns_target_server_port: 53 certbot_dns_target_server_port: 53
certbot_dns_tsig_keyname: CertbotKey certbot_dns_tsig_keyname: certbot-key
certbot_dns_key_secret: hbxr+SNi3kzzygeArFp/1xgribKLDLZEfRWWh/btH6jauMrcJ0oPyOEuP11d1Oq10HUJnTKnOtb8YlnxrM4HcA== certbot_dns_key_secret: YCUScDPbF9bscPFB0mLzRr2/V54FB7tcrifAdrrnlBI=
certbot_dns_key_algorithm: HMAC-SHA512 certbot_dns_key_algorithm: HMAC-SHA256
roles: roles:
- certbot-dns - certbot-dns

View file

@ -0,0 +1,144 @@
---
- name: Install balancer nginx
hosts: router
roles:
- nginx
# - firewall
# - chrony
vars:
###### NGINX Configuration ######
nginx_worker_connections: 16384
nginx_use_epoll: true
nginx_multi_accept: "on"
nginx_selinux: false
pki_dir: /etc/nginx/certs
nginx_webroot: /var/www
local_http_port: 8080
nginx_letsencrypt_managed: true
letsencrypt_acme_install: true
letsencrypt_acme_certs_dir: /etc/letsencrypt/live
nginx_conf_snippets:
- nginx-compression.conf
- nginx-proxy-params.conf
- nginx-server-ssl.conf
# - nginx-websockets.conf
# - nginx-browser-cache.conf
# - letsencrypt-proxy.conf
# - nginx-cors.conf
nginx_streams:
- stream: NTP
listen_port: 123
protocol: udp
responses: 1
timeout: 3
backends:
- 10.203.2.11:123
- 10.203.2.12:123
nginx_virthosts:
- virthost_name: git.local.mrwho.ru
listen: "80"
server_name: git.local.mrwho.ru
server_aliases: git.mrwho.ru
# index: index.php
# error_page: /path_to_error_page.html
ssl_enabled: true
ssl_only: true
ssl_letsencrypt_certs: "{{ nginx_letsencrypt_managed }}"
# root: "{{ nginx_webroot }}"
server_tokens: "off"
proxy_standard_setup: true
proxy_additional_options:
- proxy_cache_path /tmp/nginx_cache levels=1:2 keys_zone=cache:30m max_size=250m;
locations:
- location: /
target: "http://172.20.21.105:3000"
###### Firewalld configuration ######
firewall:
# - previous: replaced # flush previous rules
- zone: internal
state: enabled
permanent: true
service:
- ssh
- ntp
- dns
- http
- https
source:
- 10.0.0.0/8
- 172.16.0.0/12
- 192.168.0.0/16
- zone: internal
state: disabled
permanent: true
service:
- cockpit
- dhcpv6-client
- mdns
- samba-client
- zone: public
state: enabled
permanent: true
service:
- ntp
- dns
- http
- https
- zone: external
state: disabled
permanent: true
service:
- ssh
- cockpit
- dhcpv6-client
- mdns
- samba-client
- set_default_zone: internal
state: enabled
permanent: true
- interface: eth1
zone: internal
state: enabled
permanent: true
- interface: eth2
zone: internal
state: enabled
permanent: true
- interface: eth0
zone: external
state: enabled
permanent: true
- interface: tun0
zone: external
state: enabled
permanent: true
- interface: tun1
zone: external
state: enabled
permanent: true
###### chrony-client ######
chrony_disable_ntpd: true
chrony_enable: true
chrony_log_enable: true
chrony_log: measurements statistics tracking
chrony_logdir: /var/log/chrony/
chrony_ntp_servers:
- 10.203.2.11
- 10.203.2.12
chrony_rtcsync: true
chrony_timezone: Europe/Moscow

View file

@ -148,12 +148,18 @@
ansible.builtin.file: ansible.builtin.file:
path: /etc/systemd/system/nginx.service.d path: /etc/systemd/system/nginx.service.d
state: directory state: directory
owner: root
group: root
mode: 0755
when: nginx_streams | default([]) | selectattr('transparent', 'defined') | selectattr('transparent') | list | length > 0 when: nginx_streams | default([]) | selectattr('transparent', 'defined') | selectattr('transparent') | list | length > 0
- name: Systemd override для nginx — CAP_NET_RAW (transparent proxy) - name: Systemd override для nginx — CAP_NET_RAW (transparent proxy)
ansible.builtin.template: ansible.builtin.template:
src: nginx-systemd-override.j2 src: nginx-systemd-override.j2
dest: /etc/systemd/system/nginx.service.d/transparent.conf dest: /etc/systemd/system/nginx.service.d/transparent.conf
owner: root
group: root
mode: 0644
when: nginx_streams | default([]) | selectattr('transparent', 'defined') | selectattr('transparent') | list | length > 0 when: nginx_streams | default([]) | selectattr('transparent', 'defined') | selectattr('transparent') | list | length > 0
notify: notify:
- Daemon reload - Daemon reload

View file

@ -9,14 +9,14 @@
state: directory state: directory
owner: root owner: root
group: root group: root
# mode: "0755" mode: "0644"
- name: Install the nginx stream files - name: Install the nginx stream files
ansible.builtin.template: ansible.builtin.template:
src: nginx-stream.j2 src: nginx-stream.conf.j2
dest: /etc/nginx/stream.d/{{ item.stream }}.conf dest: /etc/nginx/stream.d/{{ item.stream }}.conf
owner: root owner: root
group: root group: root
# mode: "0444" mode: "0644"
loop: "{{ nginx_streams }}" loop: "{{ nginx_streams }}"
notify: Reload nginx notify: Reload nginx