use crate::types::emr::*;
use crate::{AwsHttpClient, Result};
pub struct EmrOps<'a> {
pub(crate) client: &'a AwsHttpClient,
}
impl<'a> EmrOps<'a> {
pub(crate) fn new(client: &'a AwsHttpClient) -> Self {
Self { client }
}
fn base_url(&self) -> String {
#[cfg(any(test, feature = "test-support"))]
{
if let Some(ref base) = self.client.base_url {
return base.trim_end_matches('/').to_string();
}
}
"https://elasticmapreduce.{region}.amazonaws.com".replace("{region}", self.client.region())
}
#[allow(dead_code)]
pub(crate) async fn list_clusters(
&self,
body: &ListClustersInput,
) -> Result<ListClustersOutput> {
let url = format!("{}/", self.base_url(),);
let body_bytes =
serde_json::to_vec(body).map_err(|e| crate::AwsError::InvalidResponse {
message: format!("Failed to serialize list_clusters request: {e}"),
body: None,
})?;
let response = self
.client
.post_json(
&url,
"elasticmapreduce",
"ElasticMapReduce.ListClusters",
"1.1",
&body_bytes,
)
.await?;
let response = response.error_for_status("json").await?;
let response_bytes =
response
.bytes()
.await
.map_err(|e| crate::AwsError::InvalidResponse {
message: format!("Failed to read list_clusters response: {e}"),
body: None,
})?;
serde_json::from_slice(&response_bytes).map_err(|e| crate::AwsError::InvalidResponse {
message: format!("Failed to parse list_clusters response: {e}"),
body: Some(String::from_utf8_lossy(&response_bytes).to_string()),
})
}
#[allow(dead_code)]
pub(crate) async fn describe_cluster(
&self,
body: &DescribeClusterInput,
) -> Result<DescribeClusterOutput> {
let url = format!("{}/", self.base_url(),);
let body_bytes =
serde_json::to_vec(body).map_err(|e| crate::AwsError::InvalidResponse {
message: format!("Failed to serialize describe_cluster request: {e}"),
body: None,
})?;
let response = self
.client
.post_json(
&url,
"elasticmapreduce",
"ElasticMapReduce.DescribeCluster",
"1.1",
&body_bytes,
)
.await?;
let response = response.error_for_status("json").await?;
let response_bytes =
response
.bytes()
.await
.map_err(|e| crate::AwsError::InvalidResponse {
message: format!("Failed to read describe_cluster response: {e}"),
body: None,
})?;
serde_json::from_slice(&response_bytes).map_err(|e| crate::AwsError::InvalidResponse {
message: format!("Failed to parse describe_cluster response: {e}"),
body: Some(String::from_utf8_lossy(&response_bytes).to_string()),
})
}
#[allow(dead_code)]
pub(crate) async fn terminate_job_flows(&self, body: &TerminateJobFlowsInput) -> Result<()> {
let url = format!("{}/", self.base_url(),);
let body_bytes =
serde_json::to_vec(body).map_err(|e| crate::AwsError::InvalidResponse {
message: format!("Failed to serialize terminate_job_flows request: {e}"),
body: None,
})?;
let response = self
.client
.post_json(
&url,
"elasticmapreduce",
"ElasticMapReduce.TerminateJobFlows",
"1.1",
&body_bytes,
)
.await?;
response.error_for_status("json").await?;
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_list_clusters() {
let mut mock = crate::MockClient::new();
mock.expect_post("/")
.returning_json(serde_json::to_value(ListClustersOutput::fixture()).unwrap());
let client = crate::AwsHttpClient::from_mock(mock);
let ops = EmrOps::new(&client);
let body = ListClustersInput::fixture();
let result = ops.list_clusters(&body).await;
assert!(result.is_ok());
}
#[tokio::test]
async fn test_describe_cluster() {
let mut mock = crate::MockClient::new();
mock.expect_post("/")
.returning_json(serde_json::to_value(DescribeClusterOutput::fixture()).unwrap());
let client = crate::AwsHttpClient::from_mock(mock);
let ops = EmrOps::new(&client);
let body = DescribeClusterInput::fixture();
let result = ops.describe_cluster(&body).await;
assert!(result.is_ok());
}
#[tokio::test]
async fn test_terminate_job_flows() {
let mut mock = crate::MockClient::new();
mock.expect_post("/").returning_json(serde_json::json!({}));
let client = crate::AwsHttpClient::from_mock(mock);
let ops = EmrOps::new(&client);
let body = TerminateJobFlowsInput::fixture();
let result = ops.terminate_job_flows(&body).await;
assert!(result.is_ok());
}
}