Added network functions

moto/managedblockchain/__init__.py

@@ -0,0 +1,9 @@
+from __future__ import unicode_literals
+from .models import managedblockchain_backends
+from ..core.models import base_decorator, deprecated_base_decorator
+
+managedblockchain_backend = managedblockchain_backends["us-east-1"]
+mock_managedblockchain = base_decorator(managedblockchain_backends)
+mock_managedblockchain_deprecated = deprecated_base_decorator(
+    managedblockchain_backends
+)

moto/managedblockchain/exceptions.py

@@ -0,0 +1,16 @@
+from __future__ import unicode_literals
+from moto.core.exceptions import RESTError
+
+
+class ManagedBlockchainClientError(RESTError):
+    code = 400
+
+
+class BadRequestException(ManagedBlockchainClientError):
+    def __init__(self, pretty_called_method, operation_error):
+        super(BadRequestException, self).__init__(
+            "BadRequestException",
+            "An error occurred (BadRequestException) when calling the {0} operation: {1}".format(
+                pretty_called_method, operation_error
+            ),
+        )

moto/managedblockchain/models.py

@@ -0,0 +1,151 @@
+from __future__ import unicode_literals
+
+import datetime
+
+from boto3 import Session
+
+from moto.core import BaseBackend, BaseModel
+
+from .exceptions import BadRequestException
+
+from .utils import get_network_id
+
+FRAMEWORKS = [
+    "HYPERLEDGER_FABRIC",
+]
+
+FRAMEWORKVERSIONS = [
+    "1.2",
+]
+
+EDITIONS = [
+    "STARTER",
+    "STANDARD",
+]
+
+
+class ManagedBlockchainNetwork(BaseModel):
+    def __init__(
+        self,
+        id,
+        name,
+        framework,
+        frameworkversion,
+        frameworkconfiguration,
+        voting_policy,
+        member_configuration,
+        region,
+        description=None,
+    ):
+        self.st = datetime.datetime.now(datetime.timezone.utc)
+        self.id = id
+        self.name = name
+        self.description = description
+        self.framework = framework
+        self.frameworkversion = frameworkversion
+        self.frameworkconfiguration = frameworkconfiguration
+        self.voting_policy = voting_policy
+        self.member_configuration = member_configuration
+        self.region = region
+
+    def to_dict(self):
+        frameworkattributes = {
+            "Fabric": {
+                "OrderingServiceEndpoint": "orderer.{0}.managedblockchain.{1}.amazonaws.com:30001".format(
+                    self.id, self.region
+                ),
+                "Edition": self.frameworkconfiguration["Fabric"]["Edition"],
+            }
+        }
+
+        vpcendpointname = "com.amazonaws.{0}.managedblockchain.{1}".format(
+            self.region, self.id
+        )
+        # Use iso_8601_datetime_with_milliseconds ?
+        d = {
+            "Id": self.id,
+            "Name": self.name,
+            "Framework": self.framework,
+            "FrameworkVersion": self.frameworkversion,
+            "FrameworkAttributes": frameworkattributes,
+            "VpcEndpointServiceName": vpcendpointname,
+            "VotingPolicy": self.voting_policy,
+            "Status": "AVAILABLE",
+            "CreationDate": self.st.strftime("%Y-%m-%dT%H:%M:%S.%f%z"),
+        }
+        if self.description is not None:
+            d["Description"] = self.description
+        return d
+
+
+class ManagedBlockchainBackend(BaseBackend):
+    def __init__(self, region_name):
+        self.networks = {}
+        self.region_name = region_name
+
+    def reset(self):
+        region_name = self.region_name
+        self.__dict__ = {}
+        self.__init__(region_name)
+
+    def create_network(
+        self,
+        json_body,
+    ):
+        name = json_body["Name"]
+        framework = json_body["Framework"]
+        frameworkversion = json_body["FrameworkVersion"]
+        frameworkconfiguration = json_body["FrameworkConfiguration"]
+        voting_policy = json_body["VotingPolicy"]
+        member_configuration = json_body["MemberConfiguration"]
+
+        # Check framework
+        if framework not in FRAMEWORKS:
+            raise BadRequestException("CreateNetwork", "Invalid request body")
+
+        # Check framework version
+        if frameworkversion not in FRAMEWORKVERSIONS:
+            raise BadRequestException(
+                "CreateNetwork",
+                "Invalid version {0} requested for framework HYPERLEDGER_FABRIC".format(
+                    frameworkversion
+                ),
+            )
+
+        # Check edition
+        if frameworkconfiguration["Fabric"]["Edition"] not in EDITIONS:
+            raise BadRequestException("CreateNetwork", "Invalid request body")
+
+        ## Generate network ID
+        network_id = get_network_id()
+
+        self.networks[network_id] = ManagedBlockchainNetwork(
+            id=network_id,
+            name=name,
+            framework=framework,
+            frameworkversion=frameworkversion,
+            frameworkconfiguration=frameworkconfiguration,
+            voting_policy=voting_policy,
+            member_configuration=member_configuration,
+            region=self.region_name,
+        )
+
+    def list_networks(self):
+        return self.networks.values()
+
+    def get_network(self, network_id):
+        return self.networks[network_id]
+
+
+
+managedblockchain_backends = {}
+for region in Session().get_available_regions("managedblockchain"):
+    managedblockchain_backends[region] = ManagedBlockchainBackend(region)
+for region in Session().get_available_regions(
+    "managedblockchain", partition_name="aws-us-gov"
+):
+    managedblockchain_backends[region] = ManagedBlockchainBackend(region)
+for region in Session().get_available_regions(
+    "managedblockchain", partition_name="aws-cn"
+):
+    managedblockchain_backends[region] = ManagedBlockchainBackend(region)

