SimpleDB - initial implementation (#4585)
This commit is contained in:
		
							parent
							
								
									958a129f97
								
							
						
					
					
						commit
						8b5e926ec1
					
				| @ -4087,6 +4087,22 @@ | ||||
| - [ ] update_workteam | ||||
| </details> | ||||
| 
 | ||||
| ## sdb | ||||
| <details> | ||||
| <summary>50% implemented</summary> | ||||
| 
 | ||||
| - [ ] batch_delete_attributes | ||||
| - [ ] batch_put_attributes | ||||
| - [X] create_domain | ||||
| - [ ] delete_attributes | ||||
| - [X] delete_domain | ||||
| - [ ] domain_metadata | ||||
| - [X] get_attributes | ||||
| - [X] list_domains | ||||
| - [X] put_attributes | ||||
| - [ ] select | ||||
| </details> | ||||
| 
 | ||||
| ## secretsmanager | ||||
| <details> | ||||
| <summary>68% implemented</summary> | ||||
| @ -4804,7 +4820,6 @@ | ||||
| - sagemaker-runtime | ||||
| - savingsplans | ||||
| - schemas | ||||
| - sdb | ||||
| - securityhub | ||||
| - serverlessrepo | ||||
| - service-quotas | ||||
|  | ||||
							
								
								
									
										52
									
								
								docs/docs/services/sdb.rst
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										52
									
								
								docs/docs/services/sdb.rst
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,52 @@ | ||||
| .. _implementedservice_sdb: | ||||
| 
 | ||||
| .. |start-h3| raw:: html | ||||
| 
 | ||||
|     <h3> | ||||
| 
 | ||||
| .. |end-h3| raw:: html | ||||
| 
 | ||||
|     </h3> | ||||
| 
 | ||||
| === | ||||
| sdb | ||||
| === | ||||
| 
 | ||||
| 
 | ||||
| 
 | ||||
| |start-h3| Example usage |end-h3| | ||||
| 
 | ||||
| .. sourcecode:: python | ||||
| 
 | ||||
|             @mock_sdb | ||||
|             def test_sdb_behaviour: | ||||
|                 boto3.client("sdb") | ||||
|                 ... | ||||
| 
 | ||||
| 
 | ||||
| 
 | ||||
| |start-h3| Implemented features for this service |end-h3| | ||||
| 
 | ||||
| - [ ] batch_delete_attributes | ||||
| - [ ] batch_put_attributes | ||||
| - [X] create_domain | ||||
| - [ ] delete_attributes | ||||
| - [X] delete_domain | ||||
| - [ ] domain_metadata | ||||
| - [X] get_attributes | ||||
|    | ||||
|         Behaviour for the consistent_read-attribute is not yet implemented | ||||
|          | ||||
| 
 | ||||
| - [X] list_domains | ||||
|    | ||||
|         The `max_number_of_domains` and `next_token` parameter have not been implemented yet - we simply return all domains. | ||||
|          | ||||
| 
 | ||||
| - [X] put_attributes | ||||
|    | ||||
|         Behaviour for the expected-attribute is not yet implemented. | ||||
|          | ||||
| 
 | ||||
| - [ ] select | ||||
| 
 | ||||
| @ -168,6 +168,7 @@ mock_mediastoredata = lazy_load( | ||||
| ) | ||||
| mock_efs = lazy_load(".efs", "mock_efs") | ||||
| mock_wafv2 = lazy_load(".wafv2", "mock_wafv2") | ||||
| mock_sdb = lazy_load(".sdb", "mock_sdb", boto3_name="sdb") | ||||
| 
 | ||||
| 
 | ||||
| def mock_all(): | ||||
|  | ||||
| @ -109,6 +109,7 @@ backend_url_patterns = [ | ||||
|         ), | ||||
|     ), | ||||
|     ("sagemaker", re.compile("https?://api.sagemaker\\.(.+)\\.amazonaws.com")), | ||||
|     ("sdb", re.compile("https?://sdb\\.(.+)\\.amazonaws\\.com")), | ||||
|     ("secretsmanager", re.compile("https?://secretsmanager\\.(.+)\\.amazonaws\\.com")), | ||||
|     ("ses", re.compile("https?://email\\.(.+)\\.amazonaws\\.com")), | ||||
|     ("ses", re.compile("https?://ses\\.(.+)\\.amazonaws\\.com")), | ||||
|  | ||||
							
								
								
									
										5
									
								
								moto/sdb/__init__.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										5
									
								
								moto/sdb/__init__.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,5 @@ | ||||
