Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

use botocore regions and refactor sorting

Browse Source
This commit is contained in:
Nick Stocchero 2020-08-12 17:16:47 -06:00
parent 8d5c70a924
commit fd69c93a09
3 changed files with 27 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
moto/config/models.py
View File

@ -69,29 +69,6 @@ RESOURCE_MAP = {
"AWS::IAM::Policy": policy_config_query,
}
CONFIG_REGIONS = [
"af-south-1",
"ap-east-1",
"ap-northeast-1",
"ap-northeast-2",
"ap-south-1",
"ap-southeast-1",
"ap-southeast-2",
"ca-central-1",
"eu-central-1",
"eu-north-1",
"eu-south-1",
"eu-west-1",
"eu-west-2",
"eu-west-3",
"me-south-1",
"sa-east-1",
"us-east-1",
"us-east-2",
"us-west-1",
"us-west-2",
]
def datetime2int(date):
return int(time.mktime(date.timetuple()))

47
moto/iam/config.py
View File

@ -1,5 +1,5 @@
import json
import boto3
from moto.core.exceptions import InvalidNextTokenException
from moto.core.models import ConfigQueryModel
from moto.iam import iam_backends
@ -55,8 +55,6 @@ class RoleConfigQuery(ConfigQueryModel):
# In practice, it looks like AWS will only duplicate these resources if you've "used" any roles in the region, but since
# we can't really tell if this has happened in moto, we'll just bind this to the regions in your aggregator
from moto.config.models import CONFIG_REGIONS
aggregated_regions = []
aggregator_sources = aggregator.get(
"account_aggregation_sources"
@ -64,7 +62,7 @@ class RoleConfigQuery(ConfigQueryModel):
for source in aggregator_sources:
source_dict = source.__dict__
if source_dict["all_aws_regions"]:
aggregated_regions = CONFIG_REGIONS
aggregated_regions = boto3.Session().get_available_regions("config")
break
for region in source_dict["aws_regions"]:
aggregated_regions.append(region)
@ -86,15 +84,10 @@ class RoleConfigQuery(ConfigQueryModel):
# Pagination logic, sort by role id
sorted_roles = sorted(duplicate_role_list, key=lambda role: role["_id"])
# sorted_role_ids matches indicies of sorted_roles
sorted_role_ids = list(map(lambda role: role["_id"], sorted_roles))
else:
# Non-aggregated queries are in the else block, and we can treat these like a normal config resource
# Pagination logic, sort by role id
sorted_roles = sorted(role_list, key=lambda role: role.id)
# sorted_role_ids matches indicies of sorted_roles
sorted_role_ids = list(map(lambda role: role.id, sorted_roles))
new_token = None
@ -102,16 +95,22 @@ class RoleConfigQuery(ConfigQueryModel):
if not next_token:
start = 0
else:
if next_token not in sorted_role_ids:
try:
# Find the index of the next
start = next(
index
for (index, r) in enumerate(sorted_roles)
if next_token == (r["_id"] if aggregator else r.id)
)
except StopIteration:
raise InvalidNextTokenException()
start = sorted_role_ids.index(next_token)
# Get the list of items to collect:
role_list = sorted_roles[start : (start + limit)]
if len(sorted_roles) > (start + limit):
new_token = sorted_role_ids[start + limit]
record = sorted_roles[start + limit]
new_token = record["_id"] if aggregator else record.id
return (
[
@ -213,8 +212,6 @@ class PolicyConfigQuery(ConfigQueryModel):
# In practice, it looks like AWS will only duplicate these resources if you've "used" any policies in the region, but since
# we can't really tell if this has happened in moto, we'll just bind this to the regions in your aggregator
from moto.config.models import CONFIG_REGIONS
aggregated_regions = []
aggregator_sources = aggregator.get(
"account_aggregation_sources"
@ -222,7 +219,7 @@ class PolicyConfigQuery(ConfigQueryModel):
for source in aggregator_sources:
source_dict = source.__dict__
if source_dict["all_aws_regions"]:
aggregated_regions = CONFIG_REGIONS
aggregated_regions = boto3.Session().get_available_regions("config")
break
for region in source_dict["aws_regions"]:
aggregated_regions.append(region)
@ -247,14 +244,10 @@ class PolicyConfigQuery(ConfigQueryModel):
duplicate_policy_list, key=lambda policy: policy["_id"]
)
# sorted_policy_ids matches indicies of sorted_policies
sorted_policy_ids = list(map(lambda policy: policy["_id"], sorted_policies))
else:
# Non-aggregated queries are in the else block, and we can treat these like a normal config resource
# Pagination logic, sort by role id
sorted_policies = sorted(policy_list, key=lambda role: role.id)
# sorted_policy_ids matches indicies of sorted_policies
sorted_policy_ids = list(map(lambda policy: policy.id, sorted_policies))
new_token = None
@ -262,16 +255,22 @@ class PolicyConfigQuery(ConfigQueryModel):
if not next_token:
start = 0
else:
if next_token not in sorted_policy_ids:
try:
# Find the index of the next
start = next(
index
for (index, p) in enumerate(sorted_policies)
if next_token == (p["_id"] if aggregator else p.id)
)
except StopIteration:
raise InvalidNextTokenException()
start = sorted_policy_ids.index(next_token)
# Get the list of items to collect:
policy_list = sorted_policies[start : (start + limit)]
if len(sorted_policies) > (start + limit):
new_token = sorted_policy_ids[start + limit]
record = sorted_policies[start + limit]
new_token = record["_id"] if aggregator else record.id
return (
[

6
tests/test_iam/test_iam.py
View File

@ -3235,7 +3235,8 @@ def test_role_config_dict():
def test_role_config_client():
from moto.iam.models import ACCOUNT_ID
from moto.iam.utils import random_resource_id
from moto.config.models import CONFIG_REGIONS
CONFIG_REGIONS = boto3.Session().get_available_regions("config")
iam_client = boto3.client("iam", region_name="us-west-2")
config_client = boto3.client("config", region_name="us-west-2")
@ -3659,7 +3660,8 @@ def test_policy_config_dict():
def test_policy_config_client():
from moto.iam.models import ACCOUNT_ID
from moto.iam.utils import random_policy_id
from moto.config.models import CONFIG_REGIONS
CONFIG_REGIONS = boto3.Session().get_available_regions("config")
basic_policy = {
"Version": "2012-10-17",