moto/managedblockchain/responses.py

@@ -0,0 +1,67 @@
+from __future__ import unicode_literals
+
+import json
+from six.moves.urllib.parse import urlparse, parse_qs
+
+from moto.core.responses import BaseResponse
+from .models import managedblockchain_backends
+from .utils import region_from_managedblckchain_url, networkid_from_managedblockchain_url
+
+
+class ManagedBlockchainResponse(BaseResponse):
+    def __init__(self, backend):
+        super(ManagedBlockchainResponse, self).__init__()
+        self.backend = backend
+
+    @classmethod
+    def network_response(clazz, request, full_url, headers):
+        region_name = region_from_managedblckchain_url(full_url)
+        response_instance = ManagedBlockchainResponse(managedblockchain_backends[region_name])
+        return response_instance._network_response(request, full_url, headers)
+
+    def _network_response(self, request, full_url, headers):
+        method = request.method
+        if hasattr(request, "body"):
+            body = request.body
+        else:
+            body = request.data
+        parsed_url = urlparse(full_url)
+        querystring = parse_qs(parsed_url.query, keep_blank_values=True)
+        if method == "GET":
+            return self._all_networks_response(request, full_url, headers)
+        elif method == "POST":
+            json_body = json.loads(body.decode("utf-8"))
+            return self._network_response_post(json_body, querystring, headers)
+
+    def _all_networks_response(self, request, full_url, headers):
+        mbcnetworks = self.backend.list_networks()
+        response = json.dumps(
+            {"Networks": [mbcnetwork.to_dict() for mbcnetwork in mbcnetworks]}
+        )
+        headers["content-type"] = "application/json"
+        return 200, headers, response
+
+    def _network_response_post(self, json_body, querystring, headers):
+        self.backend.create_network(json_body)
+        return 201, headers, ""
+
+    @classmethod
+    def networkid_response(clazz, request, full_url, headers):
+        region_name = region_from_managedblckchain_url(full_url)
+        response_instance = ManagedBlockchainResponse(managedblockchain_backends[region_name])
+        return response_instance._networkid_response(request, full_url, headers)
+
+    def _networkid_response(self, request, full_url, headers):
+        method = request.method
+
+        if method == "GET":
+            network_id = networkid_from_managedblockchain_url(full_url)
+            return self._networkid_response_get(network_id, headers)
+
+    def _networkid_response_get(self, network_id, headers):
+        mbcnetwork = self.backend.get_network(network_id)
+        response = json.dumps(
+            {"Network": mbcnetwork.to_dict()}
+        )
+        headers["content-type"] = "application/json"
+        return 200, headers, response

moto/managedblockchain/urls.py

@@ -0,0 +1,9 @@
+from __future__ import unicode_literals
+from .responses import ManagedBlockchainResponse
+
+url_bases = ["https?://managedblockchain.(.+).amazonaws.com"]
+
+url_paths = {
+    "{0}/networks$": ManagedBlockchainResponse.network_response,
+    "{0}/networks/(?P<networkid>[^/.]+)$": ManagedBlockchainResponse.networkid_response,
+}

moto/managedblockchain/utils.py

@@ -0,0 +1,23 @@
+import random
+import string
+
+from six.moves.urllib.parse import urlparse
+
+
+def region_from_managedblckchain_url(url):
+    domain = urlparse(url).netloc
+
+    if "." in domain:
+        return domain.split(".")[1]
+    else:
+        return "us-east-1"
+
+
+def networkid_from_managedblockchain_url(full_url):
+    return full_url.split("/")[-1]
+
+
+def get_network_id():
+    return "n-" + "".join(
+        random.choice(string.ascii_uppercase + string.digits) for _ in range(26)
+    )