GiteaTeam/demo-fastapi
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add collections

Browse Source
This commit is contained in:
Gary Kwok
2024-02-01 11:40:42 +08:00
parent 0715d7c475
commit 7af4c56c96
3198 changed files with 455118 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/__init__.py Normal file
View File

85
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/common.py Normal file
View File

@@ -0,0 +1,85 @@
# Copyright (c) 2022 Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
import json
from ansible_collections.community.general.plugins.lookup.onepassword import (
OnePassCLIv1,
OnePassCLIv2,
)
def load_file(file):
with open((os.path.join(os.path.dirname(__file__), "fixtures", file)), "r") as f:
return json.loads(f.read())
# Intentionally excludes metadata leaf nodes that would exist in real output if not relevant.
MOCK_ENTRIES = {
OnePassCLIv1: [
{
'vault_name': 'Acme "Quot\'d" Servers',
'queries': [
'0123456789',
'Mock "Quot\'d" Server'
],
'expected': ['t0pS3cret', 't0pS3cret'],
'output': load_file("v1_out_01.json"),
},
{
'vault_name': 'Acme Logins',
'queries': [
'9876543210',
'Mock Website',
'acme.com'
],
'expected': ['t0pS3cret', 't0pS3cret', 't0pS3cret'],
'output': load_file("v1_out_02.json"),
},
{
'vault_name': 'Acme Logins',
'queries': [
'864201357'
],
'expected': ['vauxhall'],
'output': load_file("v1_out_03.json"),
},
],
OnePassCLIv2: [
{
"vault_name": "Test Vault",
"queries": [
"ywvdbojsguzgrgnokmcxtydgdv",
"Authy Backup",
],
"expected": ["OctoberPoppyNuttyDraperySabbath", "OctoberPoppyNuttyDraperySabbath"],
"output": load_file("v2_out_01.json"),
},
{
# Request a custom field where ID and label are different
"vault_name": "Test Vault",
"queries": ["Dummy Login"],
"kwargs": {
"field": "password1",
},
"expected": ["data in custom field"],
"output": load_file("v2_out_02.json")
},
{
# Request data from a custom section
"vault_name": "Test Vault",
"queries": ["Duplicate Sections"],
"kwargs": {
"field": "s2 text",
"section": "Section 2",
},
"expected": ["first value"],
"output": load_file("v2_out_03.json")
},
],
}

39
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/conftest.py Normal file
View File

@@ -0,0 +1,39 @@
# Copyright (c) 2020 Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import pytest
from ansible_collections.community.general.plugins.lookup.onepassword import OnePass
OP_VERSION_FIXTURES = [
"opv1",
"opv2"
]
@pytest.fixture
def fake_op(mocker):
def _fake_op(version):
mocker.patch("ansible_collections.community.general.plugins.lookup.onepassword.OnePassCLIBase.get_current_version", return_value=version)
op = OnePass(None, None, None, None, None)
op._config._config_file_path = "/home/jin/.op/config"
mocker.patch.object(op._cli, "_run")
return op
return _fake_op
@pytest.fixture
def opv1(fake_op):
return fake_op("1.17.2")
@pytest.fixture
def opv2(fake_op):
return fake_op("2.27.2")

18
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/fixtures/v1_out_01.json Normal file
View File

@@ -0,0 +1,18 @@
{
"uuid": "0123456789",
"vaultUuid": "2468",
"overview": {
"title": "Mock \"Quot'd\" Server"
},
"details": {
"sections": [{
"title": "",
"fields": [
{"t": "username", "v": "jamesbond"},
{"t": "password", "v": "t0pS3cret"},
{"t": "notes", "v": "Test note with\nmultiple lines and trailing space.\n\n"},
{"t": "tricksy \"quot'd\" field\\", "v": "\"quot'd\" value"}
]
}]
}
}

3
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/fixtures/v1_out_01.json.license Normal file
View File

@@ -0,0 +1,3 @@
GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
SPDX-License-Identifier: GPL-3.0-or-later
SPDX-FileCopyrightText: 2022, Ansible Project

18
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/fixtures/v1_out_02.json Normal file
View File

@@ -0,0 +1,18 @@
{
"uuid": "9876543210",
"vaultUuid": "1357",
"overview": {
"title": "Mock Website",
"URLs": [
{"l": "website", "u": "https://acme.com/login"}
]
},
"details": {
"sections": [{
"title": "",
"fields": [
{"t": "password", "v": "t0pS3cret"}
]
}]
}
}

3
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/fixtures/v1_out_02.json.license Normal file
View File

@@ -0,0 +1,3 @@
GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
SPDX-License-Identifier: GPL-3.0-or-later
SPDX-FileCopyrightText: 2022, Ansible Project

20
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/fixtures/v1_out_03.json Normal file
View File

@@ -0,0 +1,20 @@
{
"uuid": "864201357",
"vaultUuid": "1357",
"overview": {
"title": "Mock Something"
},
"details": {
"fields": [
{
"value": "jbond@mi6.gov.uk",
"name": "emailAddress"
},
{
"name": "password",
"value": "vauxhall"
},
{}
]
}
}

3
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/fixtures/v1_out_03.json.license Normal file
View File

@@ -0,0 +1,3 @@
GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
SPDX-License-Identifier: GPL-3.0-or-later
SPDX-FileCopyrightText: 2022, Ansible Project

35
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/fixtures/v2_out_01.json Normal file
View File

@@ -0,0 +1,35 @@
{
"id": "ywvdbojsguzgrgnokmcxtydgdv",
"title": "Authy Backup",
"version": 1,
"vault": {
"id": "bcqxysvcnejjrwzoqrwzcqjqxc",
"name": "test vault"
},
"category": "PASSWORD",
"last_edited_by": "7FUPZ8ZNE02KSHMAIMKHIVUE17",
"created_at": "2015-01-18T13:13:38Z",
"updated_at": "2016-02-20T16:23:54Z",
"additional_information": "Jan 18, 2015, 08:13:38",
"fields": [
{
"id": "password",
"type": "CONCEALED",
"purpose": "PASSWORD",
"label": "password",
"value": "OctoberPoppyNuttyDraperySabbath",
"reference": "op://Test Vault/Authy Backup/password",
"password_details": {
"strength": "FANTASTIC"
}
},
{
"id": "notesPlain",
"type": "STRING",
"purpose": "NOTES",
"label": "notesPlain",
"value": "Backup password to restore Authy",
"reference": "op://Test Vault/Authy Backup/notesPlain"
}
]
}

3
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/fixtures/v2_out_01.json.license Normal file
View File

@@ -0,0 +1,3 @@
GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
SPDX-License-Identifier: GPL-3.0-or-later
SPDX-FileCopyrightText: 2022, Ansible Project

85
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/fixtures/v2_out_02.json Normal file
View File

@@ -0,0 +1,85 @@
{
"id": "awk4s2u44fhnrgppszcsvc663i",
"title": "Dummy Login",
"version": 4,
"vault": {
"id": "stpebbaccrq72xulgouxsk4p7y",
"name": "Personal"
},
"category": "LOGIN",
"last_edited_by": "LSGPJERUYBH7BFPHMZ2KKGL6AU",
"created_at": "2018-04-25T21:55:19Z",
"updated_at": "2022-09-02T17:51:21Z",
"additional_information": "agent.smith",
"urls": [
{
"primary": true,
"href": "https://acme.com"
}
],
"sections": [
{
"id": "add more"
},
{
"id": "gafaeg7vnqmgrklw5r6yrufyxy",
"label": "COMMANDS"
},
{
"id": "linked items",
"label": "Related Items"
}
],
"fields": [
{
"id": "username",
"type": "STRING",
"purpose": "USERNAME",
"label": "username",
"value": "agent.smith",
"reference": "op://Personal/Dummy Login/username"
},
{
"id": "password",
"type": "CONCEALED",
"purpose": "PASSWORD",
"label": "password",
"value": "FootworkDegreeReverence",
"entropy": 159.60836791992188,
"reference": "op://Personal/Dummy Login/password",
"password_details": {
"entropy": 159,
"generated": true,
"strength": "FANTASTIC"
}
},
{
"id": "notesPlain",
"type": "STRING",
"purpose": "NOTES",
"label": "notesPlain",
"reference": "op://Personal/Dummy Login/notesPlain"
},
{
"id": "7gyjekelk24ghgd4rvafspjbli",
"section": {
"id": "add more"
},
"type": "STRING",
"label": "title",
"value": "value of the field",
"reference": "op://Personal/Dummy Login/add more/title"
},
{
"id": "fx4wpzokrxn7tlb3uwpdjfptgm",
"section": {
"id": "gafaeg7vnqmgrklw5r6yrufyxy",
"label": "COMMANDS"
},
"type": "CONCEALED",
"label": "password1",
"value": "data in custom field",
"reference": "op://Personal/Dummy Login/COMMANDS/password1"
}
]
}

