Feature: Cost Explorer (#5273)

moto/__init__.py

@@ -57,6 +57,7 @@ mock_batch_simple = lazy_load(
     backend="batch_simple_backends",
 )
 mock_budgets = lazy_load(".budgets", "mock_budgets")
+mock_ce = lazy_load(".ce", "mock_ce")
 mock_cloudformation = lazy_load(".cloudformation", "mock_cloudformation")
 mock_cloudfront = lazy_load(".cloudfront", "mock_cloudfront")
 mock_cloudtrail = lazy_load(".cloudtrail", "mock_cloudtrail")

moto/backend_index.py

@@ -13,6 +13,7 @@ backend_url_patterns = [
     ("autoscaling", re.compile("https?://autoscaling\\.(.+)\\.amazonaws\\.com")),
     ("batch", re.compile("https?://batch\\.(.+)\\.amazonaws.com")),
     ("budgets", re.compile("https?://budgets\\.amazonaws\\.com")),
+    ("ce", re.compile("https?://ce\\.(.+)\\.amazonaws\\.com")),
     ("cloudformation", re.compile("https?://cloudformation\\.(.+)\\.amazonaws\\.com")),
     ("cloudfront", re.compile("https?://cloudfront\\.amazonaws\\.com")),
     ("cloudtrail", re.compile("https?://cloudtrail\\.(.+)\\.amazonaws\\.com")),

moto/ce/__init__.py

@@ -0,0 +1,5 @@
+"""ce module initialization; sets value for base decorator."""
+from .models import ce_backends
+from ..core.models import base_decorator
+
+mock_ce = base_decorator(ce_backends)

moto/ce/exceptions.py

@@ -0,0 +1,9 @@
+"""Exceptions raised by the ce service."""
+from moto.core.exceptions import JsonRESTError
+
+
+class CostCategoryNotFound(JsonRESTError):
+    def __init__(self, ccd_id):
+        super().__init__(
+            "ResourceNotFoundException", f"No Cost Categories found with ID {ccd_id}"
+        )

moto/ce/models.py

@@ -0,0 +1,89 @@
+"""CostExplorerBackend class with methods for supported APIs."""
+
+from .exceptions import CostCategoryNotFound
+from moto.core import ACCOUNT_ID, BaseBackend, BaseModel
+from moto.core.utils import BackendDict
+from uuid import uuid4
+
+
+class CostCategoryDefinition(BaseModel):
+    def __init__(self, name, rule_version, rules, default_value, split_charge_rules):
+        self.name = name
+        self.rule_version = rule_version
+        self.rules = rules
+        self.default_value = default_value
+        self.split_charge_rules = split_charge_rules
+        self.arn = f"arn:aws:ce::{ACCOUNT_ID}:costcategory/{str(uuid4())}"
+
+    def update(self, rule_version, rules, default_value, split_charge_rules):
+        self.rule_version = rule_version
+        self.rules = rules
+        self.default_value = default_value
+        self.split_charge_rules = split_charge_rules
+
+    def to_json(self):
+        return {
+            "CostCategoryArn": self.arn,
+            "Name": self.name,
+            "RuleVersion": self.rule_version,
+            "Rules": self.rules,
+            "DefaultValue": self.default_value,
+            "SplitChargeRules": self.split_charge_rules,
+        }
+
+
+class CostExplorerBackend(BaseBackend):
+    """Implementation of CostExplorer APIs."""
+
+    def __init__(self, region_name, account_id):
+        super().__init__(region_name, account_id)
+        self.cost_categories = dict()
+
+    def create_cost_category_definition(
+        self,
+        name,
+        rule_version,
+        rules,
+        default_value,
+        split_charge_rules,
+    ):
+        """
+        The EffectiveOn and ResourceTags-parameters are not yet implemented
+        """
+        ccd = CostCategoryDefinition(
+            name, rule_version, rules, default_value, split_charge_rules
+        )
+        self.cost_categories[ccd.arn] = ccd
+        return ccd.arn, ""
+
+    def describe_cost_category_definition(self, cost_category_arn):
+        """
+        The EffectiveOn-parameter is not yet implemented
+        """
+        if cost_category_arn not in self.cost_categories:
+            ccd_id = cost_category_arn.split("/")[-1]
+            raise CostCategoryNotFound(ccd_id)
+        return self.cost_categories[cost_category_arn]
+
+    def delete_cost_category_definition(self, cost_category_arn):
+        """
+        The EffectiveOn-parameter is not yet implemented
+        """
+        self.cost_categories.pop(cost_category_arn, None)
+        return cost_category_arn, ""
+
+    def update_cost_category_definition(
+        self, cost_category_arn, rule_version, rules, default_value, split_charge_rules
+    ):
+        """
+        The EffectiveOn-parameter is not yet implemented
+        """
+        cost_category = self.describe_cost_category_definition(cost_category_arn)
+        cost_category.update(rule_version, rules, default_value, split_charge_rules)
+
+        return cost_category_arn, ""
+
+
+ce_backends = BackendDict(
+    CostExplorerBackend, "ce", use_boto3_regions=False, additional_regions=["global"]
+)

moto/ce/responses.py

@@ -0,0 +1,77 @@
+"""Handles incoming ce requests, invokes methods, returns responses."""
+import json
+
+from moto.core.responses import BaseResponse
+from .models import ce_backends
+
+
+class CostExplorerResponse(BaseResponse):
+    """Handler for CostExplorer requests and responses."""
+
+    @property
+    def ce_backend(self):
+        """Return backend instance specific for this region."""
+        return ce_backends["global"]
+
+    def create_cost_category_definition(self):
+        params = json.loads(self.body)
+        name = params.get("Name")
+        rule_version = params.get("RuleVersion")
+        rules = params.get("Rules")
+        default_value = params.get("DefaultValue")
+        split_charge_rules = params.get("SplitChargeRules")
+        (
+            cost_category_arn,
+            effective_start,
+        ) = self.ce_backend.create_cost_category_definition(
+            name=name,
+            rule_version=rule_version,
+            rules=rules,
+            default_value=default_value,
+            split_charge_rules=split_charge_rules,
+        )
+        return json.dumps(
+            dict(CostCategoryArn=cost_category_arn, EffectiveStart=effective_start)
+        )
+
+    def describe_cost_category_definition(self):
+        params = json.loads(self.body)
+        cost_category_arn = params.get("CostCategoryArn")
+        cost_category = self.ce_backend.describe_cost_category_definition(
+            cost_category_arn=cost_category_arn
+        )
+        return json.dumps(dict(CostCategory=cost_category.to_json()))
+
+    def delete_cost_category_definition(self):
+        params = json.loads(self.body)
+        cost_category_arn = params.get("CostCategoryArn")
+        (
+            cost_category_arn,
+            effective_end,
+        ) = self.ce_backend.delete_cost_category_definition(
+            cost_category_arn=cost_category_arn,
+        )
+        return json.dumps(
+            dict(CostCategoryArn=cost_category_arn, EffectiveEnd=effective_end)
+        )
+
+    def update_cost_category_definition(self):
+        params = json.loads(self.body)
+        cost_category_arn = params.get("CostCategoryArn")
+        rule_version = params.get("RuleVersion")
+        rules = params.get("Rules")
+        default_value = params.get("DefaultValue")
+        split_charge_rules = params.get("SplitChargeRules")
+        (
+            cost_category_arn,
+            effective_start,
+        ) = self.ce_backend.update_cost_category_definition(
+            cost_category_arn=cost_category_arn,
+            rule_version=rule_version,
+            rules=rules,
+            default_value=default_value,
+            split_charge_rules=split_charge_rules,
+        )
+        return json.dumps(
+            dict(CostCategoryArn=cost_category_arn, EffectiveStart=effective_start)
+        )

moto/ce/urls.py

@@ -0,0 +1,11 @@
+"""ce base URL and path."""
+from .responses import CostExplorerResponse
+
+url_bases = [
+    r"https?://ce\.(.+)\.amazonaws\.com",
+]
+
+
+url_paths = {
+    "{0}/$": CostExplorerResponse.dispatch,
+}

tests/terraformtests/terraform-tests.success.txt

@@ -15,6 +15,8 @@ autoscaling:
   - TestAccAutoScalingLaunchConfiguration_
 batch:
   - TestAccBatchJobDefinition
+ce:
+  - TestAccCECostCategory
 cloudtrail:
   - TestAccCloudTrailServiceAccount
 cloudwatch:

tests/test_ce/test_ce.py

@@ -0,0 +1,108 @@
+"""Unit tests for ce-supported APIs."""
+import boto3
+import pytest
+import sure  # noqa # pylint: disable=unused-import
+
+from botocore.exceptions import ClientError
+from moto import mock_ce
+from moto.core import ACCOUNT_ID
+
+# See our Development Tips on writing tests for hints on how to write good tests:
+# http://docs.getmoto.org/en/latest/docs/contributing/development_tips/tests.html
+
+
+@mock_ce
+def test_create_cost_category_definition():
+    client = boto3.client("ce", region_name="ap-southeast-1")
+    resp = client.create_cost_category_definition(
+        Name="ccd",
+        RuleVersion="CostCategoryExpression.v1",
+        Rules=[
+            {"Value": "v", "Rule": {"CostCategories": {"Key": "k", "Values": ["v"]}}}
+        ],
+    )
+    resp.should.have.key("CostCategoryArn").match(
+        f"arn:aws:ce::{ACCOUNT_ID}:costcategory/"
+    )
+
+
+@mock_ce
+def test_describe_cost_category_definition():
+    client = boto3.client("ce", region_name="us-east-2")
+    ccd_arn = client.create_cost_category_definition(
+        Name="ccd",
+        RuleVersion="CostCategoryExpression.v1",
+        Rules=[
+            {"Value": "v", "Rule": {"CostCategories": {"Key": "k", "Values": ["v"]}}}
+        ],
+    )["CostCategoryArn"]
+
+    resp = client.describe_cost_category_definition(CostCategoryArn=ccd_arn)[
+        "CostCategory"
+    ]
+    resp.should.have.key("Name").equals("ccd")
+    resp.should.have.key("CostCategoryArn").equals(ccd_arn)
+    resp.should.have.key("RuleVersion").equals("CostCategoryExpression.v1")
+    resp.should.have.key("Rules").length_of(1)
+    resp["Rules"][0].should.equal(
+        {"Value": "v", "Rule": {"CostCategories": {"Key": "k", "Values": ["v"]}}}
+    )
+
+
+@mock_ce
+def test_delete_cost_category_definition():
+    client = boto3.client("ce", region_name="ap-southeast-1")
+    ccd_arn = client.create_cost_category_definition(
+        Name="ccd",
+        RuleVersion="CostCategoryExpression.v1",
+        Rules=[
+            {"Value": "v", "Rule": {"CostCategories": {"Key": "k", "Values": ["v"]}}}
+        ],
+    )["CostCategoryArn"]
+    ccd_id = ccd_arn.split("/")[-1]
+
+    client.delete_cost_category_definition(CostCategoryArn=ccd_arn)
+
+    with pytest.raises(ClientError) as exc:
+        client.describe_cost_category_definition(CostCategoryArn=ccd_arn)
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("ResourceNotFoundException")
+    err["Message"].should.equal(f"No Cost Categories found with ID {ccd_id}")
+
+
+@mock_ce
+def test_update_cost_category_definition():
+    client = boto3.client("ce", region_name="us-east-2")
+    ccd_arn = client.create_cost_category_definition(
+        Name="ccd",
+        RuleVersion="CostCategoryExpression.v1",
+        Rules=[
+            {"Value": "v", "Rule": {"CostCategories": {"Key": "k", "Values": ["v"]}}}
+        ],
+    )["CostCategoryArn"]
+
+    client.update_cost_category_definition(
+        CostCategoryArn=ccd_arn,
+        RuleVersion="CostCategoryExpression.v1",
+        Rules=[
+            {"Value": "v", "Rule": {"CostCategories": {"Key": "k", "Values": ["v"]}}}
+        ],
+        SplitChargeRules=[{"Source": "s", "Targets": ["t1"], "Method": "EVEN"}],
+    )
+
+    resp = client.describe_cost_category_definition(CostCategoryArn=ccd_arn)[
+        "CostCategory"
+    ]
+    resp.should.have.key("Name").equals("ccd")
+    resp.should.have.key("CostCategoryArn").equals(ccd_arn)
+    resp.should.have.key("RuleVersion").equals("CostCategoryExpression.v1")
+
+    resp.should.have.key("Rules").length_of(1)
+    resp["Rules"][0].should.equal(
+        {"Value": "v", "Rule": {"CostCategories": {"Key": "k", "Values": ["v"]}}}
+    )
+
+    resp.should.have.key("SplitChargeRules").length_of(1)
+    resp["SplitChargeRules"][0].should.equal(
+        {"Source": "s", "Targets": ["t1"], "Method": "EVEN"}
+    )