6034126612

Committed 31 Aug 2023 07:03AM UTC coverage: 77.119%. First build

Build # 6034126612

Build Type

Pull #1945

github

Committed by

Jesse-Bakker

Commit Message

Write dozer.lock

Pull Request Pull Request #1945: Write dozer.lock

Run Details

86 of 86 new or added lines in 9 files covered. (100.0%)

49169 of 63757 relevant lines covered (77.12%)

69984.15 hits per line

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

23.39

/dozer-cli/src/cli/helper.rs

use crate::config_helper::combine_config;
use crate::errors::CliError;
use crate::errors::CliError::{ConfigurationFilePathNotProvided, FailedToFindConfigurationFiles};
use crate::errors::ConfigCombineError::CannotReadConfig;
use crate::errors::OrchestrationError;
use crate::simple::SimpleOrchestrator as Dozer;

use atty::Stream;
use dozer_cache::dozer_log::camino::Utf8PathBuf;
use dozer_tracing::LabelsAndProgress;
use dozer_types::models::config::default_cache_max_map_size;
use dozer_types::prettytable::{row, Table};
use dozer_types::serde_json;
use dozer_types::{models::config::Config, serde_yaml};
use handlebars::Handlebars;
use std::collections::BTreeMap;
use std::io::{self, Read};
use std::path::PathBuf;
use std::sync::Arc;
use tokio::runtime::Runtime;

pub async fn init_dozer(
    runtime: Arc<Runtime>,
    config_paths: Vec<String>,
    config_token: Option<String>,
    config_overrides: Vec<(String, serde_json::Value)>,
    ignore_pipe: bool,
    labels: LabelsAndProgress,
) -> Result<Dozer, CliError> {
    let mut config = load_config(config_paths, config_token, ignore_pipe).await?;

    config = apply_overrides(&config, config_overrides)?;

    let cache_max_map_size = config
        .cache_max_map_size
        .unwrap_or_else(default_cache_max_map_size);
    let page_size = page_size::get() as u64;
    config.cache_max_map_size = Some(cache_max_map_size / page_size * page_size);

    let base_directory = std::env::current_dir().map_err(CliError::Io)?;
    let base_directory =
        Utf8PathBuf::try_from(base_directory).map_err(|e| CliError::Io(e.into_io_error()))?;

    Ok(Dozer::new(base_directory, config, runtime, labels))
}

pub async fn list_sources(
    runtime: Arc<Runtime>,
    config_paths: Vec<String>,
    config_token: Option<String>,
    config_overrides: Vec<(String, serde_json::Value)>,
    ignore_pipe: bool,
    filter: Option<String>,
) -> Result<(), OrchestrationError> {
    let dozer = init_dozer(
        runtime,
        config_paths,
        config_token,
        config_overrides,
        ignore_pipe,
        Default::default(),
    )
    .await?;
    let connection_map = dozer.list_connectors()?;
    let mut table_parent = Table::new();
    for (connection_name, (tables, schemas)) in connection_map {
        let mut first_table_found = false;

        for (table, schema) in tables.into_iter().zip(schemas) {
            let name = table.schema.map_or(table.name.clone(), |schema_name| {
                format!("{schema_name}.{}", table.name)
            });

            if filter
                .as_ref()
                .map_or(true, |name_part| name.contains(name_part))
            {
                if !first_table_found {
                    table_parent.add_row(row!["Connection", "Table", "Columns"]);
                    first_table_found = true;
                }
                let schema_table = schema.schema.print();

                table_parent.add_row(row![connection_name, name, schema_table]);
            }
        }

        if first_table_found {
            table_parent.add_empty_row();
        }
    }
    table_parent.printstd();
    Ok(())
}

async fn load_config(
    config_url_or_paths: Vec<String>,
    config_token: Option<String>,
    ignore_pipe: bool,
) -> Result<Config, CliError> {
    let read_stdin = atty::isnt(Stream::Stdin) && !ignore_pipe;
    let first_config_path = config_url_or_paths.get(0);
    match first_config_path {
        None => Err(ConfigurationFilePathNotProvided),
        Some(path) => {
            if path.starts_with("https://") || path.starts_with("http://") {
                load_config_from_http_url(path, config_token).await
            } else {
                load_config_from_file(config_url_or_paths, read_stdin)
            }
        }
    }
}

async fn load_config_from_http_url(
    config_url: &str,
    config_token: Option<String>,
) -> Result<Config, CliError> {
    let client = reqwest::Client::new();
    let mut get_request = client.get(config_url);
    if let Some(token) = config_token {
        get_request = get_request.bearer_auth(token);
    }
    let response: reqwest::Response = get_request.send().await?.error_for_status()?;
    let contents = response.text().await?;
    parse_config(&contents)
}

