Skip to content
This repository has been archived by the owner on Jun 13, 2024. It is now read-only.

Commit

Permalink
Refactor region selection to facilitate regular expressions
Browse files Browse the repository at this point in the history 
[Issue(s) #50, #53]
  • Loading branch information
@sopel
sopel committed Feb 26, 2014
1 parent 5187a87 commit b54bf75
Show file tree
Hide file tree
Showing 5 changed files with 118 additions and 99 deletions.
36 changes: 12 additions & 24 deletions botocross/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

import argparse
import logging
import re
botocross_log = logging.getLogger('botocross')

def configure_logger(logger, level):
Expand Down Expand Up @@ -78,7 +79,8 @@ def build_common_parser():

def build_region_parser():
parser = argparse.ArgumentParser(add_help=False)
parser.add_argument("-r", "--region", help="A region substring selector (e.g. 'us-west')")
parser.add_argument("-r", "--region", default="(ap|eu|sa|us)-[enws]",
help="A regular expression based region substring selector [default '(ap|eu|sa|us)-[enws]']")
return parser

def build_filter_parser(resource_name, add_ids=True):
Expand All @@ -102,7 +104,7 @@ def build_backup_parser(resource_name, expire_only=False, backup_retention=None)
parser.add_argument("-br", "--backup_retention", type=int, default=backup_retention,
help="The number of backups to retain (correlated via backup_set). [default: {0}]".format(default_retention_help))
parser.add_argument("-bs", "--backup_set", default='default',
help="A backup set name (determines retention correlation). [default: 'default'")
help="A backup set name (determines retention correlation). [default: 'default']")
if not expire_only:
parser.add_argument("-bt", "--backup_timeout", default=None,
help="Maximum duration to await successful resource creation - an ISO 8601 duration, e.g. 'PT8H' (8 hours). [default: None, i.e. don't await]")
Expand All @@ -113,21 +115,15 @@ def build_backup_parser(resource_name, expire_only=False, backup_retention=None)
def parse_credentials(args):
return {'aws_access_key_id': args.aws_access_key_id, 'aws_secret_access_key': args.aws_secret_access_key}

def is_region_selected(region, name):
return True if region.name.find(name) != -1 else False

def is_govcloud(region):
return True if region.name.find('gov') != -1 else False

def filter_list_by_attribute(includes, excludes, attribute):
excluded_ids = set([getattr(exclude, attribute) for exclude in excludes])
return [include for include in includes if getattr(include, attribute) not in excluded_ids]

def filter_regions(regions, region, include_govcloud=False, only_govcloud=False):
if not (include_govcloud or only_govcloud):
regions = filter(lambda x: not is_govcloud(x), regions)
if only_govcloud:
regions = filter(lambda x: is_govcloud(x), regions)
def is_region_selected(region, name):
pattern = re.compile(name, re.IGNORECASE)
return True if re.search(pattern, region.name) != None else False

def filter_regions(regions, region):
if region:
botocross_log.info("... (filtered by region '" + region + "')")
regions = filter(lambda x: is_region_selected(x, region), regions)
Expand All @@ -136,19 +132,11 @@ def filter_regions(regions, region, include_govcloud=False, only_govcloud=False)
# REVIEW: remove this S3 legacy induced partial duplication, if possible.
def is_region_selected_s3(region, name):
from botocross.s3 import RegionMap
return True if RegionMap[region].find(name) != -1 else False

# REVIEW: remove this S3 legacy induced partial duplication, if possible.
def is_govcloud_s3(region):
from botocross.s3 import RegionMap
return True if RegionMap[region].find('gov') != -1 else False
pattern = re.compile(name, re.IGNORECASE)
return True if re.search(pattern, RegionMap[region]) != None else False

# REVIEW: remove this S3 legacy induced partial duplication, if possible.
def filter_regions_s3(regions, region, include_govcloud=False, only_govcloud=False):
if not (include_govcloud or only_govcloud):
regions = filter(lambda x: not is_govcloud_s3(x), regions)
if only_govcloud:
regions = filter(lambda x: is_govcloud_s3(x), regions)
def filter_regions_s3(regions, region):
if region:
botocross_log.info("... (filtered by S3 region '" + region + "')")
regions = filter(lambda x: is_region_selected_s3(x, region), regions)
Expand Down
3 changes: 2 additions & 1 deletion botocross/s3/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,8 @@
'APNortheast': 'ap-northeast-1',
'APSoutheast': 'ap-southeast-1',
'APSoutheast2': 'ap-southeast-2',
'USGovWest': 'us-gov-west-1'}
'USGovWest': 'us-gov-west-1',
'CNNorth1': 'cn-north-1'}