3
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/fixtures/v2_out_02.json.license Normal file
View File

@@ -0,0 +1,3 @@
GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
SPDX-License-Identifier: GPL-3.0-or-later
SPDX-FileCopyrightText: 2022, Ansible Project

103
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/fixtures/v2_out_03.json Normal file
View File

@@ -0,0 +1,103 @@
{
"id": "7t7qu2r35qyvqj3crujd4dqxmy",
"title": "Duplicate Sections",
"version": 3,
"vault": {
"id": "stpebbaccrq72xulgouxsk4p7y",
"name": "Personal"
},
"category": "LOGIN",
"last_edited_by": "LSGPJERUYBH7BFPHMZ2KKGL6AU",
"created_at": "2022-11-04T17:09:18Z",
"updated_at": "2022-11-04T17:22:19Z",
"additional_information": "flora",
"urls": [
{
"label": "website",
"primary": true,
"href": "https://acme.com/login"
}
],
"sections": [
{
"id": "add more"
},
{
"id": "7osqcvd43i75teocdzbb6d7mie",
"label": "Section 2"
}
],
"fields": [
{
"id": "username",
"type": "STRING",
"purpose": "USERNAME",
"label": "username",
"value": "flora",
"reference": "op://Personal/Duplicate Sections/username"
},
{
"id": "password",
"type": "CONCEALED",
"purpose": "PASSWORD",
"label": "password",
"value": "PtZGFLAibx-erTo7ywywEvh-n4syas97n-tuF2D.b8DdqA2vCjrvRGkNQxj!Gi9R",
"entropy": 379.564697265625,
"reference": "op://Personal/Duplicate Sections/password",
"password_details": {
"entropy": 379,
"generated": true,
"strength": "FANTASTIC"
}
},
{
"id": "notesPlain",
"type": "STRING",
"purpose": "NOTES",
"label": "notesPlain",
"reference": "op://Personal/Duplicate Sections/notesPlain"
},
{
"id": "4saaazkb7arwisj6ysctb4jmm4",
"section": {
"id": "add more"
},
"type": "STRING",
"label": "text",
"value": "text field the first",
"reference": "op://Personal/Duplicate Sections/add more/text"
},
{
"id": "4vtfkj4bwcmg7d5uf62wnpkp3a",
"section": {
"id": "add more"
},
"type": "STRING",
"label": "text",
"value": "text field the second",
"reference": "op://Personal/Duplicate Sections/add more/text"
},
{
"id": "wbrjnowkrgavpooomtht36gjqu",
"section": {
"id": "7osqcvd43i75teocdzbb6d7mie",
"label": "Section 2"
},
"type": "STRING",
"label": "s2 text",
"value": "first value",
"reference": "op://Personal/Duplicate Sections/Section 2/s2 text"
},
{
"id": "bddlz2fj2pebmtfhksbmcexy7m",
"section": {
"id": "7osqcvd43i75teocdzbb6d7mie",
"label": "Section 2"
},
"type": "STRING",
"label": "s2 text",
"value": "second value",
"reference": "op://Personal/Duplicate Sections/Section 2/s2 text"
}
]
}

3
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/fixtures/v2_out_03.json.license Normal file
View File

@@ -0,0 +1,3 @@
GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
SPDX-License-Identifier: GPL-3.0-or-later
SPDX-FileCopyrightText: 2022, Ansible Project

183
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/test_onepassword.py Normal file
View File

@@ -0,0 +1,183 @@
# Copyright (c) 2020 Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import operator
import itertools
import json
import pytest
from .conftest import OP_VERSION_FIXTURES
from .common import MOCK_ENTRIES
from ansible.errors import AnsibleLookupError
from ansible.plugins.loader import lookup_loader
from ansible_collections.community.general.plugins.lookup.onepassword import (
OnePassCLIv1,
OnePassCLIv2,
)
@pytest.mark.parametrize(
("version", "version_class"),
(
("1.17.2", OnePassCLIv1),
("2.27.4", OnePassCLIv2),
)
)
def test_op_correct_cli_class(fake_op, version, version_class):
op = fake_op(version)
assert op._cli.version == version
assert isinstance(op._cli, version_class)
def test_op_unsupported_cli_version(fake_op):
with pytest.raises(AnsibleLookupError, match="is unsupported"):
fake_op("99.77.77")
@pytest.mark.parametrize("op_fixture", OP_VERSION_FIXTURES)
def test_op_set_token_with_config(op_fixture, mocker, request):
op = request.getfixturevalue(op_fixture)
token = "F5417F77529B41B595D7F9D6F76EC057"
mocker.patch("os.path.isfile", return_value=True)
mocker.patch.object(op._cli, "signin", return_value=(0, token + "\n", ""))
op.set_token()
assert op.token == token
@pytest.mark.parametrize(
("op_fixture", "message"),
[
(op, value)
for op in OP_VERSION_FIXTURES
for value in
(
"Missing required parameters",
"The operation is unauthorized",
)
]
)
def test_op_set_token_with_config_missing_args(op_fixture, message, request, mocker):
op = request.getfixturevalue(op_fixture)
mocker.patch("os.path.isfile", return_value=True)
mocker.patch.object(op._cli, "signin", return_value=(99, "", ""), side_effect=AnsibleLookupError(message))
mocker.patch.object(op._cli, "full_signin", return_value=(0, "", ""))
with pytest.raises(AnsibleLookupError, match=message):
op.set_token()
op._cli.full_signin.assert_not_called()
@pytest.mark.parametrize("op_fixture", OP_VERSION_FIXTURES)
def test_op_set_token_with_config_full_signin(op_fixture, request, mocker):
op = request.getfixturevalue(op_fixture)
mocker.patch("os.path.isfile", return_value=True)
mocker.patch.object(op._cli, "signin", return_value=(99, "", ""), side_effect=AnsibleLookupError("Raised intentionally"))
mocker.patch.object(op._cli, "full_signin", return_value=(0, "", ""))
op.set_token()
op._cli.full_signin.assert_called()
@pytest.mark.parametrize("op_fixture", OP_VERSION_FIXTURES)
def test_op_set_token_without_config(op_fixture, request, mocker):
op = request.getfixturevalue(op_fixture)
token = "B988E8A2680A4A348962751A96861FA1"
mocker.patch("os.path.isfile", return_value=False)
mocker.patch.object(op._cli, "signin", return_value=(99, "", ""))
mocker.patch.object(op._cli, "full_signin", return_value=(0, token + "\n", ""))
op.set_token()
op._cli.signin.assert_not_called()
assert op.token == token
@pytest.mark.parametrize(
("op_fixture", "login_status"),
[(op, value) for op in OP_VERSION_FIXTURES for value in [False, True]]
)
def test_op_assert_logged_in(mocker, login_status, op_fixture, request):
op = request.getfixturevalue(op_fixture)
mocker.patch.object(op._cli, "assert_logged_in", return_value=login_status)
mocker.patch.object(op, "set_token")
op.assert_logged_in()
op._cli.assert_logged_in.assert_called_once()
assert op.logged_in == login_status
if not login_status:
op.set_token.assert_called_once()
@pytest.mark.parametrize("op_fixture", OP_VERSION_FIXTURES)
def test_op_get_raw_v1(mocker, op_fixture, request):
op = request.getfixturevalue(op_fixture)
mocker.patch.object(op._cli, "get_raw", return_value=[99, "RAW OUTPUT", ""])
result = op.get_raw("some item")
assert result == "RAW OUTPUT"
op._cli.get_raw.assert_called_once()
@pytest.mark.parametrize(
("op_fixture", "output", "expected"),
(
list(itertools.chain([op], d))
for op in OP_VERSION_FIXTURES
for d in [
("RAW OUTPUT", "RAW OUTPUT"),
(None, ""),
("", ""),
]
)
)
def test_op_get_field(mocker, op_fixture, output, expected, request):
op = request.getfixturevalue(op_fixture)
mocker.patch.object(op, "get_raw", return_value=output)
mocker.patch.object(op._cli, "_parse_field", return_value=output)
result = op.get_field("some item", "some field")
assert result == expected
# This test sometimes fails on older Python versions because the gathered tests mismatch.
# Sort the fixture data to make this reliable
# https://github.com/pytest-dev/pytest-xdist/issues/432
@pytest.mark.parametrize(
("cli_class", "vault", "queries", "kwargs", "output", "expected"),
(
(_cli_class, item["vault_name"], item["queries"], item.get("kwargs", {}), item["output"], item["expected"])
for _cli_class in sorted(MOCK_ENTRIES, key=operator.attrgetter("__name__"))
for item in MOCK_ENTRIES[_cli_class]
)
)
def test_op_lookup(mocker, cli_class, vault, queries, kwargs, output, expected):
mocker.patch("ansible_collections.community.general.plugins.lookup.onepassword.OnePass._get_cli_class", cli_class)
mocker.patch("ansible_collections.community.general.plugins.lookup.onepassword.OnePass.assert_logged_in", return_value=True)
mocker.patch("ansible_collections.community.general.plugins.lookup.onepassword.OnePassCLIBase._run", return_value=(0, json.dumps(output), ""))
op_lookup = lookup_loader.get("community.general.onepassword")
result = op_lookup.run(queries, vault=vault, **kwargs)
assert result == expected
@pytest.mark.parametrize("op_fixture", OP_VERSION_FIXTURES)
def test_signin(op_fixture, request):
op = request.getfixturevalue(op_fixture)
op._cli.master_password = "master_pass"
op._cli.signin()
print(op._cli.version)
op._cli._run.assert_called_once_with(['signin', '--raw'], command_input=b"master_pass")

