8666760397

Committed 12 Apr 2024 07:02PM UTC coverage: 56.289% (+0.2%) from 56.083%

Build # 8666760397

Build Type

Pull #968

github

Committed by

web-flow

Commit Message

Merge 8c0857084 into a6cdbab81

Pull Request Pull Request #968: Support for time bucketed fixed size

Run Details

58 of 86 new or added lines in 8 files covered. (67.44%)

6 existing lines in 5 files now uncovered.

3692 of 6559 relevant lines covered (56.29%)

102.51 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.25

/src/clients/aggregator_client.rs

use crate::{
    clients::{ClientConnExt, ClientError},
    entity::{task::ProvisionableTask, Aggregator},
    handler::Error,
};
use serde::{de::DeserializeOwned, Serialize};
use trillium_client::{Client, KnownHeaderName};
use url::Url;
pub mod api_types;
pub use api_types::{AggregatorApiConfig, TaskCreate, TaskIds, TaskResponse, TaskUploadMetrics};

const CONTENT_TYPE: &str = "application/vnd.janus.aggregator+json;version=0.1";

#[derive(Debug, Clone)]
pub struct AggregatorClient {
    client: Client,
    aggregator: Aggregator,
}

impl AsRef<Client> for AggregatorClient {
    fn as_ref(&self) -> &Client {
        &self.client
    }
}

impl AggregatorClient {
    pub fn new(client: Client, aggregator: Aggregator, bearer_token: &str) -> Self {
        let client = client
            .with_base(aggregator.api_url.clone())
            .with_default_header(
                KnownHeaderName::Authorization,
                format!("Bearer {bearer_token}"),
            )
            .with_default_header(KnownHeaderName::Accept, CONTENT_TYPE);

        Self { client, aggregator }
    }

    pub async fn get_config(
        client: Client,
        base_url: Url,
        token: &str,
    ) -> Result<AggregatorApiConfig, ClientError> {
        client
            .get(base_url)
            .with_request_header(KnownHeaderName::Authorization, format!("Bearer {token}"))
            .with_request_header(KnownHeaderName::Accept, CONTENT_TYPE)
            .success_or_client_error()
            .await?
            .response_json()
            .await
            .map_err(Into::into)
    }

    pub async fn get_task_ids(&self) -> Result<Vec<String>, ClientError> {
        let mut ids = vec![];
        let mut path = String::from("task_ids");
        loop {
            let TaskIds {
                task_ids,
                pagination_token,
            } = self.get(&path).await?;

            ids.extend(task_ids);

            match pagination_token {
                Some(pagination_token) => {
                    path = format!("task_ids?pagination_token={pagination_token}");
                }
                None => break Ok(ids),
            }
        }
    }

    pub async fn get_task(&self, task_id: &str) -> Result<TaskResponse, ClientError> {
        self.get(&format!("tasks/{task_id}")).await
    }

    pub async fn get_task_upload_metrics(
        &self,
        task_id: &str,
    ) -> Result<TaskUploadMetrics, ClientError> {
        self.get(&format!("tasks/{task_id}/metrics/uploads")).await
    }

    pub async fn delete_task(&self, task_id: &str) -> Result<(), ClientError> {
        self.delete(&format!("tasks/{task_id}")).await
    }

    pub async fn create_task(&self, task: &ProvisionableTask) -> Result<TaskResponse, Error> {
        let task_create = TaskCreate::build(&self.aggregator, task)?;
        self.post("tasks", &task_create).await.map_err(Into::into)
    }

    // private below here

    async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T, ClientError> {
        self.client
            .get(path)
            .success_or_client_error()
            .await?
            .response_json()
            .await
            .map_err(ClientError::from)
    }

    async fn post<T: DeserializeOwned>(
        &self,
        path: &str,
        body: &impl Serialize,
    ) -> Result<T, ClientError> {
        self.client
            .post(path)
            .with_json_body(body)?
            .with_request_header(KnownHeaderName::ContentType, CONTENT_TYPE)
            .success_or_client_error()
            .await?
            .response_json()
            .await
            .map_err(ClientError::from)
    }

    async fn delete(&self, path: &str) -> Result<(), ClientError> {
        let _ = self.client.delete(path).success_or_client_error().await?;
        Ok(())
    }
}