pub fn load_config_from_file(
    config_path: Vec<String>,
    read_stdin: bool,
) -> Result<Config, CliError> {
    let stdin_path = PathBuf::from("<stdin>");
    let input = if read_stdin {
        let mut input = String::new();
        io::stdin()
            .read_to_string(&mut input)
            .map_err(|e| CannotReadConfig(stdin_path, e))?;
        Some(input)
    } else {
        None
    };

    let config_template = combine_config(config_path.clone(), input)?;
    match config_template {
        Some(template) => parse_config(&template),
        None => Err(FailedToFindConfigurationFiles(config_path.join(", "))),
    }
}

fn parse_config(config_template: &str) -> Result<Config, CliError> {
    let mut handlebars = Handlebars::new();
    handlebars
        .register_template_string("config", config_template)
        .map_err(|e| CliError::FailedToParseYaml(Box::new(e)))?;

    let mut data = BTreeMap::new();

    for (key, value) in std::env::vars() {
        data.insert(key, value);
    }

    let config_str = handlebars
        .render("config", &data)
        .map_err(|e| CliError::FailedToParseYaml(Box::new(e)))?;

    let config: Config = serde_yaml::from_str(&config_str)
        .map_err(|e: serde_yaml::Error| CliError::FailedToParseYaml(Box::new(e)))?;

    Ok(config)
}

/// Convert `config` to JSON, apply JSON pointer overrides, then convert back to `Config`.
fn apply_overrides(
    config: &Config,
    config_overrides: Vec<(String, serde_json::Value)>,
) -> Result<Config, CliError> {
    let mut config_json = serde_json::to_value(config).map_err(CliError::SerializeConfigToJson)?;

    for (pointer, value) in config_overrides {
        if let Some(pointee) = config_json.pointer_mut(&pointer) {
            *pointee = value;
        } else {
            return Err(CliError::MissingConfigOverride(pointer));
        }
    }

    // Directly convert `config_json` to `Config` fails, not sure why.
    let config_json_string =
        serde_json::to_string(&config_json).map_err(CliError::SerializeConfigToJson)?;
    let config: Config =
        serde_json::from_str(&config_json_string).map_err(CliError::DeserializeConfigFromJson)?;

    Ok(config)
}

pub const LOGO: &str = r"
.____   ___ __________ ____
|  _ \ / _ \__  / ____|  _ \
| | | | | | |/ /|  _| | |_) |
| |_| | |_| / /_| |___|  _ <
|____/ \___/____|_____|_| \_\
";

pub const DESCRIPTION: &str = r#"Open-source platform to build, publish and manage blazing-fast real-time data APIs in minutes. 

 If no sub commands are passed, dozer will bring up both app and api services.
"#;

#[cfg(test)]
mod tests {
    use dozer_types::models::{api_config::ApiConfig, api_security::ApiSecurity};

    use super::*;

    #[test]
    fn test_override_top_level() {
        let mut config = Config {
            app_name: "test_override_top_level".to_string(),
            ..Default::default()
        };
        config.sql = Some("sql1".to_string());
        let sql = "sql2".to_string();
        let config = apply_overrides(
            &config,
            vec![("/sql".to_string(), serde_json::to_value(&sql).unwrap())],
        )
        .unwrap();
        assert_eq!(config.sql.unwrap(), sql);
    }

    #[test]
    fn test_override_nested() {
        let mut config = Config {
            app_name: "test_override_nested".to_string(),
            ..Default::default()
        };
        config.api = Some(ApiConfig {
            api_security: Some(ApiSecurity::Jwt("secret1".to_string())),
            ..Default::default()
        });
        let api_security = ApiSecurity::Jwt("secret2".to_string());
        let config = apply_overrides(
            &config,
            vec![(
                "/api/api_security".to_string(),
                serde_json::to_value(&api_security).unwrap(),
            )],
        )
        .unwrap();
        assert_eq!(config.api.unwrap().api_security.unwrap(), api_security);
    }
}