50
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/test_onepassword_cli_v1.py Normal file
View File

@@ -0,0 +1,50 @@
# Copyright (c) 2022 Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import pytest
from ansible_collections.community.general.plugins.lookup.onepassword import OnePassCLIv1
@pytest.mark.parametrize(
("args", "rc", "expected_call_args", "expected_call_kwargs", "expected"),
(
([], 0, ["get", "account"], {"ignore_errors": True}, True,),
([], 1, ["get", "account"], {"ignore_errors": True}, False,),
(["acme"], 1, ["get", "account", "--account", "acme.1password.com"], {"ignore_errors": True}, False,),
)
)
def test_assert_logged_in(mocker, args, rc, expected_call_args, expected_call_kwargs, expected):
mocker.patch.object(OnePassCLIv1, "_run", return_value=[rc, "", ""])
op_cli = OnePassCLIv1(*args)
result = op_cli.assert_logged_in()
op_cli._run.assert_called_with(expected_call_args, **expected_call_kwargs)
assert result == expected
def test_full_signin(mocker):
mocker.patch.object(OnePassCLIv1, "_run", return_value=[0, "", ""])
op_cli = OnePassCLIv1(
subdomain="acme",
username="bob@acme.com",
secret_key="SECRET",
master_password="ONEKEYTORULETHEMALL",
)
result = op_cli.full_signin()
op_cli._run.assert_called_with([
"signin",
"acme.1password.com",
b"bob@acme.com",
b"SECRET",
"--raw",
], command_input=b"ONEKEYTORULETHEMALL")
assert result == [0, "", ""]

52
collections/ansible_collections/community/general/tests/unit/plugins/lookup/onepassword/test_onepassword_cli_v2.py Normal file
View File

@@ -0,0 +1,52 @@
# Copyright (c) 2022 Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import pytest
from ansible_collections.community.general.plugins.lookup.onepassword import OnePassCLIv2
@pytest.mark.parametrize(
("args", "out", "expected_call_args", "expected"),
(
([], "list of accounts", ["account", "get"], True,),
(["acme"], "list of accounts", ["account", "get", "--account", "acme.1password.com"], True,),
([], "", ["account", "list"], False,),
)
)
def test_assert_logged_in(mocker, args, out, expected_call_args, expected):
mocker.patch.object(OnePassCLIv2, "_run", return_value=[0, out, ""])
op_cli = OnePassCLIv2(*args)
result = op_cli.assert_logged_in()
op_cli._run.assert_called_with(expected_call_args)
assert result == expected
def test_full_signin(mocker):
mocker.patch.object(OnePassCLIv2, "_run", return_value=[0, "", ""])
op_cli = OnePassCLIv2(
subdomain="acme",
username="bob@acme.com",
secret_key="SECRET",
master_password="ONEKEYTORULETHEMALL",
)
result = op_cli.full_signin()
op_cli._run.assert_called_with(
[
"account", "add", "--raw",
"--address", "acme.1password.com",
"--email", b"bob@acme.com",
"--signin",
],
command_input=b"ONEKEYTORULETHEMALL",
environment_update={'OP_SECRET_KEY': 'SECRET'},
)
assert result == [0, "", ""]

162
collections/ansible_collections/community/general/tests/unit/plugins/lookup/test_bitwarden.py Normal file
View File

