CostExplorer - get_cost_and_usage() (#7385)
This commit is contained in:
Bert Blommers
GitHub
parent
6505971ecb
commit
3a8d3a3a08
@ -735,7 +735,7 @@
|
||||
|
||||
## ce
|
||||
<details>
|
||||
<summary>18% implemented</summary>
|
||||
<summary>21% implemented</summary>
|
||||
|
||||
- [ ] create_anomaly_monitor
|
||||
- [ ] create_anomaly_subscription
|
||||
@ -747,7 +747,7 @@
|
||||
- [ ] get_anomalies
|
||||
- [ ] get_anomaly_monitors
|
||||
- [ ] get_anomaly_subscriptions
|
||||
- [ ] get_cost_and_usage
|
||||
- [X] get_cost_and_usage
|
||||
- [ ] get_cost_and_usage_with_resources
|
||||
- [ ] get_cost_categories
|
||||
- [ ] get_cost_forecast
|
||||
|
||||
@ -38,7 +38,50 @@ ce
|
||||
- [ ] get_anomalies
|
||||
- [ ] get_anomaly_monitors
|
||||
- [ ] get_anomaly_subscriptions
|
||||
- [ ] get_cost_and_usage
|
||||
- [X] get_cost_and_usage
|
||||
|
||||
There is no validation yet on any of the input parameters.
|
||||
|
||||
Cost or usage is not tracked by Moto, so this call will return nothing by default.
|
||||
|
||||
You can use a dedicated API to override this, by configuring a queue of expected results.
|
||||
|
||||
A request to `get_cost_and_usage` will take the first result from that queue, and assign it to the provided parameters. Subsequent requests using the same parameters will return the same result. Other requests using different parameters will take the next result from the queue, or return an empty result if the queue is empty.
|
||||
|
||||
Configure this queue by making an HTTP request to `/moto-api/static/ce/cost-and-usage-results`. An example invocation looks like this:
|
||||
|
||||
.. sourcecode:: python
|
||||
|
||||
result = {
|
||||
"results": [
|
||||
{
|
||||
"ResultsByTime": [
|
||||
{
|
||||
"TimePeriod": {"Start": "2024-01-01", "End": "2024-01-02"},
|
||||
"Total": {
|
||||
"BlendedCost": {"Amount": "0.0101516483", "Unit": "USD"}
|
||||
},
|
||||
"Groups": [],
|
||||
"Estimated": False
|
||||
}
|
||||
],
|
||||
"DimensionValueAttributes": [{"Value": "v", "Attributes": {"a": "b"}}]
|
||||
},
|
||||
{
|
||||
...
|
||||
},
|
||||
]
|
||||
}
|
||||
resp = requests.post(
|
||||
"http://motoapi.amazonaws.com:5000/moto-api/static/ce/cost-and-usage-results",
|
||||
json=expected_results,
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
|
||||
ce = boto3.client("ce", region_name="us-east-1")
|
||||
resp = ce.get_cost_and_usage(...)
|
||||
|
||||
|
||||
- [ ] get_cost_and_usage_with_resources
|
||||
- [ ] get_cost_categories
|
||||
- [ ] get_cost_forecast
|
||||
|
||||
@ -74,6 +74,8 @@ class CostExplorerBackend(BaseBackend):
|
||||
def __init__(self, region_name: str, account_id: str):
|
||||
super().__init__(region_name, account_id)
|
||||
self.cost_categories: Dict[str, CostCategoryDefinition] = dict()
|
||||
self.cost_usage_results_queue: List[Dict[str, Any]] = []
|
||||
self.cost_usage_results: Dict[str, Dict[str, Any]] = {}
|
||||
self.tagger = TaggingService()
|
||||
|
||||
def create_cost_category_definition(
|
||||
@ -154,6 +156,57 @@ class CostExplorerBackend(BaseBackend):
|
||||
def untag_resource(self, resource_arn: str, tag_keys: List[str]) -> None:
|
||||
self.tagger.untag_resource_using_names(resource_arn, tag_keys)
|
||||
|
||||
def get_cost_and_usage(self, body: str) -> Dict[str, Any]:
|
||||
"""
|
||||
There is no validation yet on any of the input parameters.
|
||||
|
||||
Cost or usage is not tracked by Moto, so this call will return nothing by default.
|
||||
|
||||
You can use a dedicated API to override this, by configuring a queue of expected results.
|
||||
|
||||
A request to `get_cost_and_usage` will take the first result from that queue, and assign it to the provided parameters. Subsequent requests using the same parameters will return the same result. Other requests using different parameters will take the next result from the queue, or return an empty result if the queue is empty.
|
||||
|
||||
Configure this queue by making an HTTP request to `/moto-api/static/ce/cost-and-usage-results`. An example invocation looks like this:
|
||||
|
||||
.. sourcecode:: python
|
||||
|
||||
result = {
|
||||
"results": [
|
||||
{
|
||||
"ResultsByTime": [
|
||||
{
|
||||
"TimePeriod": {"Start": "2024-01-01", "End": "2024-01-02"},
|
||||
"Total": {
|
||||
"BlendedCost": {"Amount": "0.0101516483", "Unit": "USD"}
|
||||
},
|
||||
"Groups": [],
|
||||
"Estimated": False
|
||||
}
|
||||
],
|
||||
"DimensionValueAttributes": [{"Value": "v", "Attributes": {"a": "b"}}]
|
||||
},
|
||||
{
|
||||
...
|
||||
},
|
||||
]
|
||||
}
|
||||
resp = requests.post(
|
||||
"http://motoapi.amazonaws.com:5000/moto-api/static/ce/cost-and-usage-results",
|
||||
json=expected_results,
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
|
||||
ce = boto3.client("ce", region_name="us-east-1")
|
||||
resp = ce.get_cost_and_usage(...)
|
||||
"""
|
||||
default_result: Dict[str, Any] = {
|
||||
"ResultsByTime": [],
|
||||
"DimensionValueAttributes": [],
|
||||
}
|
||||
if body not in self.cost_usage_results and self.cost_usage_results_queue:
|
||||
self.cost_usage_results[body] = self.cost_usage_results_queue.pop(0)
|
||||
return self.cost_usage_results.get(body, default_result)
|
||||
|
||||
|
||||
ce_backends = BackendDict(
|
||||
CostExplorerBackend, "ce", use_boto3_regions=False, additional_regions=["global"]
|
||||
|
||||
@ -102,3 +102,7 @@ class CostExplorerResponse(BaseResponse):
|
||||
tag_names = params.get("ResourceTagKeys")
|
||||
self.ce_backend.untag_resource(resource_arn, tag_names)
|
||||
return json.dumps({})
|
||||
|
||||
def get_cost_and_usage(self) -> str:
|
||||
resp = self.ce_backend.get_cost_and_usage(self.body)
|
||||
return json.dumps(resp)
|
||||
|
||||
@ -42,6 +42,12 @@ class MotoAPIBackend(BaseBackend):
|
||||
results = QueryResults(rows=rows, column_info=column_info)
|
||||
backend.query_results_queue.append(results)
|
||||
|
||||
def set_ce_cost_usage(self, result: Dict[str, Any], account_id: str) -> None:
|
||||
from moto.ce.models import ce_backends
|
||||
|
||||
backend = ce_backends[account_id]["global"]
|
||||
backend.cost_usage_results_queue.append(result)
|
||||
|
||||
def set_sagemaker_result(
|
||||
self,
|
||||
body: str,
|
||||
|
||||
@ -168,6 +168,23 @@ class MotoAPIResponse(BaseResponse):
|
||||
)
|
||||
return 201, {}, ""
|
||||
|
||||
def set_ce_cost_usage_result(
|
||||
self,
|
||||
request: Any,
|
||||
full_url: str, # pylint: disable=unused-argument
|
||||
headers: Any,
|
||||
) -> TYPE_RESPONSE:
|
||||
from .models import moto_api_backend
|
||||
|
||||
request_body_size = int(headers["Content-Length"])
|
||||
body = request.environ["wsgi.input"].read(request_body_size).decode("utf-8")
|
||||
body = json.loads(body)
|
||||
account_id = body.get("account_id", DEFAULT_ACCOUNT_ID)
|
||||
|
||||
for result in body.get("results", []):
|
||||
moto_api_backend.set_ce_cost_usage(result=result, account_id=account_id)
|
||||
return 201, {}, ""
|
||||
|
||||
def set_sagemaker_result(
|
||||
self,
|
||||
request: Any,
|
||||
|
||||
@ -13,6 +13,7 @@ url_paths = {
|
||||
"{0}/moto-api/reset-auth": response_instance.reset_auth_response,
|
||||
"{0}/moto-api/seed": response_instance.seed,
|
||||
"{0}/moto-api/static/athena/query-results": response_instance.set_athena_result,
|
||||
"{0}/moto-api/static/ce/cost-and-usage-results": response_instance.set_ce_cost_usage_result,
|
||||
"{0}/moto-api/static/inspector2/findings-results": response_instance.set_inspector2_findings_result,
|
||||
"{0}/moto-api/static/sagemaker/endpoint-results": response_instance.set_sagemaker_result,
|
||||
"{0}/moto-api/static/rds-data/statement-results": response_instance.set_rds_data_result,
|
||||
|
||||
64
tests/test_ce/test_ce_cost_and_usage.py
Normal file
64
tests/test_ce/test_ce_cost_and_usage.py
Normal file
@ -0,0 +1,64 @@
|
||||
import boto3
|
||||
import requests
|
||||
|
||||
from moto import mock_aws, settings
|
||||
|
||||
# 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_aws
|
||||
def test_get_cost_and_usage():
|
||||
ce = boto3.client("ce", region_name="eu-west-1")
|
||||
|
||||
resp = ce.get_cost_and_usage(
|
||||
TimePeriod={"Start": "2024-01-01", "End": "2024-02-01"},
|
||||
Granularity="DAILY",
|
||||
Metrics=["UNBLENDEDCOST"],
|
||||
)
|
||||
|
||||
assert resp["DimensionValueAttributes"] == []
|
||||
assert resp["ResultsByTime"] == []
|
||||
|
||||
|
||||
@mock_aws
|
||||
def test_set_query_results():
|
||||
base_url = (
|
||||
settings.test_server_mode_endpoint()
|
||||
if settings.TEST_SERVER_MODE
|
||||
else "http://motoapi.amazonaws.com"
|
||||
)
|
||||
|
||||
cost_and_usage_result = {
|
||||
"results": [
|
||||
{
|
||||
"ResultsByTime": [
|
||||
{
|
||||
"TimePeriod": {"Start": "2024-01-01", "End": "2024-01-02"},
|
||||
"Total": {
|
||||
"BlendedCost": {"Amount": "0.0101516483", "Unit": "USD"}
|
||||
},
|
||||
"Groups": [],
|
||||
"Estimated": False,
|
||||
}
|
||||
],
|
||||
"DimensionValueAttributes": [{"Value": "v", "Attributes": {"a": "b"}}],
|
||||
}
|
||||
],
|
||||
}
|
||||
resp = requests.post(
|
||||
f"{base_url}/moto-api/static/ce/cost-and-usage-results",
|
||||
json=cost_and_usage_result,
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
|
||||
ce = boto3.client("ce", region_name="us-west-1")
|
||||
|
||||
resp = ce.get_cost_and_usage(
|
||||
TimePeriod={"Start": "2024-01-01", "End": "2024-02-01"},
|
||||
Granularity="DAILY",
|
||||
Metrics=["BLENDEDCOST", "AMORTIZEDCOST"],
|
||||
)
|
||||
|
||||
resp.pop("ResponseMetadata")
|
||||
assert resp == cost_and_usage_result["results"][0]
|
||||
Loading…
Reference in New Issue
Block a user