| """sdb module initialization; sets value for base decorator.""" | ||||
| from .models import sdb_backends | ||||
| from ..core.models import base_decorator | ||||
| 
 | ||||
| mock_sdb = base_decorator(sdb_backends) | ||||
							
								
								
									
										45
									
								
								moto/sdb/exceptions.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										45
									
								
								moto/sdb/exceptions.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,45 @@ | ||||
| """Exceptions raised by the sdb service.""" | ||||
| from moto.core.exceptions import RESTError | ||||
| 
 | ||||
| 
 | ||||
| SDB_ERROR = """<?xml version="1.0"?> | ||||
| <Response> | ||||
|     <Errors> | ||||
|         <Error> | ||||
|             <Code>{{ error_type }}</Code> | ||||
|             <Message>{{ message }}</Message> | ||||
|             <BoxUsage>0.0055590278</BoxUsage> | ||||
|         </Error> | ||||
|     </Errors> | ||||
|     <RequestID>ba3a8c86-dc37-0a45-ef44-c6cf7876a62f</RequestID> | ||||
| </Response>""" | ||||
| 
 | ||||
| 
 | ||||
| class InvalidParameterError(RESTError): | ||||
|     code = 400 | ||||
| 
 | ||||
|     def __init__(self, **kwargs): | ||||
|         kwargs.setdefault("template", "sdb_error") | ||||
|         self.templates["sdb_error"] = SDB_ERROR | ||||
|         kwargs["error_type"] = "InvalidParameterValue" | ||||
|         super().__init__(**kwargs) | ||||
| 
 | ||||
| 
 | ||||
| class InvalidDomainName(InvalidParameterError): | ||||
|     code = 400 | ||||
| 
 | ||||
|     def __init__(self, domain_name): | ||||
|         super().__init__( | ||||
|             message=f"Value ({domain_name}) for parameter DomainName is invalid. " | ||||
|         ) | ||||
| 
 | ||||
| 
 | ||||
| class UnknownDomainName(RESTError): | ||||
|     code = 400 | ||||
| 
 | ||||
|     def __init__(self, **kwargs): | ||||
|         kwargs.setdefault("template", "sdb_error") | ||||
|         self.templates["sdb_error"] = SDB_ERROR | ||||
|         kwargs["error_type"] = "NoSuchDomain" | ||||
|         kwargs["message"] = "The specified domain does not exist." | ||||
|         super().__init__(**kwargs) | ||||
							
								
								
									
										109
									
								
								moto/sdb/models.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										109
									
								
								moto/sdb/models.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,109 @@ | ||||
| """SimpleDBBackend class with methods for supported APIs.""" | ||||
| import re | ||||
| from boto3 import Session | ||||
| from collections import defaultdict | ||||
| from moto.core import BaseBackend, BaseModel | ||||
| from threading import Lock | ||||
| 
 | ||||
| from .exceptions import InvalidDomainName, UnknownDomainName | ||||
| 
 | ||||
| 
 | ||||
| class FakeItem(BaseModel): | ||||
|     def __init__(self): | ||||
|         self.attributes = [] | ||||
|         self.lock = Lock() | ||||
| 
 | ||||
|     def get_attributes(self, names): | ||||
|         if not names: | ||||
|             return self.attributes | ||||
|         return [attr for attr in self.attributes if attr["name"] in names] | ||||
| 
 | ||||
|     def put_attributes(self, attributes): | ||||
|         # Replacing attributes involves quite a few loops | ||||
|         # Lock this, so we know noone else touches this list while we're operating on it | ||||
|         with self.lock: | ||||
|             for attr in attributes: | ||||
|                 if attr.get("replace", "false").lower() == "true": | ||||
|                     self._remove_attributes(attr["name"]) | ||||
|                 self.attributes.append(attr) | ||||
| 
 | ||||
|     def _remove_attributes(self, name): | ||||
|         self.attributes = [attr for attr in self.attributes if attr["name"] != name] | ||||
| 
 | ||||
| 
 | ||||
| class FakeDomain(BaseModel): | ||||
|     def __init__(self, name): | ||||
|         self.name = name | ||||
|         self.items = defaultdict(FakeItem) | ||||
| 
 | ||||
|     def get(self, item_name, attribute_names): | ||||
|         item = self.items[item_name] | ||||
|         return item.get_attributes(attribute_names) | ||||
| 
 | ||||