@@ -0,0 +1,162 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2022, Jonathan Lung <lungj@heresjono.com>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from argparse import ArgumentParser
from ansible_collections.community.general.tests.unit.compat import unittest
from ansible_collections.community.general.tests.unit.compat.mock import patch
from ansible.errors import AnsibleError
from ansible.module_utils import six
from ansible.plugins.loader import lookup_loader
from ansible_collections.community.general.plugins.lookup.bitwarden import LookupModule, Bitwarden, BitwardenException
MOCK_RECORDS = [
{
"collectionIds": [],
"deletedDate": None,
"favorite": False,
"fields": [
{
"linkedId": None,
"name": "a_new_secret",
"type": 1,
"value": "this is a new secret"
},
{
"linkedId": None,
"name": "not so secret",
"type": 0,
"value": "not secret"
}
],
"folderId": "3b12a9da-7c49-40b8-ad33-aede017a7ead",
"id": "90992f63-ddb6-4e76-8bfc-aede016ca5eb",
"login": {
"password": "passwordA3",
"passwordRevisionDate": "2022-07-26T23:03:23.399Z",
"totp": None,
"username": "userA"
},
"name": "a_test",
"notes": None,
"object": "item",
"organizationId": None,
"passwordHistory": [
{
"lastUsedDate": "2022-07-26T23:03:23.405Z",
"password": "a_new_secret: this is secret"
},
{
"lastUsedDate": "2022-07-26T23:03:23.399Z",
"password": "passwordA2"
},
{
"lastUsedDate": "2022-07-26T22:59:52.885Z",
"password": "passwordA"
}
],
"reprompt": 0,
"revisionDate": "2022-07-26T23:03:23.743Z",
"type": 1
},
{
"collectionIds": [],
"deletedDate": None,
"favorite": False,
"folderId": None,
"id": "5ebd4d31-104c-49fc-a09c-aedf003d28ad",
"login": {
"password": "b",
"passwordRevisionDate": None,
"totp": None,
"username": "a"
},
"name": "dupe_name",
"notes": None,
"object": "item",
"organizationId": None,
"reprompt": 0,
"revisionDate": "2022-07-27T03:42:40.353Z",
"type": 1
},
{
"collectionIds": [],
"deletedDate": None,
"favorite": False,
"folderId": None,
"id": "90657653-6695-496d-9431-aedf003d3015",
"login": {
"password": "d",
"passwordRevisionDate": None,
"totp": None,
"username": "c"
},
"name": "dupe_name",
"notes": None,
"object": "item",
"organizationId": None,
"reprompt": 0,
"revisionDate": "2022-07-27T03:42:46.673Z",
"type": 1
}
]
class MockBitwarden(Bitwarden):
logged_in = True
def _get_matches(self, search_value, search_field="name"):
return list(filter(lambda record: record[search_field] == search_value, MOCK_RECORDS))
class LoggedOutMockBitwarden(MockBitwarden):
logged_in = False
class TestLookupModule(unittest.TestCase):
def setUp(self):
self.lookup = lookup_loader.get('community.general.bitwarden')
@patch('ansible_collections.community.general.plugins.lookup.bitwarden._bitwarden', new=MockBitwarden())
def test_bitwarden_plugin_no_match(self):
# Entry 0, "a_test" of the test input should have no duplicates.
self.assertEqual([], self.lookup.run(['not_here'], field='password')[0])
@patch('ansible_collections.community.general.plugins.lookup.bitwarden._bitwarden', new=MockBitwarden())
def test_bitwarden_plugin_fields(self):
# Entry 0, "a_test" of the test input should have no duplicates.
record = MOCK_RECORDS[0]
record_name = record['name']
for k, v in six.iteritems(record['login']):
self.assertEqual([v],
self.lookup.run([record_name], field=k)[0])
@patch('ansible_collections.community.general.plugins.lookup.bitwarden._bitwarden', new=MockBitwarden())
def test_bitwarden_plugin_duplicates(self):
# There are two records with name dupe_name; we need to be order-insensitive with
# checking what was retrieved.
self.assertEqual(set(['b', 'd']),
set(self.lookup.run(['dupe_name'], field='password')[0]))
@patch('ansible_collections.community.general.plugins.lookup.bitwarden._bitwarden', new=MockBitwarden())
def test_bitwarden_plugin_full_item(self):
# Try to retrieve the full record of the first entry where the name is "a_name".
self.assertEqual([MOCK_RECORDS[0]],
self.lookup.run(['a_test'])[0])
@patch('ansible_collections.community.general.plugins.lookup.bitwarden._bitwarden', LoggedOutMockBitwarden())
def test_bitwarden_plugin_logged_out(self):
record = MOCK_RECORDS[0]
record_name = record['name']
with self.assertRaises(AnsibleError):
self.lookup.run([record_name], field='password')

45
collections/ansible_collections/community/general/tests/unit/plugins/lookup/test_dependent.py Normal file
View File

@@ -0,0 +1,45 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2020-2021, Felix Fontein <felix@fontein.de>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
# Make coding more python3-ish
from __future__ import absolute_import, division, print_function
__metaclass__ = type
from ansible_collections.community.internal_test_tools.tests.unit.compat.unittest import TestCase
from ansible_collections.community.internal_test_tools.tests.unit.compat.mock import (
MagicMock,
)
from ansible.plugins.loader import lookup_loader
class TestLookupModule(TestCase):
def setUp(self):
templar = MagicMock()
templar._loader = None
self.lookup = lookup_loader.get("community.general.dependent", templar=templar)
def test_empty(self):
self.assertListEqual(self.lookup.run([], None), [])
def test_simple(self):
self.assertListEqual(
self.lookup.run(
[
{'a': '[1, 2]'},
{'b': '[item.a + 3, item.a + 6]'},
{'c': '[item.a + item.b * 10]'},
],
{},
),
[
{'a': 1, 'b': 4, 'c': 41},
{'a': 1, 'b': 7, 'c': 71},
{'a': 2, 'b': 5, 'c': 52},
{'a': 2, 'b': 8, 'c': 82},
],
)

44
collections/ansible_collections/community/general/tests/unit/plugins/lookup/test_dsv.py Normal file
View File

@@ -0,0 +1,44 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2020, Adam Migus <adam@migus.org>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
# Make coding more python3-ish
from __future__ import absolute_import, division, print_function
__metaclass__ = type
from ansible_collections.community.general.tests.unit.compat.unittest import TestCase
from ansible_collections.community.general.tests.unit.compat.mock import (
patch,
MagicMock,
)
from ansible_collections.community.general.plugins.lookup import dsv
from ansible.plugins.loader import lookup_loader
class MockSecretsVault(MagicMock):
RESPONSE = '{"foo": "bar"}'
def get_secret_json(self, path):
return self.RESPONSE
class TestLookupModule(TestCase):
def setUp(self):
dsv.sdk_is_missing = False
self.lookup = lookup_loader.get("community.general.dsv")
@patch(
"ansible_collections.community.general.plugins.lookup.dsv.LookupModule.Client",
MockSecretsVault(),
)
def test_get_secret_json(self):
self.assertListEqual(
[MockSecretsVault.RESPONSE],
self.lookup.run(
["/dummy"],
[],
**{"tenant": "dummy", "client_id": "dummy", "client_secret": "dummy", }
),
)

58
collections/ansible_collections/community/general/tests/unit/plugins/lookup/test_etcd3.py Normal file
View File

@@ -0,0 +1,58 @@
# -*- coding: utf-8 -*-
#
# Copyright (c) 2020, SCC France, Eric Belhomme <ebelhomme@fr.scc.com>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import pytest
from ansible_collections.community.general.tests.unit.compat import unittest
from ansible_collections.community.general.tests.unit.compat.mock import patch, MagicMock
from ansible.errors import AnsibleError
from ansible_collections.community.general.plugins.lookup import etcd3
from ansible.plugins.loader import lookup_loader
class FakeKVMetadata:
def __init__(self, keyvalue, header):
self.key = keyvalue
self.create_revision = ''
self.mod_revision = ''
self.version = ''
self.lease_id = ''
self.response_header = header
class FakeEtcd3Client(MagicMock):
def get_prefix(self, key):
for i in range(1, 4):
yield self.get('{0}_{1}'.format(key, i))
def get(self, key):
return ("{0} value".format(key), FakeKVMetadata(key, None))
class TestLookupModule(unittest.TestCase):
def setUp(self):
etcd3.HAS_ETCD = True
self.lookup = lookup_loader.get('community.general.etcd3')
@patch('ansible_collections.community.general.plugins.lookup.etcd3.etcd3_client', FakeEtcd3Client())
def test_key(self):
expected_result = [{'key': 'a_key', 'value': 'a_key value'}]
self.assertListEqual(expected_result, self.lookup.run(['a_key'], []))
@patch('ansible_collections.community.general.plugins.lookup.etcd3.etcd3_client', FakeEtcd3Client())
def test_key_prefix(self):
expected_result = [
{'key': 'a_key_1', 'value': 'a_key_1 value'},
{'key': 'a_key_2', 'value': 'a_key_2 value'},
{'key': 'a_key_3', 'value': 'a_key_3 value'},
]
self.assertListEqual(expected_result, self.lookup.run(['a_key'], [], **{'prefix': True}))

175
collections/ansible_collections/community/general/tests/unit/plugins/lookup/test_lastpass.py Normal file
View File

