5709656380

pending completion

Build # 5709656380

Build Type

push

github

Committed by

web-flow

Commit Message

Version bump (#1808)

Run Details

45512 of 59772 relevant lines covered (76.14%)

39312.43 hits per line

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

80.67

/dozer-api/src/rest/mod.rs

use std::sync::Arc;

// Exports
use crate::errors::ApiInitError;
use crate::rest::api_generator::health_route;
use crate::{
    auth::api::{auth_route, validate},
    CacheEndpoint,
};
use actix_cors::Cors;
use actix_web::dev::Server;
use actix_web::middleware::DefaultHeaders;
use actix_web::{
    body::MessageBody,
    dev::{Service, ServiceFactory, ServiceRequest, ServiceResponse},
    middleware::{Condition, Logger},
    web, App, HttpMessage, HttpServer,
};
use actix_web_httpauth::middleware::HttpAuthentication;
use dozer_types::{log::info, models::api_config::RestApiOptions};
use dozer_types::{
    models::api_security::ApiSecurity,
    serde::{self, Deserialize, Serialize},
};
use futures_util::Future;
use tracing_actix_web::TracingLogger;

mod api_generator;
mod rest_metric_middleware;

#[derive(Debug, Serialize, Deserialize, Eq, PartialEq, Clone)]
#[serde(crate = "self::serde")]
enum CorsOptions {
    Permissive,
    // origins, max_age
    Custom(Vec<String>, usize),
}

pub const DOZER_SERVER_NAME_HEADER: &str = "x-dozer-server-name";

#[derive(Clone)]
pub struct ApiServer {
    shutdown_timeout: u64,
    port: u16,
    cors: CorsOptions,
    security: Option<ApiSecurity>,
    host: String,
}

impl Default for ApiServer {
    fn default() -> Self {
        Self {
            shutdown_timeout: 0,
            port: 8080,
            cors: CorsOptions::Permissive,
            security: None,
            host: "0.0.0.0".to_owned(),
        }
    }
}

impl ApiServer {
    pub fn new(rest_config: RestApiOptions, security: Option<ApiSecurity>) -> Self {
        Self {
            shutdown_timeout: 0,
            port: rest_config.port as u16,
            cors: CorsOptions::Permissive,
            security,
            host: rest_config.host,
        }
    }
    fn get_cors(cors: CorsOptions) -> Cors {
        match cors {
            CorsOptions::Permissive => Cors::permissive(),
            CorsOptions::Custom(origins, max_age) => origins
                .into_iter()
                .fold(Cors::default(), |cors, origin| cors.allowed_origin(&origin))
                .max_age(max_age),
        }
    }

    fn create_app_entry(
        security: Option<ApiSecurity>,
        cors: CorsOptions,
        mut cache_endpoints: Vec<Arc<CacheEndpoint>>,
    ) -> App<
        impl ServiceFactory<
            ServiceRequest,
            Response = ServiceResponse<impl MessageBody>,
            Config = (),
            InitError = (),
            Error = actix_web::Error,
        >,
    > {
        let endpoint_paths: Vec<String> = cache_endpoints
            .iter()
            .map(|cache_endpoint| cache_endpoint.endpoint.path.clone())
            .collect();

        let mut app = App::new()
            .app_data(web::Data::new(endpoint_paths))
            .wrap(Logger::default())
            .wrap(TracingLogger::default())
            .wrap(DefaultHeaders::new().add((
                DOZER_SERVER_NAME_HEADER,
                gethostname::gethostname().to_string_lossy().into_owned(),
            )));

        let is_auth_configured = if let Some(api_security) = security {
            // Injecting API Security
            app = app.app_data(api_security);
            true
        } else {
            false
        };
        let auth_middleware =
            Condition::new(is_auth_configured, HttpAuthentication::bearer(validate));

        let cors_middleware = Self::get_cors(cors);

        //reverse sort cache endpoints by path length to ensure that the most specific path is matched first
        cache_endpoints.sort_by(|a, b| b.endpoint.path.len().cmp(&a.endpoint.path.len()));

        cache_endpoints
            .into_iter()
            .fold(app, |app, cache_endpoint| {
                let endpoint = &cache_endpoint.endpoint;
                let scope = &endpoint.path;
                app.service(
                    web::scope(scope)
                        .wrap(rest_metric_middleware::RestMetric)
                        // Inject cache_endpoint for generated functions
                        .wrap_fn(move |req, srv| {
                            req.extensions_mut().insert(cache_endpoint.clone());
                            srv.call(req)
                        })
                        .route("/count", web::post().to(api_generator::count))
                        .route("/query", web::post().to(api_generator::query))
                        .route("/phase", web::post().to(api_generator::get_phase))
                        .route("/oapi", web::post().to(api_generator::generate_oapi))
                        .route("/{id}", web::get().to(api_generator::get))
                        .route("/", web::get().to(api_generator::list))
                        .route("", web::get().to(api_generator::list)),
                )
            })
            // Attach token generation route
            .route("/auth/token", web::post().to(auth_route))
            // Attach health route
            .route("/health", web::get().to(health_route))
            .route("/", web::get().to(list_endpoint_paths))
            .route("", web::get().to(list_endpoint_paths))
            // Wrap Api Validator
            .wrap(auth_middleware)
            // Wrap CORS around api validator. Required to return the right headers.
            .wrap(cors_middleware)
    }

    pub fn run(
        self,
        cache_endpoints: Vec<Arc<CacheEndpoint>>,
        shutdown: impl Future<Output = ()> + Send + 'static,
    ) -> Result<Server, ApiInitError> {
        info!(
            "Starting Rest Api Server on http://{}:{} with security: {}",
            self.host,
            self.port,
            self.security
                .as_ref()
                .map_or("None".to_string(), |s| match s {
                    ApiSecurity::Jwt(_) => "JWT".to_string(),
                })
        );
        let cors = self.cors;
        let security = self.security;
        let address = format!("{}:{}", self.host, self.port);
        let server = HttpServer::new(move || {
            ApiServer::create_app_entry(security.clone(), cors.clone(), cache_endpoints.clone())
        })
        .bind(&address)
        .map_err(|e| ApiInitError::FailedToBindToAddress(address, e))?
        .disable_signals()
        .shutdown_timeout(self.shutdown_timeout)
        .run();

        let server_handle = server.handle();
        tokio::spawn(async move {
            shutdown.await;
            server_handle.stop(true).await;
        });

        Ok(server)
    }
}

async fn list_endpoint_paths(endpoints: web::Data<Vec<String>>) -> web::Json<Vec<String>> {
    web::Json(endpoints.get_ref().clone())
}

#[cfg(test)]
mod tests;

1	use std::sync::Arc;
2
3	// Exports
4	use crate::errors::ApiInitError;
5	use crate::rest::api_generator::health_route;
6	use crate::{
7	auth::api::{auth_route, validate},
8	CacheEndpoint,
9	};
10	use actix_cors::Cors;
11	use actix_web::dev::Server;
12	use actix_web::middleware::DefaultHeaders;
13	use actix_web::{
14	body::MessageBody,
15	dev::{Service, ServiceFactory, ServiceRequest, ServiceResponse},
16	middleware::{Condition, Logger},
17	web, App, HttpMessage, HttpServer,
18	};
19	use actix_web_httpauth::middleware::HttpAuthentication;
20	use dozer_types::{log::info, models::api_config::RestApiOptions};
21	use dozer_types::{
22	models::api_security::ApiSecurity,
23	serde::{self, Deserialize, Serialize},
24	};
25	use futures_util::Future;
26	use tracing_actix_web::TracingLogger;
27
28	mod api_generator;
29	mod rest_metric_middleware;
30		×
31	#[derive(Debug, Serialize, Deserialize, Eq, PartialEq, Clone)]	102✔
32	#[serde(crate = "self::serde")]
33	enum CorsOptions {
34	Permissive,
35	// origins, max_age
36	Custom(Vec<String>, usize),
37	}
38
39	pub const DOZER_SERVER_NAME_HEADER: &str = "x-dozer-server-name";
40		×
41	#[derive(Clone)]	×
42	pub struct ApiServer {
43	shutdown_timeout: u64,
44	port: u16,
45	cors: CorsOptions,
46	security: Option<ApiSecurity>,
47	host: String,
48	}
49
50	impl Default for ApiServer {	×
51	fn default() -> Self {	×
52	Self {	×
53	shutdown_timeout: 0,	×
54	port: 8080,	×
55	cors: CorsOptions::Permissive,	×
56	security: None,	×
57	host: "0.0.0.0".to_owned(),	×
58	}	×
59	}	×
60	}
61
62	impl ApiServer {	×
63	pub fn new(rest_config: RestApiOptions, security: Option<ApiSecurity>) -> Self {	9✔
64	Self {	9✔
65	shutdown_timeout: 0,	9✔
66	port: rest_config.port as u16,	9✔
67	cors: CorsOptions::Permissive,	9✔
68	security,	9✔
69	host: rest_config.host,	9✔
70	}	9✔
71	}	9✔
72	fn get_cors(cors: CorsOptions) -> Cors {	83✔
73	match cors {	83✔
74	CorsOptions::Permissive => Cors::permissive(),	83✔
75	CorsOptions::Custom(origins, max_age) => origins	×
76	.into_iter()	×
77	.fold(Cors::default(), \|cors, origin\| cors.allowed_origin(&origin))	×
78	.max_age(max_age),	×
79	}	×
80	}	83✔
81		×
82	fn create_app_entry(	83✔
83	security: Option<ApiSecurity>,	83✔
84	cors: CorsOptions,	83✔
85	mut cache_endpoints: Vec<Arc<CacheEndpoint>>,	83✔
86	) -> App<	83✔
87	impl ServiceFactory<	83✔
88	ServiceRequest,	83✔
89	Response = ServiceResponse<impl MessageBody>,	83✔
90	Config = (),	83✔
91	InitError = (),	83✔
92	Error = actix_web::Error,	83✔
93	>,	83✔
94	> {	83✔
95	let endpoint_paths: Vec<String> = cache_endpoints	83✔
96	.iter()	83✔
97	.map(\|cache_endpoint\| cache_endpoint.endpoint.path.clone())	84✔
98	.collect();	83✔
99		83✔
100	let mut app = App::new()	83✔
101	.app_data(web::Data::new(endpoint_paths))	83✔
102	.wrap(Logger::default())	83✔
103	.wrap(TracingLogger::default())	83✔
104	.wrap(DefaultHeaders::new().add((	83✔
105	DOZER_SERVER_NAME_HEADER,	83✔
106	gethostname::gethostname().to_string_lossy().into_owned(),	83✔
107	)));	83✔
108		×
109	let is_auth_configured = if let Some(api_security) = security {	83✔
110	// Injecting API Security	×
111	app = app.app_data(api_security);	4✔
112	true	4✔
113	} else {	×
114	false	79✔
115	};	×
116	let auth_middleware =	83✔
117	Condition::new(is_auth_configured, HttpAuthentication::bearer(validate));	83✔
118		83✔
119	let cors_middleware = Self::get_cors(cors);	83✔
120		83✔
121	//reverse sort cache endpoints by path length to ensure that the most specific path is matched first	83✔
122	cache_endpoints.sort_by(\|a, b\| b.endpoint.path.len().cmp(&a.endpoint.path.len()));	83✔
123		83✔
124	cache_endpoints	83✔
125	.into_iter()	83✔
126	.fold(app, \|app, cache_endpoint\| {	84✔
127	let endpoint = &cache_endpoint.endpoint;	84✔
128	let scope = &endpoint.path;	84✔
129	app.service(	84✔
130	web::scope(scope)	84✔
131	.wrap(rest_metric_middleware::RestMetric)	84✔
132	// Inject cache_endpoint for generated functions	84✔
133	.wrap_fn(move \|req, srv\| {	88✔
134	req.extensions_mut().insert(cache_endpoint.clone());	16✔
135	srv.call(req)	16✔
136	})	88✔
137	.route("/count", web::post().to(api_generator::count))	84✔
138	.route("/query", web::post().to(api_generator::query))	84✔
139	.route("/phase", web::post().to(api_generator::get_phase))	84✔
140	.route("/oapi", web::post().to(api_generator::generate_oapi))	84✔
141	.route("/{id}", web::get().to(api_generator::get))	84✔
142	.route("/", web::get().to(api_generator::list))	84✔
143	.route("", web::get().to(api_generator::list)),	84✔
144	)	84✔
145	})	84✔
146	// Attach token generation route	83✔
147	.route("/auth/token", web::post().to(auth_route))	83✔
148	// Attach health route	83✔
149	.route("/health", web::get().to(health_route))	83✔
150	.route("/", web::get().to(list_endpoint_paths))	83✔
151	.route("", web::get().to(list_endpoint_paths))	83✔
152	// Wrap Api Validator	83✔
153	.wrap(auth_middleware)	83✔
154	// Wrap CORS around api validator. Required to return the right headers.	83✔
155	.wrap(cors_middleware)	83✔
156	}	83✔
157		×
158	pub fn run(	6✔
159	self,	6✔
160	cache_endpoints: Vec<Arc<CacheEndpoint>>,	6✔
161	shutdown: impl Future<Output = ()> + Send + 'static,	6✔
162	) -> Result<Server, ApiInitError> {	6✔
163	info!(	6✔
164	"Starting Rest Api Server on http://{}:{} with security: {}",	6✔
165	self.host,	6✔
166	self.port,	6✔
167	self.security	6✔
168	.as_ref()	6✔
169	.map_or("None".to_string(), \|s\| match s {	6✔
170	ApiSecurity::Jwt(_) => "JWT".to_string(),	×
171	})	6✔
172	);	×
173	let cors = self.cors;	6✔
174	let security = self.security;	6✔
175	let address = format!("{}:{}", self.host, self.port);	6✔
176	let server = HttpServer::new(move \|\| {	48✔
177	ApiServer::create_app_entry(security.clone(), cors.clone(), cache_endpoints.clone())	48✔
178	})	48✔
179	.bind(&address)	6✔
180	.map_err(\|e\| ApiInitError::FailedToBindToAddress(address, e))?	6✔
181	.disable_signals()	6✔
182	.shutdown_timeout(self.shutdown_timeout)	6✔
183	.run();	6✔
184		6✔
185	let server_handle = server.handle();	6✔
186	tokio::spawn(async move {	6✔
187	shutdown.await;	6✔
188	server_handle.stop(true).await;	6✔
189	});	6✔
190		6✔
191	Ok(server)	6✔
192	}	6✔
193	}	×
194
195	async fn list_endpoint_paths(endpoints: web::Data<Vec<String>>) -> web::Json<Vec<String>> {	1✔
196	web::Json(endpoints.get_ref().clone())	1✔
197	}	1✔
198		×
199	#[cfg(test)]
200	mod tests;

getdozer / dozer / 5709656380

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