diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index 5ad269377..812c29ac7 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -4087,6 +4087,22 @@
- [ ] update_workteam
+## sdb
+
+50% implemented
+
+- [ ] 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
+
+
## secretsmanager
68% implemented
@@ -4804,7 +4820,6 @@
- sagemaker-runtime
- savingsplans
- schemas
-- sdb
- securityhub
- serverlessrepo
- service-quotas
diff --git a/docs/docs/services/sdb.rst b/docs/docs/services/sdb.rst
new file mode 100644
index 000000000..889090ea5
--- /dev/null
+++ b/docs/docs/services/sdb.rst
@@ -0,0 +1,52 @@
+.. _implementedservice_sdb:
+
+.. |start-h3| raw:: html
+
+
+
+.. |end-h3| raw:: html
+
+
+
+===
+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
+
diff --git a/moto/__init__.py b/moto/__init__.py
index 20d0f5784..f5f9ba92b 100644
--- a/moto/__init__.py
+++ b/moto/__init__.py
@@ -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():
diff --git a/moto/backend_index.py b/moto/backend_index.py
index 59157f58a..b1f3ce537 100644
--- a/moto/backend_index.py
+++ b/moto/backend_index.py
@@ -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")),
diff --git a/moto/sdb/__init__.py b/moto/sdb/__init__.py
new file mode 100644
index 000000000..3d1cdfd5d
--- /dev/null
+++ b/moto/sdb/__init__.py
@@ -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)
diff --git a/moto/sdb/exceptions.py b/moto/sdb/exceptions.py
new file mode 100644
index 000000000..ec588d2f3
--- /dev/null
+++ b/moto/sdb/exceptions.py
@@ -0,0 +1,45 @@
+"""Exceptions raised by the sdb service."""
+from moto.core.exceptions import RESTError
+
+
+SDB_ERROR = """
+
+
+
+ {{ error_type }}
+ {{ message }}
+ 0.0055590278
+
+
+ ba3a8c86-dc37-0a45-ef44-c6cf7876a62f
+"""
+
+
+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)
diff --git a/moto/sdb/models.py b/moto/sdb/models.py
new file mode 100644
index 000000000..1fa84860e
--- /dev/null
+++ b/moto/sdb/models.py
@@ -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)
diff --git a/moto/sdb/responses.py b/moto/sdb/responses.py
new file mode 100644
index 000000000..d96c6745f
--- /dev/null
+++ b/moto/sdb/responses.py
@@ -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 = """
+
+"""
+
+
+LIST_DOMAINS_TEMPLATE = """
+
+
+ {% for name in domain_names %}
+ {{ name }}
+ {% endfor %}
+ {{ next_token }}
+
+
+"""
+
+DELETE_DOMAIN_TEMPLATE = """
+
+
+ 64d9c3ac-ef19-2e3d-7a03-9ea46205eb71
+ 0.0055590278
+
+"""
+
+PUT_ATTRIBUTES_TEMPLATE = """
+
+"""
+
+GET_ATTRIBUTES_TEMPLATE = """
+
+ 1549581b-12b7-11e3-895e-1334aEXAMPLE
+
+
+{% for attribute in attributes %}
+
+ {{ attribute["name"] }}
+ {{ attribute["value"] }}
+
+{% endfor %}
+
+"""
diff --git a/moto/sdb/urls.py b/moto/sdb/urls.py
new file mode 100644
index 000000000..f83502bde
--- /dev/null
+++ b/moto/sdb/urls.py
@@ -0,0 +1,7 @@
+from .responses import SimpleDBResponse
+
+url_bases = [
+ r"https?://sdb\.(.+)\.amazonaws\.com",
+]
+
+url_paths = {"{0}/$": SimpleDBResponse.dispatch}
diff --git a/moto/server.py b/moto/server.py
index 28cd5d123..2e87ac4bc 100644
--- a/moto/server.py
+++ b/moto/server.py
@@ -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):
diff --git a/tests/test_sdb/__init__.py b/tests/test_sdb/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/test_sdb/test_sdb_attributes.py b/tests/test_sdb/test_sdb_attributes.py
new file mode 100644
index 000000000..3bae659d9
--- /dev/null
+++ b/tests/test_sdb/test_sdb_attributes.py
@@ -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"}])
diff --git a/tests/test_sdb/test_sdb_domains.py b/tests/test_sdb/test_sdb_domains.py
new file mode 100644
index 000000000..1f3c0f4e9
--- /dev/null
+++ b/tests/test_sdb/test_sdb_domains.py
@@ -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")
diff --git a/tests/test_sdb/test_server.py b/tests/test_sdb/test_server.py
new file mode 100644
index 000000000..99a51232f
--- /dev/null
+++ b/tests/test_sdb/test_server.py
@@ -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")