@@ -0,0 +1,175 @@
# Copyright (c) 2016 Andrew Zenk <azenk@umn.edu>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from argparse import ArgumentParser
from ansible_collections.community.general.tests.unit.compat import unittest
from ansible_collections.community.general.tests.unit.compat.mock import patch
from ansible.errors import AnsibleError
from ansible.module_utils import six
from ansible.plugins.loader import lookup_loader
from ansible_collections.community.general.plugins.lookup.lastpass import LookupModule, LPass, LPassException
MOCK_ENTRIES = [{'username': 'user',
'name': 'Mock Entry',
'password': 't0pS3cret passphrase entry!',
'url': 'https://localhost/login',
'notes': 'Test\nnote with multiple lines.\n',
'id': '0123456789'}]
class MockLPass(LPass):
_mock_logged_out = False
_mock_disconnected = False
def _lookup_mock_entry(self, key):
for entry in MOCK_ENTRIES:
if key == entry['id'] or key == entry['name']:
return entry
def _run(self, args, stdin=None, expected_rc=0):
# Mock behavior of lpass executable
base_options = ArgumentParser(add_help=False)
base_options.add_argument('--color', default="auto", choices=['auto', 'always', 'never'])
p = ArgumentParser()
sp = p.add_subparsers(help='command', dest='subparser_name')
logout_p = sp.add_parser('logout', parents=[base_options], help='logout')
show_p = sp.add_parser('show', parents=[base_options], help='show entry details')
field_group = show_p.add_mutually_exclusive_group(required=True)
for field in MOCK_ENTRIES[0].keys():
field_group.add_argument("--{0}".format(field), default=False, action='store_true')
field_group.add_argument('--field', default=None)
show_p.add_argument('selector', help='Unique Name or ID')
args = p.parse_args(args)
def mock_exit(output='', error='', rc=0):
if rc != expected_rc:
raise LPassException(error)
return output, error
if args.color != 'never':
return mock_exit(error='Error: Mock only supports --color=never', rc=1)
if args.subparser_name == 'logout':
if self._mock_logged_out:
return mock_exit(error='Error: Not currently logged in', rc=1)
logged_in_error = 'Are you sure you would like to log out? [Y/n]'
if stdin and stdin.lower() == 'n\n':
return mock_exit(output='Log out: aborted.', error=logged_in_error, rc=1)
elif stdin and stdin.lower() == 'y\n':
return mock_exit(output='Log out: complete.', error=logged_in_error, rc=0)
else:
return mock_exit(error='Error: aborted response', rc=1)
if args.subparser_name == 'show':
if self._mock_logged_out:
return mock_exit(error='Error: Could not find decryption key.' +
' Perhaps you need to login with `lpass login`.', rc=1)
if self._mock_disconnected:
return mock_exit(error='Error: Couldn\'t resolve host name.', rc=1)
mock_entry = self._lookup_mock_entry(args.selector)
if args.field:
return mock_exit(output=mock_entry.get(args.field, ''))
elif args.password:
return mock_exit(output=mock_entry.get('password', ''))
elif args.username:
return mock_exit(output=mock_entry.get('username', ''))
elif args.url:
return mock_exit(output=mock_entry.get('url', ''))
elif args.name:
return mock_exit(output=mock_entry.get('name', ''))
elif args.id:
return mock_exit(output=mock_entry.get('id', ''))
elif args.notes:
return mock_exit(output=mock_entry.get('notes', ''))
raise LPassException('We should never get here')
class DisconnectedMockLPass(MockLPass):
_mock_disconnected = True
class LoggedOutMockLPass(MockLPass):
_mock_logged_out = True
class TestLPass(unittest.TestCase):
def setUp(self):
self.lookup = lookup_loader.get('community.general.lastpass')
def test_lastpass_cli_path(self):
lp = MockLPass(path='/dev/null')
self.assertEqual('/dev/null', lp.cli_path)
def test_lastpass_build_args_logout(self):
lp = MockLPass()
self.assertEqual(['logout', '--color=never'], lp._build_args("logout"))
def test_lastpass_logged_in_true(self):
lp = MockLPass()
self.assertTrue(lp.logged_in)
def test_lastpass_logged_in_false(self):
lp = LoggedOutMockLPass()
self.assertFalse(lp.logged_in)
def test_lastpass_show_disconnected(self):
lp = DisconnectedMockLPass()
with self.assertRaises(LPassException):
lp.get_field('0123456789', 'username')
def test_lastpass_show(self):
lp = MockLPass()
for entry in MOCK_ENTRIES:
entry_id = entry.get('id')
for k, v in six.iteritems(entry):
self.assertEqual(v.strip(), lp.get_field(entry_id, k))
class TestLastpassPlugin(unittest.TestCase):
def setUp(self):
self.lookup = lookup_loader.get('community.general.lastpass')
@patch('ansible_collections.community.general.plugins.lookup.lastpass.LPass', new=MockLPass)
def test_lastpass_plugin_normal(self):
for entry in MOCK_ENTRIES:
entry_id = entry.get('id')
for k, v in six.iteritems(entry):
self.assertEqual(v.strip(),
self.lookup.run([entry_id], field=k)[0])
@patch('ansible_collections.community.general.plugins.lookup.lastpass.LPass', LoggedOutMockLPass)
def test_lastpass_plugin_logged_out(self):
entry = MOCK_ENTRIES[0]
entry_id = entry.get('id')
with self.assertRaises(AnsibleError):
self.lookup.run([entry_id], field='password')
@patch('ansible_collections.community.general.plugins.lookup.lastpass.LPass', DisconnectedMockLPass)
def test_lastpass_plugin_disconnected(self):
entry = MOCK_ENTRIES[0]
entry_id = entry.get('id')
with self.assertRaises(AnsibleError):
self.lookup.run([entry_id], field='password')

537
collections/ansible_collections/community/general/tests/unit/plugins/lookup/test_manifold.py Normal file
View File