|     def put(self, item_name, attributes): | ||||
|         item = self.items[item_name] | ||||
|         item.put_attributes(attributes) | ||||
| 
 | ||||
| 
 | ||||
| class SimpleDBBackend(BaseBackend): | ||||
|     def __init__(self, region_name=None): | ||||
|         self.region_name = region_name | ||||
|         self.domains = dict() | ||||
| 
 | ||||
|     def reset(self): | ||||
|         region_name = self.region_name | ||||
|         self.__dict__ = {} | ||||
|         self.__init__(region_name) | ||||
| 
 | ||||
|     def create_domain(self, domain_name): | ||||
|         self._validate_domain_name(domain_name) | ||||
|         self.domains[domain_name] = FakeDomain(name=domain_name) | ||||
| 
 | ||||
|     def list_domains(self, max_number_of_domains, next_token): | ||||
|         """ | ||||
|         The `max_number_of_domains` and `next_token` parameter have not been implemented yet - we simply return all domains. | ||||
|         """ | ||||
|         return self.domains.keys(), None | ||||
| 
 | ||||
|     def delete_domain(self, domain_name): | ||||
|         self._validate_domain_name(domain_name) | ||||
|         # Ignore unknown domains - AWS does the same | ||||
|         self.domains.pop(domain_name, None) | ||||
| 
 | ||||
|     def _validate_domain_name(self, domain_name): | ||||
|         # Domain Name needs to have at least 3 chars | ||||
|         # Can only contain characters: a-z, A-Z, 0-9, '_', '-', and '.' | ||||
|         if not re.match("^[a-zA-Z0-9-_.]{3,}$", domain_name): | ||||
|             raise InvalidDomainName(domain_name) | ||||
| 
 | ||||
|     def _get_domain(self, domain_name): | ||||
|         if domain_name not in self.domains: | ||||
|             raise UnknownDomainName() | ||||
|         return self.domains[domain_name] | ||||
| 
 | ||||
|     def get_attributes(self, domain_name, item_name, attribute_names, consistent_read): | ||||
|         """ | ||||
|         Behaviour for the consistent_read-attribute is not yet implemented | ||||
|         """ | ||||
|         self._validate_domain_name(domain_name) | ||||
|         domain = self._get_domain(domain_name) | ||||
|         return domain.get(item_name, attribute_names) | ||||
| 
 | ||||
|     def put_attributes(self, domain_name, item_name, attributes, expected): | ||||
|         """ | ||||
|         Behaviour for the expected-attribute is not yet implemented. | ||||
|         """ | ||||
|         self._validate_domain_name(domain_name) | ||||
|         domain = self._get_domain(domain_name) | ||||
|         domain.put(item_name, attributes) | ||||
| 
 | ||||
| 
 | ||||
| sdb_backends = {} | ||||
| for available_region in Session().get_available_regions("sdb"): | ||||
|     sdb_backends[available_region] = SimpleDBBackend(available_region) | ||||
| for available_region in Session().get_available_regions( | ||||
|     "sdb", partition_name="aws-us-gov" | ||||
| ): | ||||
|     sdb_backends[available_region] = SimpleDBBackend(available_region) | ||||
| for available_region in Session().get_available_regions("sdb", partition_name="aws-cn"): | ||||
|     sdb_backends[available_region] = SimpleDBBackend(available_region) | ||||
							
								
								
									
										100
									
								
								moto/sdb/responses.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										100
									
								
								moto/sdb/responses.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,100 @@ | ||||
| from moto.core.responses import BaseResponse | ||||
| from .models import sdb_backends | ||||
| 
 | ||||
| 
 | ||||
| class SimpleDBResponse(BaseResponse): | ||||
|     @property | ||||
|     def sdb_backend(self): | ||||
|         return sdb_backends[self.region] | ||||
| 
 | ||||
|     def create_domain(self): | ||||
|         domain_name = self._get_param("DomainName") | ||||
|         self.sdb_backend.create_domain(domain_name=domain_name,) | ||||
|         template = self.response_template(CREATE_DOMAIN_TEMPLATE) | ||||
|         return template.render() | ||||
| 
 | ||||
|     def delete_domain(self): | ||||
|         domain_name = self._get_param("DomainName") | ||||
|         self.sdb_backend.delete_domain(domain_name=domain_name,) | ||||
|         template = self.response_template(DELETE_DOMAIN_TEMPLATE) | ||||
|         return template.render() | ||||
| 
 | ||||
