mirror of
https://github.com/ansible-collections/community.general.git
synced 2025-04-24 11:21:25 -07:00
New Filter plugin from_csv (#2037)
* Added from_csv filter and integration tests * Cleaning up whitespace * Adding changelog fragment * Updated changelog fragment name * Removed temp fragment * Refactoring csv functions Part 1 * Syncing refactored csv modules/filters * Adding unit tests for csv Module_Util * Updating changelog fragment * Correcting whitespace in unit test * Improving changelog fragment Co-authored-by: Felix Fontein <felix@fontein.de> * Update changelogs/fragments/2037-add-from-csv-filter.yml Co-authored-by: Felix Fontein <felix@fontein.de>
This commit is contained in:
parent
c147d2fb98
commit
6529390901
8 changed files with 383 additions and 47 deletions
164
tests/unit/plugins/module_utils/test_csv.py
Normal file
164
tests/unit/plugins/module_utils/test_csv.py
Normal file
|
@ -0,0 +1,164 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
|
||||
|
||||
from __future__ import (absolute_import, division, print_function)
|
||||
__metaclass__ = type
|
||||
|
||||
import pytest
|
||||
|
||||
from ansible_collections.community.general.plugins.module_utils import csv
|
||||
|
||||
|
||||
VALID_CSV = [
|
||||
(
|
||||
'excel',
|
||||
{},
|
||||
None,
|
||||
"id,name,role\n1,foo,bar\n2,bar,baz",
|
||||
[
|
||||
{
|
||||
"id": "1",
|
||||
"name": "foo",
|
||||
"role": "bar",
|
||||
},
|
||||
{
|
||||
"id": "2",
|
||||
"name": "bar",
|
||||
"role": "baz",
|
||||
},
|
||||
]
|
||||
),
|
||||
(
|
||||
'excel',
|
||||
{"skipinitialspace": True},
|
||||
None,
|
||||
"id,name,role\n1, foo, bar\n2, bar, baz",
|
||||
[
|
||||
{
|
||||
"id": "1",
|
||||
"name": "foo",
|
||||
"role": "bar",
|
||||
},
|
||||
{
|
||||
"id": "2",
|
||||
"name": "bar",
|
||||
"role": "baz",
|
||||
},
|
||||
]
|
||||
),
|
||||
(
|
||||
'excel',
|
||||
{"delimiter": '|'},
|
||||
None,
|
||||
"id|name|role\n1|foo|bar\n2|bar|baz",
|
||||
[
|
||||
{
|
||||
"id": "1",
|
||||
"name": "foo",
|
||||
"role": "bar",
|
||||
},
|
||||
{
|
||||
"id": "2",
|
||||
"name": "bar",
|
||||
"role": "baz",
|
||||
},
|
||||
]
|
||||
),
|
||||
(
|
||||
'unix',
|
||||
{},
|
||||
None,
|
||||
"id,name,role\n1,foo,bar\n2,bar,baz",
|
||||
[
|
||||
{
|
||||
"id": "1",
|
||||
"name": "foo",
|
||||
"role": "bar",
|
||||
},
|
||||
{
|
||||
"id": "2",
|
||||
"name": "bar",
|
||||
"role": "baz",
|
||||
},
|
||||
]
|
||||
),
|
||||
(
|
||||
'excel',
|
||||
{},
|
||||
['id', 'name', 'role'],
|
||||
"1,foo,bar\n2,bar,baz",
|
||||
[
|
||||
{
|
||||
"id": "1",
|
||||
"name": "foo",
|
||||
"role": "bar",
|
||||
},
|
||||
{
|
||||
"id": "2",
|
||||
"name": "bar",
|
||||
"role": "baz",
|
||||
},
|
||||
]
|
||||
),
|
||||
]
|
||||
|
||||
INVALID_CSV = [
|
||||
(
|
||||
'excel',
|
||||
{'strict': True},
|
||||
None,
|
||||
'id,name,role\n1,"f"oo",bar\n2,bar,baz',
|
||||
),
|
||||
]
|
||||
|
||||
INVALID_DIALECT = [
|
||||
(
|
||||
'invalid',
|
||||
{},
|
||||
None,
|
||||
"id,name,role\n1,foo,bar\n2,bar,baz",
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("dialect,dialect_params,fieldnames,data,expected", VALID_CSV)
|
||||
def test_valid_csv(data, dialect, dialect_params, fieldnames, expected):
|
||||
dialect = csv.initialize_dialect(dialect, **dialect_params)
|
||||
reader = csv.read_csv(data, dialect, fieldnames)
|
||||
result = True
|
||||
|
||||
for idx, row in enumerate(reader):
|
||||
for k, v in row.items():
|
||||
if expected[idx][k] != v:
|
||||
result = False
|
||||
break
|
||||
|
||||
assert result
|
||||
|
||||
|
||||
@pytest.mark.parametrize("dialect,dialect_params,fieldnames,data", INVALID_CSV)
|
||||
def test_invalid_csv(data, dialect, dialect_params, fieldnames):
|
||||
dialect = csv.initialize_dialect(dialect, **dialect_params)
|
||||
reader = csv.read_csv(data, dialect, fieldnames)
|
||||
result = False
|
||||
|
||||
try:
|
||||
for row in reader:
|
||||
continue
|
||||
except csv.CSVError:
|
||||
result = True
|
||||
|
||||
assert result
|
||||
|
||||
|
||||
@pytest.mark.parametrize("dialect,dialect_params,fieldnames,data", INVALID_DIALECT)
|
||||
def test_invalid_dialect(data, dialect, dialect_params, fieldnames):
|
||||
result = False
|
||||
|
||||
try:
|
||||
dialect = csv.initialize_dialect(dialect, **dialect_params)
|
||||
except csv.DialectNotAvailableError:
|
||||
result = True
|
||||
|
||||
assert result
|
Loading…
Add table
Add a link
Reference in a new issue