@@ -0,0 +1,537 @@
# Copyright (c) 2018, Arigato Machine Inc.
# Copyright (c) 2018, Ansible Project
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from ansible_collections.community.general.tests.unit.compat import unittest
from ansible_collections.community.general.tests.unit.compat.mock import patch, call
from ansible.errors import AnsibleError
from ansible.module_utils.urls import ConnectionError, SSLValidationError
from ansible.module_utils.six.moves.urllib.error import HTTPError, URLError
from ansible.module_utils import six
from ansible.plugins.loader import lookup_loader
from ansible_collections.community.general.plugins.lookup.manifold import ManifoldApiClient, LookupModule, ApiError
import json
import os
API_FIXTURES = {
'https://api.marketplace.manifold.co/v1/resources':
[
{
"body": {
"label": "resource-1",
"name": "Resource 1"
},
"id": "rid-1"
},
{
"body": {
"label": "resource-2",
"name": "Resource 2"
},
"id": "rid-2"
}
],
'https://api.marketplace.manifold.co/v1/resources?label=resource-1':
[
{
"body": {
"label": "resource-1",
"name": "Resource 1"
},
"id": "rid-1"
}
],
'https://api.marketplace.manifold.co/v1/resources?label=resource-2':
[
{
"body": {
"label": "resource-2",
"name": "Resource 2"
},
"id": "rid-2"
}
],
'https://api.marketplace.manifold.co/v1/resources?team_id=tid-1':
[
{
"body": {
"label": "resource-1",
"name": "Resource 1"
},
"id": "rid-1"
}
],
'https://api.marketplace.manifold.co/v1/resources?project_id=pid-1':
[
{
"body": {
"label": "resource-2",
"name": "Resource 2"
},
"id": "rid-2"
}
],
'https://api.marketplace.manifold.co/v1/resources?project_id=pid-2':
[
{
"body": {
"label": "resource-1",
"name": "Resource 1"
},
"id": "rid-1"
},
{
"body": {
"label": "resource-3",
"name": "Resource 3"
},
"id": "rid-3"
}
],
'https://api.marketplace.manifold.co/v1/resources?team_id=tid-1&project_id=pid-1':
[
{
"body": {
"label": "resource-1",
"name": "Resource 1"
},
"id": "rid-1"
}
],
'https://api.marketplace.manifold.co/v1/projects':
[
{
"body": {
"label": "project-1",
"name": "Project 1",
},
"id": "pid-1",
},
{
"body": {
"label": "project-2",
"name": "Project 2",
},
"id": "pid-2",
}
],
'https://api.marketplace.manifold.co/v1/projects?label=project-2':
[
{
"body": {
"label": "project-2",
"name": "Project 2",
},
"id": "pid-2",
}
],
'https://api.marketplace.manifold.co/v1/credentials?resource_id=rid-1':
[
{
"body": {
"resource_id": "rid-1",
"values": {
"RESOURCE_TOKEN_1": "token-1",
"RESOURCE_TOKEN_2": "token-2"
}
},
"id": "cid-1",
}
],
'https://api.marketplace.manifold.co/v1/credentials?resource_id=rid-2':
[
{
"body": {
"resource_id": "rid-2",
"values": {
"RESOURCE_TOKEN_3": "token-3",
"RESOURCE_TOKEN_4": "token-4"
}
},
"id": "cid-2",
}
],
'https://api.marketplace.manifold.co/v1/credentials?resource_id=rid-3':
[
{
"body": {
"resource_id": "rid-3",
"values": {
"RESOURCE_TOKEN_1": "token-5",
"RESOURCE_TOKEN_2": "token-6"
}
},
"id": "cid-3",
}
],
'https://api.identity.manifold.co/v1/teams':
[
{
"id": "tid-1",
"body": {
"name": "Team 1",
"label": "team-1"
}
},
{
"id": "tid-2",
"body": {
"name": "Team 2",
"label": "team-2"
}
}
]
}
def mock_fixture(open_url_mock, fixture=None, data=None, headers=None):
if not headers:
headers = {}
if fixture:
data = json.dumps(API_FIXTURES[fixture])
if 'content-type' not in headers:
headers['content-type'] = 'application/json'
open_url_mock.return_value.read.return_value = data
open_url_mock.return_value.headers = headers
class TestManifoldApiClient(unittest.TestCase):
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_request_sends_default_headers(self, open_url_mock):
mock_fixture(open_url_mock, data='hello')
client = ManifoldApiClient('token-123')
client.request('test', 'endpoint')
open_url_mock.assert_called_with('https://api.test.manifold.co/v1/endpoint',
headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'},
http_agent='python-manifold-ansible-1.0.0')
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_request_decodes_json(self, open_url_mock):
mock_fixture(open_url_mock, fixture='https://api.marketplace.manifold.co/v1/resources')
client = ManifoldApiClient('token-123')
self.assertIsInstance(client.request('marketplace', 'resources'), list)
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_request_streams_text(self, open_url_mock):
mock_fixture(open_url_mock, data='hello', headers={'content-type': "text/plain"})
client = ManifoldApiClient('token-123')
self.assertEqual('hello', client.request('test', 'endpoint'))
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_request_processes_parameterized_headers(self, open_url_mock):
mock_fixture(open_url_mock, data='hello')
client = ManifoldApiClient('token-123')
client.request('test', 'endpoint', headers={'X-HEADER': 'MANIFOLD'})
open_url_mock.assert_called_with('https://api.test.manifold.co/v1/endpoint',
headers={'Accept': '*/*', 'Authorization': 'Bearer token-123',
'X-HEADER': 'MANIFOLD'},
http_agent='python-manifold-ansible-1.0.0')
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_request_passes_arbitrary_parameters(self, open_url_mock):
mock_fixture(open_url_mock, data='hello')
client = ManifoldApiClient('token-123')
client.request('test', 'endpoint', use_proxy=False, timeout=5)
open_url_mock.assert_called_with('https://api.test.manifold.co/v1/endpoint',
headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'},
http_agent='python-manifold-ansible-1.0.0',
use_proxy=False, timeout=5)
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_request_raises_on_incorrect_json(self, open_url_mock):
mock_fixture(open_url_mock, data='noJson', headers={'content-type': "application/json"})
client = ManifoldApiClient('token-123')
with self.assertRaises(ApiError) as context:
client.request('test', 'endpoint')
self.assertEqual('JSON response can\'t be parsed while requesting https://api.test.manifold.co/v1/endpoint:\n'
'noJson',
str(context.exception))
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_request_raises_on_status_500(self, open_url_mock):
open_url_mock.side_effect = HTTPError('https://api.test.manifold.co/v1/endpoint',
500, 'Server error', {}, six.StringIO('ERROR'))
client = ManifoldApiClient('token-123')
with self.assertRaises(ApiError) as context:
client.request('test', 'endpoint')
self.assertEqual('Server returned: HTTP Error 500: Server error while requesting '
'https://api.test.manifold.co/v1/endpoint:\nERROR',
str(context.exception))
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_request_raises_on_bad_url(self, open_url_mock):
open_url_mock.side_effect = URLError('URL is invalid')
client = ManifoldApiClient('token-123')
with self.assertRaises(ApiError) as context:
client.request('test', 'endpoint')
self.assertEqual('Failed lookup url for https://api.test.manifold.co/v1/endpoint : <url'
'open error URL is invalid>',
str(context.exception))
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_request_raises_on_ssl_error(self, open_url_mock):
open_url_mock.side_effect = SSLValidationError('SSL Error')
client = ManifoldApiClient('token-123')
with self.assertRaises(ApiError) as context:
client.request('test', 'endpoint')
self.assertEqual('Error validating the server\'s certificate for https://api.test.manifold.co/v1/endpoint: '
'SSL Error',
str(context.exception))
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_request_raises_on_connection_error(self, open_url_mock):
open_url_mock.side_effect = ConnectionError('Unknown connection error')
client = ManifoldApiClient('token-123')
with self.assertRaises(ApiError) as context:
client.request('test', 'endpoint')
self.assertEqual('Error connecting to https://api.test.manifold.co/v1/endpoint: Unknown connection error',
str(context.exception))
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_get_resources_get_all(self, open_url_mock):
url = 'https://api.marketplace.manifold.co/v1/resources'
mock_fixture(open_url_mock, fixture=url)
client = ManifoldApiClient('token-123')
self.assertListEqual(API_FIXTURES[url], client.get_resources())
open_url_mock.assert_called_with(url,
headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'},
http_agent='python-manifold-ansible-1.0.0')
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_get_resources_filter_label(self, open_url_mock):
url = 'https://api.marketplace.manifold.co/v1/resources?label=resource-1'
mock_fixture(open_url_mock, fixture=url)
client = ManifoldApiClient('token-123')
self.assertListEqual(API_FIXTURES[url], client.get_resources(label='resource-1'))
open_url_mock.assert_called_with(url,
headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'},
http_agent='python-manifold-ansible-1.0.0')
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_get_resources_filter_team_and_project(self, open_url_mock):
url = 'https://api.marketplace.manifold.co/v1/resources?team_id=tid-1&project_id=pid-1'
mock_fixture(open_url_mock, fixture=url)
client = ManifoldApiClient('token-123')
self.assertListEqual(API_FIXTURES[url], client.get_resources(team_id='tid-1', project_id='pid-1'))
args, kwargs = open_url_mock.call_args
url_called = args[0]
# Dict order is not guaranteed, so an url may have querystring parameters order randomized
self.assertIn('team_id=tid-1', url_called)
self.assertIn('project_id=pid-1', url_called)
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_get_teams_get_all(self, open_url_mock):
url = 'https://api.identity.manifold.co/v1/teams'
mock_fixture(open_url_mock, fixture=url)
client = ManifoldApiClient('token-123')
self.assertListEqual(API_FIXTURES[url], client.get_teams())
open_url_mock.assert_called_with(url,
headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'},
http_agent='python-manifold-ansible-1.0.0')
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_get_teams_filter_label(self, open_url_mock):
url = 'https://api.identity.manifold.co/v1/teams'
mock_fixture(open_url_mock, fixture=url)
client = ManifoldApiClient('token-123')
self.assertListEqual(API_FIXTURES[url][1:2], client.get_teams(label='team-2'))
open_url_mock.assert_called_with(url,
headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'},
http_agent='python-manifold-ansible-1.0.0')
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_get_projects_get_all(self, open_url_mock):
url = 'https://api.marketplace.manifold.co/v1/projects'
mock_fixture(open_url_mock, fixture=url)
client = ManifoldApiClient('token-123')
self.assertListEqual(API_FIXTURES[url], client.get_projects())
open_url_mock.assert_called_with(url,
headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'},
http_agent='python-manifold-ansible-1.0.0')
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_get_projects_filter_label(self, open_url_mock):
url = 'https://api.marketplace.manifold.co/v1/projects?label=project-2'
mock_fixture(open_url_mock, fixture=url)
client = ManifoldApiClient('token-123')
self.assertListEqual(API_FIXTURES[url], client.get_projects(label='project-2'))
open_url_mock.assert_called_with(url,
headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'},
http_agent='python-manifold-ansible-1.0.0')
@patch('ansible_collections.community.general.plugins.lookup.manifold.open_url')
def test_get_credentials(self, open_url_mock):
url = 'https://api.marketplace.manifold.co/v1/credentials?resource_id=rid-1'
mock_fixture(open_url_mock, fixture=url)
client = ManifoldApiClient('token-123')
self.assertListEqual(API_FIXTURES[url], client.get_credentials(resource_id='rid-1'))
open_url_mock.assert_called_with(url,
headers={'Accept': '*/*', 'Authorization': 'Bearer token-123'},
http_agent='python-manifold-ansible-1.0.0')
class TestLookupModule(unittest.TestCase):
def setUp(self):
self.lookup = lookup_loader.get('community.general.manifold')
@patch('ansible_collections.community.general.plugins.lookup.manifold.ManifoldApiClient')
def test_get_all(self, client_mock):
expected_result = [{'RESOURCE_TOKEN_1': 'token-1',
'RESOURCE_TOKEN_2': 'token-2',
'RESOURCE_TOKEN_3': 'token-3',
'RESOURCE_TOKEN_4': 'token-4'
}]
client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources']
client_mock.return_value.get_credentials.side_effect = lambda x: API_FIXTURES['https://api.marketplace.manifold.co/v1/'
'credentials?resource_id={0}'.format(x)]
self.assertListEqual(expected_result, self.lookup.run([], api_token='token-123'))
client_mock.assert_called_with('token-123')
client_mock.return_value.get_resources.assert_called_with(team_id=None, project_id=None)
@patch('ansible_collections.community.general.plugins.lookup.manifold.ManifoldApiClient')
def test_get_one_resource(self, client_mock):
expected_result = [{'RESOURCE_TOKEN_3': 'token-3',
'RESOURCE_TOKEN_4': 'token-4'
}]
client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources?label=resource-2']
client_mock.return_value.get_credentials.side_effect = lambda x: API_FIXTURES['https://api.marketplace.manifold.co/v1/'
'credentials?resource_id={0}'.format(x)]
self.assertListEqual(expected_result, self.lookup.run(['resource-2'], api_token='token-123'))
client_mock.return_value.get_resources.assert_called_with(team_id=None, project_id=None, label='resource-2')
@patch('ansible_collections.community.general.plugins.lookup.manifold.ManifoldApiClient')
def test_get_two_resources(self, client_mock):
expected_result = [{'RESOURCE_TOKEN_1': 'token-1',
'RESOURCE_TOKEN_2': 'token-2',
'RESOURCE_TOKEN_3': 'token-3',
'RESOURCE_TOKEN_4': 'token-4'
}]
client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources']
client_mock.return_value.get_credentials.side_effect = lambda x: API_FIXTURES['https://api.marketplace.manifold.co/v1/'
'credentials?resource_id={0}'.format(x)]
self.assertListEqual(expected_result, self.lookup.run(['resource-1', 'resource-2'], api_token='token-123'))
client_mock.assert_called_with('token-123')
client_mock.return_value.get_resources.assert_called_with(team_id=None, project_id=None)
@patch('ansible_collections.community.general.plugins.lookup.manifold.display')
@patch('ansible_collections.community.general.plugins.lookup.manifold.ManifoldApiClient')
def test_get_resources_with_same_credential_names(self, client_mock, display_mock):
expected_result = [{'RESOURCE_TOKEN_1': 'token-5',
'RESOURCE_TOKEN_2': 'token-6'
}]
client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources?project_id=pid-2']
client_mock.return_value.get_projects.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/projects?label=project-2']
client_mock.return_value.get_credentials.side_effect = lambda x: API_FIXTURES['https://api.marketplace.manifold.co/v1/'
'credentials?resource_id={0}'.format(x)]
self.assertListEqual(expected_result, self.lookup.run([], api_token='token-123', project='project-2'))
client_mock.assert_called_with('token-123')
display_mock.warning.assert_has_calls([
call("'RESOURCE_TOKEN_1' with label 'resource-1' was replaced by resource data with label 'resource-3'"),
call("'RESOURCE_TOKEN_2' with label 'resource-1' was replaced by resource data with label 'resource-3'")],
any_order=True
)
client_mock.return_value.get_resources.assert_called_with(team_id=None, project_id='pid-2')
@patch('ansible_collections.community.general.plugins.lookup.manifold.ManifoldApiClient')
def test_filter_by_team(self, client_mock):
expected_result = [{'RESOURCE_TOKEN_1': 'token-1',
'RESOURCE_TOKEN_2': 'token-2'
}]
client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources?team_id=tid-1']
client_mock.return_value.get_teams.return_value = API_FIXTURES['https://api.identity.manifold.co/v1/teams'][0:1]
client_mock.return_value.get_credentials.side_effect = lambda x: API_FIXTURES['https://api.marketplace.manifold.co/v1/'
'credentials?resource_id={0}'.format(x)]
self.assertListEqual(expected_result, self.lookup.run([], api_token='token-123', team='team-1'))
client_mock.assert_called_with('token-123')
client_mock.return_value.get_resources.assert_called_with(team_id='tid-1', project_id=None)
@patch('ansible_collections.community.general.plugins.lookup.manifold.ManifoldApiClient')
def test_filter_by_project(self, client_mock):
expected_result = [{'RESOURCE_TOKEN_3': 'token-3',
'RESOURCE_TOKEN_4': 'token-4'
}]
client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources?project_id=pid-1']
client_mock.return_value.get_projects.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/projects'][0:1]
client_mock.return_value.get_credentials.side_effect = lambda x: API_FIXTURES['https://api.marketplace.manifold.co/v1/'
'credentials?resource_id={0}'.format(x)]
self.assertListEqual(expected_result, self.lookup.run([], api_token='token-123', project='project-1'))
client_mock.assert_called_with('token-123')
client_mock.return_value.get_resources.assert_called_with(team_id=None, project_id='pid-1')
@patch('ansible_collections.community.general.plugins.lookup.manifold.ManifoldApiClient')
def test_filter_by_team_and_project(self, client_mock):
expected_result = [{'RESOURCE_TOKEN_1': 'token-1',
'RESOURCE_TOKEN_2': 'token-2'
}]
client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources?team_id=tid-1&project_id=pid-1']
client_mock.return_value.get_teams.return_value = API_FIXTURES['https://api.identity.manifold.co/v1/teams'][0:1]
client_mock.return_value.get_projects.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/projects'][0:1]
client_mock.return_value.get_credentials.side_effect = lambda x: API_FIXTURES['https://api.marketplace.manifold.co/v1/'
'credentials?resource_id={0}'.format(x)]
self.assertListEqual(expected_result, self.lookup.run([], api_token='token-123', project='project-1'))
client_mock.assert_called_with('token-123')
client_mock.return_value.get_resources.assert_called_with(team_id=None, project_id='pid-1')
@patch('ansible_collections.community.general.plugins.lookup.manifold.ManifoldApiClient')
def test_raise_team_doesnt_exist(self, client_mock):
client_mock.return_value.get_teams.return_value = []
with self.assertRaises(AnsibleError) as context:
self.lookup.run([], api_token='token-123', team='no-team')
self.assertEqual("Team 'no-team' does not exist",
str(context.exception))
@patch('ansible_collections.community.general.plugins.lookup.manifold.ManifoldApiClient')
def test_raise_project_doesnt_exist(self, client_mock):
client_mock.return_value.get_projects.return_value = []
with self.assertRaises(AnsibleError) as context:
self.lookup.run([], api_token='token-123', project='no-project')
self.assertEqual("Project 'no-project' does not exist",
str(context.exception))
@patch('ansible_collections.community.general.plugins.lookup.manifold.ManifoldApiClient')
def test_raise_resource_doesnt_exist(self, client_mock):
client_mock.return_value.get_resources.return_value = API_FIXTURES['https://api.marketplace.manifold.co/v1/resources']
with self.assertRaises(AnsibleError) as context:
self.lookup.run(['resource-1', 'no-resource-1', 'no-resource-2'], api_token='token-123')
self.assertEqual("Resource(s) no-resource-1, no-resource-2 do not exist",
str(context.exception))
@patch('ansible_collections.community.general.plugins.lookup.manifold.ManifoldApiClient')
def test_catch_api_error(self, client_mock):
client_mock.side_effect = ApiError('Generic error')
with self.assertRaises(AnsibleError) as context:
self.lookup.run([], api_token='token-123')
self.assertEqual("API Error: Generic error",
str(context.exception))
@patch('ansible_collections.community.general.plugins.lookup.manifold.ManifoldApiClient')
def test_catch_unhandled_exception(self, client_mock):
client_mock.side_effect = Exception('Unknown error')
with self.assertRaises(AnsibleError) as context:
self.lookup.run([], api_token='token-123')
self.assertTrue('Exception: Unknown error' in str(context.exception))
@patch('ansible_collections.community.general.plugins.lookup.manifold.ManifoldApiClient')
def test_falls_back_to_env_var(self, client_mock):
client_mock.return_value.get_resources.return_value = []
client_mock.return_value.get_credentials.return_value = []
try:
os.environ['MANIFOLD_API_TOKEN'] = 'token-321'
self.lookup.run([])
finally:
os.environ.pop('MANIFOLD_API_TOKEN', None)
client_mock.assert_called_with('token-321')
@patch('ansible_collections.community.general.plugins.lookup.manifold.ManifoldApiClient')
def test_falls_raises_on_no_token(self, client_mock):
client_mock.return_value.get_resources.return_value = []
client_mock.return_value.get_credentials.return_value = []
os.environ.pop('MANIFOLD_API_TOKEN', None)
with self.assertRaises(AnsibleError) as context:
self.lookup.run([])
assert 'api_token' in str(context.exception)