|     def list_domains(self): | ||||
|         max_number_of_domains = self._get_int_param("MaxNumberOfDomains") | ||||
|         next_token = self._get_param("NextToken") | ||||
|         domain_names, next_token = self.sdb_backend.list_domains( | ||||
|             max_number_of_domains=max_number_of_domains, next_token=next_token, | ||||
|         ) | ||||
|         template = self.response_template(LIST_DOMAINS_TEMPLATE) | ||||
|         return template.render(domain_names=domain_names, next_token=next_token) | ||||
| 
 | ||||
|     def get_attributes(self): | ||||
|         domain_name = self._get_param("DomainName") | ||||
|         item_name = self._get_param("ItemName") | ||||
|         attribute_names = self._get_multi_param("AttributeName.") | ||||
|         consistent_read = self._get_param("ConsistentRead") | ||||
|         attributes = self.sdb_backend.get_attributes( | ||||
|             domain_name=domain_name, | ||||
|             item_name=item_name, | ||||
|             attribute_names=attribute_names, | ||||
|             consistent_read=consistent_read, | ||||
|         ) | ||||
|         template = self.response_template(GET_ATTRIBUTES_TEMPLATE) | ||||
|         return template.render(attributes=attributes) | ||||
| 
 | ||||
|     def put_attributes(self): | ||||
|         domain_name = self._get_param("DomainName") | ||||
|         item_name = self._get_param("ItemName") | ||||
|         attributes = self._get_list_prefix("Attribute") | ||||
|         expected = self._get_param("Expected") | ||||
|         self.sdb_backend.put_attributes( | ||||
|             domain_name=domain_name, | ||||
|             item_name=item_name, | ||||
|             attributes=attributes, | ||||
|             expected=expected, | ||||
|         ) | ||||
|         template = self.response_template(PUT_ATTRIBUTES_TEMPLATE) | ||||
|         return template.render() | ||||
| 
 | ||||
| 
 | ||||
| CREATE_DOMAIN_TEMPLATE = """<?xml version="1.0" encoding="UTF-8"?> | ||||
| <CreateDomainResult  xmlns="http://sdb.amazonaws.com/doc/2009-04-15/"></CreateDomainResult> | ||||
| """ | ||||
| 
 | ||||
| 
 | ||||
| LIST_DOMAINS_TEMPLATE = """<?xml version="1.0" encoding="UTF-8"?> | ||||
| <ListDomainsResponse  xmlns="http://sdb.amazonaws.com/doc/2009-04-15/"> | ||||
|     <ListDomainsResult> | ||||
|         {% for name in domain_names %} | ||||
|             <DomainName>{{ name }}</DomainName> | ||||
|         {% endfor %} | ||||
|         <NextToken>{{ next_token }}</NextToken> | ||||
|     </ListDomainsResult> | ||||
| </ListDomainsResponse> | ||||
| """ | ||||
| 
 | ||||
| DELETE_DOMAIN_TEMPLATE = """<?xml version="1.0"?> | ||||
| <DeleteDomainResponse xmlns="http://sdb.amazonaws.com/doc/2009-04-15/"> | ||||
|   <ResponseMetadata> | ||||
|     <RequestId>64d9c3ac-ef19-2e3d-7a03-9ea46205eb71</RequestId> | ||||
|     <BoxUsage>0.0055590278</BoxUsage> | ||||
|   </ResponseMetadata> | ||||
| </DeleteDomainResponse>""" | ||||
| 
 | ||||
| PUT_ATTRIBUTES_TEMPLATE = """<?xml version="1.0" encoding="UTF-8"?> | ||||
| <PutAttributesResult xmlns="http://sdb.amazonaws.com/doc/2009-04-15/"></PutAttributesResult> | ||||
| """ | ||||
| 
 | ||||
| GET_ATTRIBUTES_TEMPLATE = """<GetAttributesResponse xmlns="http://sdb.amazonaws.com/doc/2009-04-15/"> | ||||
|   <ResponseMetadata> | ||||
|     <RequestId>1549581b-12b7-11e3-895e-1334aEXAMPLE</RequestId> | ||||
|   </ResponseMetadata> | ||||
|   <GetAttributesResult> | ||||
| {% for attribute in attributes %} | ||||
|       <Attribute> | ||||
|         <Name>{{ attribute["name"] }}</Name> | ||||
|         <Value>{{ attribute["value"] }}</Value> | ||||
|       </Attribute> | ||||
| {% endfor %} | ||||
|   </GetAttributesResult> | ||||
| </GetAttributesResponse>""" | ||||
							
								
								
									
										7
									
								
								moto/sdb/urls.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										7
									
								
								moto/sdb/urls.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,7 @@ | ||||