1	use crate::config_helper::combine_config;
2	use crate::errors::CliError;
3	use crate::errors::CliError::{ConfigurationFilePathNotProvided, FailedToFindConfigurationFiles};
4	use crate::errors::ConfigCombineError::CannotReadConfig;
5	use crate::errors::OrchestrationError;
6	use crate::simple::SimpleOrchestrator as Dozer;
7
8	use atty::Stream;
9	use dozer_cache::dozer_log::camino::Utf8PathBuf;
10	use dozer_tracing::LabelsAndProgress;
11	use dozer_types::models::config::default_cache_max_map_size;
12	use dozer_types::prettytable::{row, Table};
13	use dozer_types::serde_json;
14	use dozer_types::{models::config::Config, serde_yaml};
15	use handlebars::Handlebars;
16	use std::collections::BTreeMap;
17	use std::io::{self, Read};
18	use std::path::PathBuf;
19	use std::sync::Arc;	×
20	use tokio::runtime::Runtime;	×
21		×
22	pub async fn init_dozer(	×
23	runtime: Arc<Runtime>,	×
24	config_paths: Vec<String>,	×
25	config_token: Option<String>,	×
26	config_overrides: Vec<(String, serde_json::Value)>,	×
27	ignore_pipe: bool,	×
28	labels: LabelsAndProgress,	×
29	) -> Result<Dozer, CliError> {	×
30	let mut config = load_config(config_paths, config_token, ignore_pipe).await?;	×
31		×
32	config = apply_overrides(&config, config_overrides)?;	×
33		×
34	let cache_max_map_size = config	×
35	.cache_max_map_size	×
36	.unwrap_or_else(default_cache_max_map_size);	×
37	let page_size = page_size::get() as u64;	×
38	config.cache_max_map_size = Some(cache_max_map_size / page_size * page_size);	×
39		×
40	let base_directory = std::env::current_dir().map_err(CliError::Io)?;	×
41	let base_directory =	×
42	Utf8PathBuf::try_from(base_directory).map_err(\|e\| CliError::Io(e.into_io_error()))?;	×
43		×
44	Ok(Dozer::new(base_directory, config, runtime, labels))	×
45	}	×
46		×
47	pub async fn list_sources(	×
48	runtime: Arc<Runtime>,	×
49	config_paths: Vec<String>,	×
50	config_token: Option<String>,	×
51	config_overrides: Vec<(String, serde_json::Value)>,	×
52	ignore_pipe: bool,	×
53	filter: Option<String>,	×
54	) -> Result<(), OrchestrationError> {	×
55	let dozer = init_dozer(	×
56	runtime,	×
57	config_paths,	×
58	config_token,	×
59	config_overrides,	×
60	ignore_pipe,	×
61	Default::default(),	×
62	)	×
63	.await?;	×
64	let connection_map = dozer.list_connectors()?;	×
65	let mut table_parent = Table::new();	×
66	for (connection_name, (tables, schemas)) in connection_map {	×
67	let mut first_table_found = false;	×
68		×
69	for (table, schema) in tables.into_iter().zip(schemas) {	×
70	let name = table.schema.map_or(table.name.clone(), \|schema_name\| {	×
71	format!("{schema_name}.{}", table.name)	×
72	});	×
73		×
74	if filter	×
75	.as_ref()	×
76	.map_or(true, \|name_part\| name.contains(name_part))	×
77	{	×
78	if !first_table_found {	×
79	table_parent.add_row(row!["Connection", "Table", "Columns"]);	×
80	first_table_found = true;	×
81	}	×
82	let schema_table = schema.schema.print();	×
83		×
84	table_parent.add_row(row![connection_name, name, schema_table]);	×
85	}	×
86	}	×
87		×
88	if first_table_found {	×
89	table_parent.add_empty_row();	×
90	}	×
91	}	×
92	table_parent.printstd();	×
93	Ok(())	×
94	}	×
95		×
96	async fn load_config(	×
97	config_url_or_paths: Vec<String>,	×
98	config_token: Option<String>,	×
99	ignore_pipe: bool,	×
100	) -> Result<Config, CliError> {	×
101	let read_stdin = atty::isnt(Stream::Stdin) && !ignore_pipe;	×
102	let first_config_path = config_url_or_paths.get(0);	×
103	match first_config_path {	×
104	None => Err(ConfigurationFilePathNotProvided),	×
105	Some(path) => {	×
106	if path.starts_with("https://") \|\| path.starts_with("http://") {	×
107	load_config_from_http_url(path, config_token).await	×
108	} else {	×
109	load_config_from_file(config_url_or_paths, read_stdin)	×
110	}	×
111	}	×
112	}	×
113	}	×
114		×
115	async fn load_config_from_http_url(	×
116	config_url: &str,	×
117	config_token: Option<String>,	×
118	) -> Result<Config, CliError> {	×
119	let client = reqwest::Client::new();	×
120	let mut get_request = client.get(config_url);	×
121	if let Some(token) = config_token {	×
122	get_request = get_request.bearer_auth(token);	×
123	}	×
124	let response: reqwest::Response = get_request.send().await?.error_for_status()?;	×
125	let contents = response.text().await?;	×
126	parse_config(&contents)	×
127	}	×
128		×
129	pub fn load_config_from_file(	×
130	config_path: Vec<String>,	×
131	read_stdin: bool,	×
132	) -> Result<Config, CliError> {	×
133	let stdin_path = PathBuf::from("<stdin>");	×
134	let input = if read_stdin {	×
135	let mut input = String::new();	×
136	io::stdin()	×
137	.read_to_string(&mut input)	×
138	.map_err(\|e\| CannotReadConfig(stdin_path, e))?;	×
139	Some(input)	×
140	} else {	×
141	None	×
142	};	×
143		×
144	let config_template = combine_config(config_path.clone(), input)?;	×
145	match config_template {	×
146	Some(template) => parse_config(&template),	×
147	None => Err(FailedToFindConfigurationFiles(config_path.join(", "))),	×
148	}	×
149	}	×
150		×
151	fn parse_config(config_template: &str) -> Result<Config, CliError> {	×
152	let mut handlebars = Handlebars::new();	×
153	handlebars	×
154	.register_template_string("config", config_template)	×
155	.map_err(\|e\| CliError::FailedToParseYaml(Box::new(e)))?;	×
156		×
157	let mut data = BTreeMap::new();	×
158		×
159	for (key, value) in std::env::vars() {	×
160	data.insert(key, value);	×
161	}	×
162		×
163	let config_str = handlebars	×
164	.render("config", &data)	×
165	.map_err(\|e\| CliError::FailedToParseYaml(Box::new(e)))?;	×
166
167	let config: Config = serde_yaml::from_str(&config_str)	×
168	.map_err(\|e: serde_yaml::Error\| CliError::FailedToParseYaml(Box::new(e)))?;	×
169		×
170	Ok(config)	×
171	}	×
172		×
173	/// Convert `config` to JSON, apply JSON pointer overrides, then convert back to `Config`.	×
174	fn apply_overrides(	2✔
175	config: &Config,	2✔
176	config_overrides: Vec<(String, serde_json::Value)>,	2✔
177	) -> Result<Config, CliError> {	2✔
178	let mut config_json = serde_json::to_value(config).map_err(CliError::SerializeConfigToJson)?;	2✔
179
180	for (pointer, value) in config_overrides {	4✔
181	if let Some(pointee) = config_json.pointer_mut(&pointer) {	2✔
182	*pointee = value;	2✔
183	} else {	2✔
184	return Err(CliError::MissingConfigOverride(pointer));	×
185	}	×
186	}	×
187		×
188	// Directly convert `config_json` to `Config` fails, not sure why.	×
189	let config_json_string =	2✔
190	serde_json::to_string(&config_json).map_err(CliError::SerializeConfigToJson)?;	2✔
191	let config: Config =	2✔
192	serde_json::from_str(&config_json_string).map_err(CliError::DeserializeConfigFromJson)?;	2✔
193
194	Ok(config)	2✔
195	}	2✔
196
197	pub const LOGO: &str = r"
198	.____ ___ __________ ____
199	\| _ \ / _ \__ / ____\| _ \
200	\| \| \| \| \| \| \|/ /\| _\| \| \|_) \|
201	\| \|_\| \| \|_\| / /_\| \|___\| _ <
202	\|____/ \___/____\|_____\|_\| \_\
203	";
204
205	pub const DESCRIPTION: &str = r#"Open-source platform to build, publish and manage blazing-fast real-time data APIs in minutes.
206
207	If no sub commands are passed, dozer will bring up both app and api services.
208	"#;
209		×
210	#[cfg(test)]	×
211	mod tests {	×
212	use dozer_types::models::{api_config::ApiConfig, api_security::ApiSecurity};	×
213		×
214	use super::*;	×
215		×
216	#[test]	1✔
217	fn test_override_top_level() {	1✔
218	let mut config = Config {	1✔
219	app_name: "test_override_top_level".to_string(),	1✔
220	..Default::default()	1✔
221	};	1✔
222	config.sql = Some("sql1".to_string());	1✔
223	let sql = "sql2".to_string();	1✔
224	let config = apply_overrides(	1✔
225	&config,	1✔
226	vec![("/sql".to_string(), serde_json::to_value(&sql).unwrap())],	1✔
227	)	1✔
228	.unwrap();	1✔
229	assert_eq!(config.sql.unwrap(), sql);	1✔
230	}	1✔
231		×
232	#[test]	1✔
233	fn test_override_nested() {	1✔
234	let mut config = Config {	1✔
235	app_name: "test_override_nested".to_string(),	1✔
236	..Default::default()	1✔
237	};	1✔
238	config.api = Some(ApiConfig {	1✔
239	api_security: Some(ApiSecurity::Jwt("secret1".to_string())),	1✔
240	..Default::default()	1✔
241	});	1✔
242	let api_security = ApiSecurity::Jwt("secret2".to_string());	1✔
243	let config = apply_overrides(	1✔
244	&config,	1✔
245	vec![(	1✔
246	"/api/api_security".to_string(),	1✔
247	serde_json::to_value(&api_security).unwrap(),	1✔
248	)],	1✔
249	)	1✔
250	.unwrap();	1✔
251	assert_eq!(config.api.unwrap().api_security.unwrap(), api_security);	1✔
252	}	1✔
253	}

getdozer / dozer / 6034126612

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