# NOTE: S3 region handling differs in an unfortunate way (likely a legacy issue) and requires special treatment.
def class_iterator(Class):
Expand Down
2 changes: 1 addition & 1 deletion setup.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@
license="MIT",
platforms="Posix; MacOS X; Windows",
install_requires=[
"boto >= 2.15.0",
"boto >= 2.24.0",
"isodate >= 0.4.8",
],
classifiers=[
Expand Down
4 changes: 2 additions & 2 deletions tests/s3/test_package.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,9 @@
import unittest

class S3PackageTest(unittest.TestCase):
num_regions = 9;
num_regions = 10;

def test_region_map(self):
from botocross.s3 import RegionMap

self.assertTrue(self.num_regions == len(RegionMap))
self.assertEquals(self.num_regions, len(RegionMap))
172 changes: 101 additions & 71 deletions tests/test_package.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,10 @@
log = logging.getLogger('botocross')

class TestPackage(unittest.TestCase):
num_regions = 9;
num_regions = 10;
num_regions_global = 8;
num_regions_gov = 1;
num_regions_cn = 1;

def test_configure_logging(self):
from botocross import configure_logging
Expand Down Expand Up @@ -71,104 +73,132 @@ def test_build_filter_params_with_identical_keys(self):
self.assertTrue(isinstance(v, list))
self.assertEquals(2, len(v))

def test_filter_regions(self):
from botocross import filter_regions
import boto.ec2
def test_region_numbers(self):
self.assertEquals(self.num_regions , self.num_regions_global + self.num_regions_gov + self.num_regions_cn)

def filter_regions_all(self, regions, filter):
log.info("... [All]")

log.info("Testing region filter (GovCloud excluded):")
all_regions = boto.ec2.regions()
filtered_regions = filter_regions(all_regions, "us-west-1")
filtered_regions = filter(regions, "us-west-1")
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions(all_regions, "us-west")
filtered_regions = filter(regions, "us-west")
self.assertEquals(2, len(filtered_regions))
filtered_regions = filter_regions(all_regions, "west-1")
filtered_regions = filter(regions, "west")
self.assertEquals(4, len(filtered_regions))
filtered_regions = filter(regions, "")
self.assertEquals(self.num_regions, len(filtered_regions))
filtered_regions = filter(regions, None)
self.assertEquals(self.num_regions , len(filtered_regions))

def filter_regions_global_only(self, regions, filter):
log.info("... [Global only]")

filtered_regions = filter(regions, "us-west-1")
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter(regions, "us-west")
self.assertEquals(2, len(filtered_regions))
filtered_regions = filter_regions(all_regions, "west")
filtered_regions = filter(regions, "west")
self.assertEquals(4, len(filtered_regions))
filtered_regions = filter(regions, "(ap|eu|sa|us)-[enws]")
self.assertEquals(self.num_regions_global, len(filtered_regions))

def filter_regions_gov_excluded(self, regions, filter):
log.info("... [GovCloud excluded]")

filtered_regions = filter(regions, "us-west-1")
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter(regions, "us-west")
self.assertEquals(2, len(filtered_regions))
filtered_regions = filter(regions, "(?<!gov)-west-1")
self.assertEquals(2, len(filtered_regions))
filtered_regions = filter(regions, "(?<!gov)-west")
self.assertEquals(3, len(filtered_regions))
filtered_regions = filter_regions(all_regions, None)
filtered_regions = filter(regions, "(?<!gov)-[nwse]")
self.assertEquals(self.num_regions - self.num_regions_gov, len(filtered_regions))

def test_filter_regions_gov_included(self):
from botocross import filter_regions
import boto.ec2
def filter_regions_gov_included(self, regions, filter):
log.info("... [GovCloud included]")

log.info("Testing region filter (GovCloud included):")
all_regions = boto.ec2.regions()
filtered_regions = filter_regions(all_regions, "us-west-1", True, False)
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions(all_regions, "us-west", True, False)
filtered_regions = filter(regions, "us.*-west-1")
self.assertEquals(2, len(filtered_regions))
filtered_regions = filter_regions(all_regions, "west-1", True, False)
filtered_regions = filter(regions, "us.*-west")
self.assertEquals(3, len(filtered_regions))
filtered_regions = filter(regions, "west-1")
self.assertEquals(3, len(filtered_regions))
filtered_regions = filter_regions(all_regions, "west", True, False)
filtered_regions = filter(regions, "west")
self.assertEquals(4, len(filtered_regions))
filtered_regions = filter_regions(all_regions, None, True, False)
filtered_regions = filter(regions, None)
self.assertEquals(self.num_regions, len(filtered_regions))

def test_filter_regions_gov_only(self):
from botocross import filter_regions
import boto.ec2
def filter_regions_gov_only(self, regions, filter):
log.info("... [GovCloud only]")

log.info("Testing region filter (GovCloud only):")
all_regions = boto.ec2.regions()
filtered_regions = filter_regions(all_regions, "us-gov-west-1", False, True)
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions(all_regions, "us-gov-west", False, True)
filtered_regions = filter(regions, "us-gov-west-1")
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions(all_regions, "gov-west", False, True)
filtered_regions = filter(regions, "us-gov-west")
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions(all_regions, "gov", False, True)
filtered_regions = filter(regions, "gov-west")
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions(all_regions, None, False, True)
filtered_regions = filter(regions, "gov")
self.assertEquals(self.num_regions_gov, len(filtered_regions))

def test_filter_regions_s3(self):
from botocross import filter_regions_s3
from botocross.s3 import RegionMap
def filter_regions_cn_excluded(self, regions, filter):
log.info("... [China excluded]")

log.info("Testing S3 region filter (GovCloud excluded):")
all_regions = RegionMap
filtered_regions = filter_regions_s3(all_regions, "us-west-1")
filtered_regions = filter(regions, "ap-northeast-1")
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions_s3(all_regions, "us-west")
self.assertEquals(2, len(filtered_regions))
filtered_regions = filter_regions_s3(all_regions, "west-1")
self.assertEquals(2, len(filtered_regions))
filtered_regions = filter_regions_s3(all_regions, "west")
self.assertEquals(3, len(filtered_regions))
filtered_regions = filter_regions_s3(all_regions, None)
self.assertEquals(self.num_regions - self.num_regions_gov, len(filtered_regions))
filtered_regions = filter(regions, "ap-northeast")
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter(regions, "(?<!cn)-north")
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter(regions, "(?<!cn)-[nwse]")
self.assertEquals(self.num_regions - self.num_regions_cn, len(filtered_regions))

def test_filter_regions_s3_gov_included(self):
from botocross import filter_regions_s3
from botocross.s3 import RegionMap
def filter_regions_cn_included(self, regions, filter):
log.info("... [China included]")

log.info("Testing S3 region filter (GovCloud included):")
all_regions = RegionMap
filtered_regions = filter_regions_s3(all_regions, "us-gov-west-1", True, False)
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions_s3(all_regions, "us-gov-west", True, False)
filtered_regions = filter(regions, "cn-north-1")
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions_s3(all_regions, "gov-west", True, False)
filtered_regions = filter(regions, "cn-north")
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions_s3(all_regions, "gov", True, False)
filtered_regions = filter(regions, "north-1")
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions_s3(all_regions, None, True, False)
filtered_regions = filter(regions, "north")
self.assertEquals(2, len(filtered_regions))
filtered_regions = filter(regions, None)
self.assertEquals(self.num_regions, len(filtered_regions))

def test_filter_regions_s3_gov_only(self):
from botocross import filter_regions_s3
from botocross.s3 import RegionMap
def filter_regions_cn_only(self, regions, filter):
log.info("... [China only]")

log.info("Testing S3 region filter (GovCloud only):")
all_regions = RegionMap
filtered_regions = filter_regions_s3(all_regions, "us-gov-west-1", False, True)
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions_s3(all_regions, "us-gov-west", False, True)
filtered_regions = filter(regions, "cn-north-1")
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions_s3(all_regions, "gov-west", False, True)
filtered_regions = filter(regions, "cn-north")
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions_s3(all_regions, "gov", False, True)
self.assertEquals(1, len(filtered_regions))
filtered_regions = filter_regions_s3(all_regions, None, False, True)
self.assertEquals(self.num_regions_gov, len(filtered_regions))
filtered_regions = filter(regions, "cn")
self.assertEquals(self.num_regions_cn, len(filtered_regions))

def filter_regions(self, regions, filter):
self.filter_regions_all(regions, filter)
self.filter_regions_global_only(regions, filter)
self.filter_regions_gov_excluded(regions, filter)
self.filter_regions_gov_included(regions, filter)
self.filter_regions_gov_only(regions, filter)
self.filter_regions_cn_excluded(regions, filter)
self.filter_regions_cn_included(regions, filter)
self.filter_regions_cn_only(regions, filter)

def test_filter_regions(self):
from botocross import filter_regions as filter
import boto.ec2

log.info("Testing region filters:")
regions = boto.ec2.regions()
self.filter_regions(regions, filter)

def test_filter_regions_s3(self):
from botocross import filter_regions_s3 as filter
from botocross.s3 import RegionMap as regions

log.info("Testing S3 region filters:")
self.filter_regions(regions, filter)

0 comments on commit b54bf75

Please sign in to comment.