1	use crate::{
2	clients::{ClientConnExt, ClientError},
3	entity::{task::ProvisionableTask, Aggregator},
4	handler::Error,
5	};
6	use serde::{de::DeserializeOwned, Serialize};
7	use trillium_client::{Client, KnownHeaderName};
8	use url::Url;
9	pub mod api_types;
10	pub use api_types::{AggregatorApiConfig, TaskCreate, TaskIds, TaskResponse, TaskUploadMetrics};
11
12	const CONTENT_TYPE: &str = "application/vnd.janus.aggregator+json;version=0.1";
13
14	#[derive(Debug, Clone)]
15	pub struct AggregatorClient {
16	client: Client,
17	aggregator: Aggregator,
18	}
19
20	impl AsRef<Client> for AggregatorClient {
21	fn as_ref(&self) -> &Client {	×
22	&self.client	×
23	}	×
24	}
25
26	impl AggregatorClient {
27	pub fn new(client: Client, aggregator: Aggregator, bearer_token: &str) -> Self {	26✔
28	let client = client	26✔
29	.with_base(aggregator.api_url.clone())	26✔
30	.with_default_header(	26✔
31	KnownHeaderName::Authorization,	26✔
32	format!("Bearer {bearer_token}"),	26✔
33	)	26✔
34	.with_default_header(KnownHeaderName::Accept, CONTENT_TYPE);	26✔
35		26✔
36	Self { client, aggregator }	26✔
37	}	26✔
38
39	pub async fn get_config(	25✔
40	client: Client,	25✔
41	base_url: Url,	25✔
42	token: &str,	25✔
43	) -> Result<AggregatorApiConfig, ClientError> {	25✔
44	client	25✔
45	.get(base_url)	25✔
46	.with_request_header(KnownHeaderName::Authorization, format!("Bearer {token}"))	25✔
47	.with_request_header(KnownHeaderName::Accept, CONTENT_TYPE)	25✔
48	.success_or_client_error()	25✔
49	.await?	14✔
50	.response_json()	22✔
51	.await	×
52	.map_err(Into::into)	22✔
53	}	25✔
54
55	pub async fn get_task_ids(&self) -> Result<Vec<String>, ClientError> {	2✔
56	let mut ids = vec![];	2✔
57	let mut path = String::from("task_ids");	2✔
58	loop {
59	let TaskIds {
60	task_ids,	6✔
61	pagination_token,	6✔
62	} = self.get(&path).await?;	6✔
63
64	ids.extend(task_ids);	6✔
65		6✔
66	match pagination_token {	6✔
67	Some(pagination_token) => {	4✔
68	path = format!("task_ids?pagination_token={pagination_token}");	4✔
69	}	4✔
70	None => break Ok(ids),	2✔
71	}
72	}
73	}	2✔
74
75	pub async fn get_task(&self, task_id: &str) -> Result<TaskResponse, ClientError> {	5✔
76	self.get(&format!("tasks/{task_id}")).await	5✔
77	}	5✔
78
79	pub async fn get_task_upload_metrics(	3✔
80	&self,	3✔
81	task_id: &str,	3✔
82	) -> Result<TaskUploadMetrics, ClientError> {	3✔
83	self.get(&format!("tasks/{task_id}/metrics/uploads")).await	3✔
84	}	3✔
85
86	pub async fn delete_task(&self, task_id: &str) -> Result<(), ClientError> {	2✔
87	self.delete(&format!("tasks/{task_id}")).await	2✔
88	}	2✔
89
90	pub async fn create_task(&self, task: &ProvisionableTask) -> Result<TaskResponse, Error> {	14✔
91	let task_create = TaskCreate::build(&self.aggregator, task)?;	14✔
92	self.post("tasks", &task_create).await.map_err(Into::into)	19✔
93	}	14✔
94
95	// private below here
96
97	async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T, ClientError> {	14✔
98	self.client	14✔
99	.get(path)	14✔
100	.success_or_client_error()	14✔
101	.await?	11✔
102	.response_json()	14✔
UNCOV 103	.await	×
104	.map_err(ClientError::from)	14✔
105	}	14✔
106
107	async fn post<T: DeserializeOwned>(	14✔
108	&self,	14✔
109	path: &str,	14✔
110	body: &impl Serialize,	14✔
111	) -> Result<T, ClientError> {	14✔
112	self.client	14✔
113	.post(path)	14✔
114	.with_json_body(body)?	14✔
115	.with_request_header(KnownHeaderName::ContentType, CONTENT_TYPE)	14✔
116	.success_or_client_error()	14✔
117	.await?	18✔
118	.response_json()	14✔
119	.await	1✔
120	.map_err(ClientError::from)	14✔
121	}	14✔
122
123	async fn delete(&self, path: &str) -> Result<(), ClientError> {	2✔
124	let _ = self.client.delete(path).success_or_client_error().await?;	2✔
125	Ok(())	2✔
126	}	2✔
127	}

divviup / divviup-api / 8666760397

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous