From b4343baa16f2109c2b094d19ac3c1c0e4d307dc4 Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Thu, 24 Mar 2022 20:58:13 -0100 Subject: [PATCH] RDS - simplify filter-parameter logic (#4966) --- moto/rds/responses.py | 10 ++++++---- moto/rds/utils.py | 30 ------------------------------ tests/test_rds/test_utils.py | 36 ------------------------------------ 3 files changed, 6 insertions(+), 70 deletions(-) diff --git a/moto/rds/responses.py b/moto/rds/responses.py index 03fa1cf36..ada2466b8 100644 --- a/moto/rds/responses.py +++ b/moto/rds/responses.py @@ -4,7 +4,6 @@ from moto.core.responses import BaseResponse from moto.ec2.models import ec2_backends from .models import rds_backends from .exceptions import DBParameterGroupNotFoundError -from .utils import filters_from_querystring class RDSResponse(BaseResponse): @@ -171,7 +170,8 @@ class RDSResponse(BaseResponse): def describe_db_instances(self): db_instance_identifier = self._get_param("DBInstanceIdentifier") - filters = filters_from_querystring(self.querystring) + filters = self._get_multi_param("Filters.Filter.") + filters = {f["Name"]: f["Values"] for f in filters} all_instances = list( self.backend.describe_databases(db_instance_identifier, filters=filters) ) @@ -240,7 +240,8 @@ class RDSResponse(BaseResponse): def describe_db_snapshots(self): db_instance_identifier = self._get_param("DBInstanceIdentifier") db_snapshot_identifier = self._get_param("DBSnapshotIdentifier") - filters = filters_from_querystring(self.querystring) + filters = self._get_multi_param("Filters.Filter.") + filters = {f["Name"]: f["Values"] for f in filters} snapshots = self.backend.describe_database_snapshots( db_instance_identifier, db_snapshot_identifier, filters ) @@ -549,7 +550,8 @@ class RDSResponse(BaseResponse): def describe_db_cluster_snapshots(self): db_cluster_identifier = self._get_param("DBClusterIdentifier") db_snapshot_identifier = self._get_param("DBClusterSnapshotIdentifier") - filters = filters_from_querystring(self.querystring) + filters = self._get_multi_param("Filters.Filter.") + filters = {f["Name"]: f["Values"] for f in filters} snapshots = self.backend.describe_db_cluster_snapshots( db_cluster_identifier, db_snapshot_identifier, filters ) diff --git a/moto/rds/utils.py b/moto/rds/utils.py index ddc5784f8..07569c962 100644 --- a/moto/rds/utils.py +++ b/moto/rds/utils.py @@ -1,4 +1,3 @@ -import re from collections import namedtuple from botocore.utils import merge_dicts @@ -18,35 +17,6 @@ FilterDef = namedtuple( ) -def filters_from_querystring(querystring): - """Parses filters out of the query string computed by the - moto.core.responses.BaseResponse class. - - :param dict[str, list[str]] querystring: - The `moto`-processed URL query string dictionary. - :returns: - Dict mapping filter names to filter values. - :rtype: - dict[str, list[str]] - """ - response_values = {} - for key, value in sorted(querystring.items()): - match = re.search(r"Filters.Filter.(\d).Name", key) - if match: - filter_index = match.groups()[0] - value_prefix = "Filters.Filter.{0}.Value".format(filter_index) - filter_values = [ - filter_value[0] - for filter_key, filter_value in querystring.items() - if filter_key.startswith(value_prefix) - ] - # The AWS query protocol serializes empty lists as an empty string. - if filter_values == [""]: - filter_values = [] - response_values[value[0]] = filter_values - return response_values - - def get_object_value(obj, attr): """Retrieves an arbitrary attribute value from an object. diff --git a/tests/test_rds/test_utils.py b/tests/test_rds/test_utils.py index 15a3968b5..5dede5e34 100644 --- a/tests/test_rds/test_utils.py +++ b/tests/test_rds/test_utils.py @@ -4,7 +4,6 @@ from moto.rds.utils import ( FilterDef, apply_filter, merge_filters, - filters_from_querystring, validate_filters, ) @@ -135,38 +134,3 @@ class TestMergingFilters(object): assert len(merged.keys()) == 4 for key in merged.keys(): assert merged[key] == ["value1", "value2"] - - -class TestParsingFiltersFromQuerystring(object): - def test_parse_empty_list(self): - # The AWS query protocol serializes empty lists as an empty string. - querystring = { - "Filters.Filter.1.Name": ["empty-filter"], - "Filters.Filter.1.Value.1": [""], - } - filters = filters_from_querystring(querystring) - assert filters == {"empty-filter": []} - - def test_multiple_values(self): - querystring = { - "Filters.Filter.1.Name": ["multi-value"], - "Filters.Filter.1.Value.1": ["value1"], - "Filters.Filter.1.Value.2": ["value2"], - } - filters = filters_from_querystring(querystring) - values = filters["multi-value"] - assert len(values) == 2 - assert "value1" in values - assert "value2" in values - - def test_multiple_filters(self): - querystring = { - "Filters.Filter.1.Name": ["filter-1"], - "Filters.Filter.1.Value.1": ["value1"], - "Filters.Filter.2.Name": ["filter-2"], - "Filters.Filter.2.Value.1": ["value2"], - } - filters = filters_from_querystring(querystring) - assert len(filters.keys()) == 2 - assert filters["filter-1"] == ["value1"] - assert filters["filter-2"] == ["value2"]