44
collections/ansible_collections/community/general/tests/unit/plugins/lookup/test_revbitspss.py Normal file
View File

@@ -0,0 +1,44 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2021, RevBits <info@revbits.com>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import absolute_import, division, print_function
__metaclass__ = type
from ansible_collections.community.general.tests.unit.compat.unittest import TestCase
from ansible_collections.community.general.tests.unit.compat.mock import (
patch,
MagicMock,
)
from ansible_collections.community.general.plugins.lookup import revbitspss
from ansible.plugins.loader import lookup_loader
class MockPamSecrets(MagicMock):
RESPONSE = 'dummy value'
def get_pam_secret(self, path):
return self.RESPONSE
class TestLookupModule(TestCase):
def setUp(self):
revbitspss.ANOTHER_LIBRARY_IMPORT_ERROR = None
self.lookup = lookup_loader.get("community.general.revbitspss")
@patch(
"ansible_collections.community.general.plugins.lookup.revbitspss.LookupModule.Client",
MockPamSecrets(),
)
def test_get_pam_secret(self):
terms = ['dummy secret']
variables = []
kwargs = {
"base_url": 'https://dummy.url',
"api_key": 'dummy'
}
self.assertListEqual(
[{'dummy secret': 'dummy value'}],
self.lookup.run(terms, variables, **kwargs)
)