| from .responses import SimpleDBResponse | ||||
| 
 | ||||
| url_bases = [ | ||||
|     r"https?://sdb\.(.+)\.amazonaws\.com", | ||||
| ] | ||||
| 
 | ||||
| url_paths = {"{0}/$": SimpleDBResponse.dispatch} | ||||
| @ -42,6 +42,9 @@ SIGNING_ALIASES = { | ||||
|     "iotdata": "data.iot", | ||||
| } | ||||
| 
 | ||||
| # Some services are only recognizable by the version | ||||
| SERVICE_BY_VERSION = {"2009-04-15": "sdb"} | ||||
| 
 | ||||
| 
 | ||||
| class DomainDispatcherApplication(object): | ||||
|     """ | ||||
| @ -79,9 +82,10 @@ class DomainDispatcherApplication(object): | ||||
|                 ) | ||||
|             ) | ||||
| 
 | ||||
|     def infer_service_region_host(self, environ): | ||||
|     def infer_service_region_host(self, body, environ): | ||||
|         auth = environ.get("HTTP_AUTHORIZATION") | ||||
|         target = environ.get("HTTP_X_AMZ_TARGET") | ||||
|         service = None | ||||
|         if auth: | ||||
|             # Signed request | ||||
|             # Parse auth header to find service assuming a SigV4 request | ||||
| @ -100,18 +104,20 @@ class DomainDispatcherApplication(object): | ||||
|                 service, region = DEFAULT_SERVICE_REGION | ||||
|         else: | ||||
|             # Unsigned request | ||||
|             action = self.get_action_from_body(environ) | ||||
|             action = self.get_action_from_body(body) | ||||
|             if target: | ||||
|                 service, _ = target.split(".", 1) | ||||
|                 service, region = UNSIGNED_REQUESTS.get(service, DEFAULT_SERVICE_REGION) | ||||
|             elif action and action in UNSIGNED_ACTIONS: | ||||
|                 # See if we can match the Action to a known service | ||||
|                 service, region = UNSIGNED_ACTIONS.get(action) | ||||
|             else: | ||||
|             if not service: | ||||
|                 service, region = self.get_service_from_body(body, environ) | ||||
|             if not service: | ||||
|                 service, region = self.get_service_from_path(environ) | ||||
|                 if not service: | ||||
|                     # S3 is the last resort when the target is also unknown | ||||
|                     service, region = DEFAULT_SERVICE_REGION | ||||
|             if not service: | ||||
|                 # S3 is the last resort when the target is also unknown | ||||
|                 service, region = DEFAULT_SERVICE_REGION | ||||
| 
 | ||||
|         if service == "mediastore" and not target: | ||||
|             # All MediaStore API calls have a target header | ||||
| @ -161,8 +167,9 @@ class DomainDispatcherApplication(object): | ||||
|         with self.lock: | ||||
|             backend = self.get_backend_for_host(host) | ||||
|             if not backend: | ||||
|                 # No regular backend found; try parsing other headers | ||||
|                 host = self.infer_service_region_host(environ) | ||||
|                 # No regular backend found; try parsing body/other headers | ||||
|                 body = self._get_body(environ) | ||||
|                 host = self.infer_service_region_host(body, environ) | ||||
|                 backend = self.get_backend_for_host(host) | ||||
| 
 | ||||
|             app = self.app_instances.get(backend, None) | ||||
| @ -171,7 +178,7 @@ class DomainDispatcherApplication(object): | ||||
|                 self.app_instances[backend] = app | ||||
|             return app | ||||
| 
 | ||||
|     def get_action_from_body(self, environ): | ||||
|     def _get_body(self, environ): | ||||
|         body = None | ||||
|         try: | ||||
|             # AWS requests use querystrings as the body (Action=x&Data=y&...) | ||||
| @ -181,15 +188,38 @@ class DomainDispatcherApplication(object): | ||||
|             request_body_size = int(environ["CONTENT_LENGTH"]) | ||||
|             if simple_form and request_body_size: | ||||
|                 body = environ["wsgi.input"].read(request_body_size).decode("utf-8") | ||||
|                 body_dict = dict(x.split("=") for x in body.split("&")) | ||||
|                 return body_dict["Action"] | ||||
|         except (KeyError, ValueError): | ||||
|             pass | ||||
|         finally: | ||||
|             if body: | ||||
|                 # We've consumed the body = need to reset it | ||||
|                 environ["wsgi.input"] = io.StringIO(body) | ||||
|         return None | ||||
|         return body | ||||
| 
 | ||||
|     def get_service_from_body(self, body, environ): | ||||
|         # Some services have the SDK Version in the body | ||||
|         # If the version is unique, we can derive the service from it | ||||
|         version = self.get_version_from_body(body) | ||||
|         if version and version in SERVICE_BY_VERSION: | ||||
|             # Boto3/1.20.7 Python/3.8.10 Linux/5.11.0-40-generic Botocore/1.23.7 region/eu-west-1 | ||||
|             region = environ.get("HTTP_USER_AGENT", "").split("/")[-1] | ||||
|             return SERVICE_BY_VERSION[version], region | ||||
|         return None, None | ||||
| 
 | ||||
|     def get_version_from_body(self, body): | ||||
|         try: | ||||
|             body_dict = dict(x.split("=") for x in body.split("&")) | ||||
|             return body_dict["Version"] | ||||
|         except (AttributeError, KeyError, ValueError): | ||||
|             return None | ||||
| 
 | ||||
|     def get_action_from_body(self, body): | ||||
|         try: | ||||
|             # AWS requests use querystrings as the body (Action=x&Data=y&...) | ||||
|             body_dict = dict(x.split("=") for x in body.split("&")) | ||||
|             return body_dict["Action"] | ||||
|         except (AttributeError, KeyError, ValueError): | ||||
|             return None | ||||
| 
 | ||||
|     def get_service_from_path(self, environ): | ||||
|         # Moto sometimes needs to send a HTTP request to itself | ||||
| @ -198,7 +228,7 @@ class DomainDispatcherApplication(object): | ||||
|             path_info = environ.get("PATH_INFO", "/") | ||||
|             service, region = path_info[1 : path_info.index("/", 1)].split("_") | ||||
|             return service, region | ||||
|         except (KeyError, ValueError): | ||||
|         except (AttributeError, KeyError, ValueError): | ||||
|             return None, None | ||||
| 
 | ||||
|     def __call__(self, environ, start_response): | ||||
|  | ||||
							
								
								
									
										0
									
								
								tests/test_sdb/__init__.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										0
									
								
								tests/test_sdb/__init__.py
									
									
									
									
									
										Normal file
									
								
							
							
								
								
									
										167
									
								
								tests/test_sdb/test_sdb_attributes.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										167
									
								
								tests/test_sdb/test_sdb_attributes.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,167 @@ | ||||
| import boto3 | ||||
| import pytest | ||||
| import sure  # noqa # pylint: disable=unused-import | ||||
| 
 | ||||
| from botocore.exceptions import ClientError | ||||
| from moto import mock_sdb | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| def test_put_attributes_unknown_domain(): | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     with pytest.raises(ClientError) as exc: | ||||
|         sdb.put_attributes( | ||||
|             DomainName="aaaa", ItemName="asdf", Attributes=[{"Name": "a", "Value": "b"}] | ||||
|         ) | ||||
|     err = exc.value.response["Error"] | ||||
|     err["Code"].should.equal("NoSuchDomain") | ||||
|     err["Message"].should.equal("The specified domain does not exist.") | ||||
|     err.should.have.key("BoxUsage") | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| def test_put_attributes_invalid_domain(): | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     with pytest.raises(ClientError) as exc: | ||||
|         sdb.put_attributes( | ||||
|             DomainName="a", ItemName="asdf", Attributes=[{"Name": "a", "Value": "b"}] | ||||
|         ) | ||||
|     err = exc.value.response["Error"] | ||||
|     err["Code"].should.equal("InvalidParameterValue") | ||||
|     err["Message"].should.equal("Value (a) for parameter DomainName is invalid. ") | ||||
|     err.should.have.key("BoxUsage") | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| def test_get_attributes_unknown_domain(): | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     with pytest.raises(ClientError) as exc: | ||||
|         sdb.get_attributes(DomainName="aaaa", ItemName="asdf") | ||||
|     err = exc.value.response["Error"] | ||||
|     err["Code"].should.equal("NoSuchDomain") | ||||
|     err["Message"].should.equal("The specified domain does not exist.") | ||||
|     err.should.have.key("BoxUsage") | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| def test_get_attributes_invalid_domain(): | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     with pytest.raises(ClientError) as exc: | ||||
|         sdb.get_attributes(DomainName="a", ItemName="asdf") | ||||
|     err = exc.value.response["Error"] | ||||
|     err["Code"].should.equal("InvalidParameterValue") | ||||
|     err["Message"].should.equal("Value (a) for parameter DomainName is invalid. ") | ||||
|     err.should.have.key("BoxUsage") | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| def test_put_and_get_attributes(): | ||||
|     name = "mydomain" | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     sdb.create_domain(DomainName=name) | ||||
| 
 | ||||
|     sdb.put_attributes( | ||||
|         DomainName=name, ItemName="asdf", Attributes=[{"Name": "a", "Value": "b"}] | ||||
|     ) | ||||
| 
 | ||||
|     attrs = sdb.get_attributes(DomainName=name, ItemName="asdf")["Attributes"] | ||||
|     attrs.should.equal([{"Name": "a", "Value": "b"}]) | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| def test_put_multiple_and_get_attributes(): | ||||
|     name = "mydomain" | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     sdb.create_domain(DomainName=name) | ||||
| 
 | ||||
|     sdb.put_attributes( | ||||
|         DomainName=name, ItemName="asdf", Attributes=[{"Name": "a", "Value": "b"}] | ||||
|     ) | ||||
|     sdb.put_attributes( | ||||
|         DomainName=name, ItemName="jklp", Attributes=[{"Name": "a", "Value": "val"}] | ||||
|     ) | ||||
|     sdb.put_attributes( | ||||
|         DomainName=name, ItemName="asdf", Attributes=[{"Name": "a", "Value": "c"}] | ||||
|     ) | ||||
|     sdb.put_attributes( | ||||
|         DomainName=name, ItemName="asdf", Attributes=[{"Name": "d", "Value": "e"}] | ||||
|     ) | ||||
| 
 | ||||
|     attrs = sdb.get_attributes(DomainName=name, ItemName="asdf")["Attributes"] | ||||
|     attrs.should.equal( | ||||
|         [ | ||||
|             {"Name": "a", "Value": "b"}, | ||||
|             {"Name": "a", "Value": "c"}, | ||||
|             {"Name": "d", "Value": "e"}, | ||||
|         ] | ||||
|     ) | ||||
| 
 | ||||
|     attrs = sdb.get_attributes(DomainName=name, ItemName="jklp")["Attributes"] | ||||
|     attrs.should.equal([{"Name": "a", "Value": "val"}]) | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| def test_put_replace_and_get_attributes(): | ||||
|     name = "mydomain" | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     sdb.create_domain(DomainName=name) | ||||
| 
 | ||||
|     sdb.put_attributes( | ||||
|         DomainName=name, ItemName="asdf", Attributes=[{"Name": "a", "Value": "b"}] | ||||
|     ) | ||||
|     sdb.put_attributes( | ||||
|         DomainName=name, ItemName="asdf", Attributes=[{"Name": "a", "Value": "c"}] | ||||
|     ) | ||||
|     sdb.put_attributes( | ||||
|         DomainName=name, ItemName="asdf", Attributes=[{"Name": "d", "Value": "e"}] | ||||
|     ) | ||||
|     sdb.put_attributes( | ||||
|         DomainName=name, | ||||
|         ItemName="asdf", | ||||
|         Attributes=[ | ||||
|             {"Name": "a", "Value": "f", "Replace": True}, | ||||
|             {"Name": "d", "Value": "g"}, | ||||
|         ], | ||||
|     ) | ||||
| 
 | ||||
|     attrs = sdb.get_attributes(DomainName=name, ItemName="asdf")["Attributes"] | ||||
|     attrs.should.have.length_of(3) | ||||
|     attrs.should.contain({"Name": "a", "Value": "f"}) | ||||
|     attrs.should.contain({"Name": "d", "Value": "e"}) | ||||
|     attrs.should.contain({"Name": "d", "Value": "g"}) | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| def test_put_and_get_multiple_attributes(): | ||||
|     name = "mydomain" | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     sdb.create_domain(DomainName=name) | ||||
| 
 | ||||
|     sdb.put_attributes( | ||||
|         DomainName=name, | ||||
|         ItemName="asdf", | ||||
|         Attributes=[{"Name": "a", "Value": "b"}, {"Name": "attr2", "Value": "myvalue"}], | ||||
|     ) | ||||
| 
 | ||||
|     attrs = sdb.get_attributes(DomainName=name, ItemName="asdf")["Attributes"] | ||||
|     attrs.should.equal( | ||||
|         [{"Name": "a", "Value": "b"}, {"Name": "attr2", "Value": "myvalue"}] | ||||
|     ) | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| def test_get_attributes_by_name(): | ||||
|     name = "mydomain" | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     sdb.create_domain(DomainName=name) | ||||
| 
 | ||||
|     sdb.put_attributes( | ||||
|         DomainName=name, | ||||
|         ItemName="asdf", | ||||
|         Attributes=[{"Name": "a", "Value": "b"}, {"Name": "attr2", "Value": "myvalue"}], | ||||
|     ) | ||||
| 
 | ||||
|     attrs = sdb.get_attributes( | ||||
|         DomainName=name, ItemName="asdf", AttributeNames=["attr2"] | ||||
|     )["Attributes"] | ||||
|     attrs.should.equal([{"Name": "attr2", "Value": "myvalue"}]) | ||||
							
								
								
									
										68
									
								
								tests/test_sdb/test_sdb_domains.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										68
									
								
								tests/test_sdb/test_sdb_domains.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,68 @@ | ||||
| import boto3 | ||||
| import pytest | ||||
| import sure  # noqa # pylint: disable=unused-import | ||||
| 
 | ||||
| from botocore.exceptions import ClientError | ||||
| from moto import mock_sdb | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| @pytest.mark.parametrize("name", ["", "a", "a#", "aaa#", "as@asdff", "asf'qwer"]) | ||||
| def test_create_domain_invalid(name): | ||||
|     # Error handling is always the same | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     with pytest.raises(ClientError) as exc: | ||||
|         sdb.create_domain(DomainName=name) | ||||
|     err = exc.value.response["Error"] | ||||
|     err["Code"].should.equal("InvalidParameterValue") | ||||
|     err["Message"].should.equal(f"Value ({name}) for parameter DomainName is invalid. ") | ||||
|     err.should.have.key("BoxUsage") | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| @pytest.mark.parametrize( | ||||
|     "name", ["abc", "ABc", "a00", "as-df", "jk_kl", "qw.rt", "asfljaejadslfsl"] | ||||
| ) | ||||
| def test_create_domain_valid(name): | ||||
|     # a-z, A-Z, 0-9, '_', '-', and '.' | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     sdb.create_domain(DomainName=name) | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| def test_create_domain_and_list(): | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     sdb.create_domain(DomainName="mydomain") | ||||
| 
 | ||||
|     all_domains = sdb.list_domains()["DomainNames"] | ||||
|     all_domains.should.equal(["mydomain"]) | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| def test_delete_domain(): | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     sdb.create_domain(DomainName="mydomain") | ||||
|     sdb.delete_domain(DomainName="mydomain") | ||||
| 
 | ||||
|     all_domains = sdb.list_domains() | ||||
|     all_domains.shouldnt.have.key("DomainNames") | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| def test_delete_domain_unknown(): | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     sdb.delete_domain(DomainName="unknown") | ||||
| 
 | ||||
|     all_domains = sdb.list_domains() | ||||
|     all_domains.shouldnt.have.key("DomainNames") | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| def test_delete_domain_invalid(): | ||||
|     sdb = boto3.client("sdb", region_name="eu-west-1") | ||||
|     with pytest.raises(ClientError) as exc: | ||||
|         sdb.delete_domain(DomainName="a") | ||||
|     err = exc.value.response["Error"] | ||||
|     err["Code"].should.equal("InvalidParameterValue") | ||||
|     err["Message"].should.equal(f"Value (a) for parameter DomainName is invalid. ") | ||||
|     err.should.have.key("BoxUsage") | ||||
							
								
								
									
										15
									
								
								tests/test_sdb/test_server.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										15
									
								
								tests/test_sdb/test_server.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,15 @@ | ||||
| """Test different server responses.""" | ||||
| import sure  # noqa # pylint: disable=unused-import | ||||
| 
 | ||||
| import moto.server as server | ||||
| from moto import mock_sdb | ||||
| 
 | ||||
| 
 | ||||
| @mock_sdb | ||||
| def test_sdb_list(): | ||||
|     backend = server.create_backend_app("sdb") | ||||
|     test_client = backend.test_client() | ||||
| 
 | ||||
|     resp = test_client.post("/", data={"Action": "ListDomains"}) | ||||
|     resp.status_code.should.equal(200) | ||||
|     str(resp.data).should.contain("ListDomainsResult") | ||||
		Loading…
	
	
			
			x
			
			
		
	
		Reference in New Issue
	
	Block a user