120
collections/ansible_collections/community/general/tests/unit/plugins/lookup/test_tss.py Normal file
View File

@@ -0,0 +1,120 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2020, Adam Migus <adam@migus.org>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
# Make coding more python3-ish
from __future__ import absolute_import, division, print_function
__metaclass__ = type
from ansible_collections.community.general.tests.unit.compat.unittest import TestCase
from ansible_collections.community.general.tests.unit.compat.mock import (
patch,
DEFAULT,
MagicMock,
)
from ansible_collections.community.general.plugins.lookup import tss
from ansible.plugins.loader import lookup_loader
TSS_IMPORT_PATH = 'ansible_collections.community.general.plugins.lookup.tss'
def make_absolute(name):
return '.'.join([TSS_IMPORT_PATH, name])
class SecretServerError(Exception):
def __init__(self):
self.message = ''
class MockSecretServer(MagicMock):
RESPONSE = '{"foo": "bar"}'
def get_secret_json(self, path):
return self.RESPONSE
class MockFaultySecretServer(MagicMock):
def get_secret_json(self, path):
raise SecretServerError
@patch(make_absolute('SecretServer'), MockSecretServer())
class TestTSSClient(TestCase):
def setUp(self):
self.server_params = {
'base_url': '',
'username': '',
'domain': '',
'password': '',
'api_path_uri': '',
'token_path_uri': '',
}
def test_from_params(self):
with patch(make_absolute('HAS_TSS_AUTHORIZER'), False):
self.assert_client_version('v0')
with patch.dict(self.server_params, {'domain': 'foo'}):
with self.assertRaises(tss.AnsibleError):
self._get_client()
with patch.multiple(TSS_IMPORT_PATH,
HAS_TSS_AUTHORIZER=True,
PasswordGrantAuthorizer=DEFAULT,
DomainPasswordGrantAuthorizer=DEFAULT):
self.assert_client_version('v1')
with patch.dict(self.server_params, {'domain': 'foo'}):
self.assert_client_version('v1')
def assert_client_version(self, version):
version_to_class = {
'v0': tss.TSSClientV0,
'v1': tss.TSSClientV1
}
client = self._get_client()
self.assertIsInstance(client, version_to_class[version])
def _get_client(self):
return tss.TSSClient.from_params(**self.server_params)
class TestLookupModule(TestCase):
VALID_TERMS = [1]
INVALID_TERMS = ['foo']
def setUp(self):
self.lookup = lookup_loader.get("community.general.tss")
@patch.multiple(TSS_IMPORT_PATH,
HAS_TSS_SDK=False,
SecretServer=MockSecretServer)
def test_missing_sdk(self):
with self.assertRaises(tss.AnsibleError):
self._run_lookup(self.VALID_TERMS)
@patch.multiple(TSS_IMPORT_PATH,
HAS_TSS_SDK=True,
SecretServerError=SecretServerError)
def test_get_secret_json(self):
with patch(make_absolute('SecretServer'), MockSecretServer):
self.assertListEqual([MockSecretServer.RESPONSE], self._run_lookup(self.VALID_TERMS))
with self.assertRaises(tss.AnsibleOptionsError):
self._run_lookup(self.INVALID_TERMS)
with patch(make_absolute('SecretServer'), MockFaultySecretServer):
with self.assertRaises(tss.AnsibleError):
self._run_lookup(self.VALID_TERMS)
def _run_lookup(self, terms, variables=None, **kwargs):
variables = variables or []
kwargs = kwargs or {"base_url": "dummy", "username": "dummy", "password": "dummy"}
return self.lookup.run(terms, variables, **kwargs)