6009657516

Committed 29 Aug 2023 08:13AM UTC coverage: 76.652% (-1.4%) from 78.07%

Build # 6009657516

Build Type

push

github

Committed by

web-flow

Commit Message

chore: Create unit tests workflow (#1910)

* chore: Update for Rust 1.72.0

Rust 1.72.0 has introduced a bunch of new lints. Here we fix them all.

`let ... else` finally gets formatted.

* chire: Create unit tests workflow

* Rename and remove useless steps

* remove env vars

* Add concurrency group

* Test unit workflow on 4 cores

* Add mysql service to unit tests

---------

Co-authored-by: chubei <914745487@qq.com>

Run Details

48982 of 63902 relevant lines covered (76.65%)

48394.25 hits per line

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

0.0

/dozer-cli/src/cli/init.rs

use crate::errors::{CliError, OrchestrationError};
use dozer_types::constants::{DEFAULT_LAMBDAS_DIRECTORY, DEFAULT_QUERIES_DIRECTORY};
use dozer_types::log::warn;
use dozer_types::models::config::{default_cache_dir, default_home_dir, get_cache_dir};
use dozer_types::{
    constants::DEFAULT_CONFIG_PATH,
    ingestion_types::{
        EthConfig, EthFilter, EthLogConfig, EthProviderConfig, MongodbConfig, MySQLConfig,
        S3Details, S3Storage, SnowflakeConfig,
    },
    log::info,
    models::{
        config::Config,
        connection::{Connection, ConnectionConfig, PostgresConfig},
    },
    serde_yaml,
};
use rustyline::history::DefaultHistory;
use rustyline::{
    completion::{Completer, Pair},
    Context,
};
use rustyline::{error::ReadlineError, Editor};
use rustyline_derive::{Helper, Highlighter, Hinter, Validator};
use std::path::{Path, PathBuf};

#[derive(Helper, Highlighter, Hinter, Validator)]
pub struct InitHelper {}

impl Completer for InitHelper {
    type Candidate = Pair;
    fn complete(
        &self,
        line: &str,
        _pos: usize,
        _ctx: &Context,
    ) -> rustyline::Result<(usize, Vec<Self::Candidate>)> {
        let line = format!("{line}_");
        let mut tokens = line.split_whitespace();
        let mut last_token = String::from(tokens.next_back().unwrap());
        last_token.pop();
        let candidates: Vec<String> = vec![
            "Postgres".to_owned(),
            "Ethereum".to_owned(),
            "Snowflake".to_owned(),
            "MySQL".to_owned(),
            "S3".to_owned(),
            "MongoDB".to_owned(),
        ];
        let mut match_pair: Vec<Pair> = candidates
            .iter()
            .filter_map(|f| {
                if f.to_lowercase().starts_with(&last_token.to_lowercase()) {
                    Some(Pair {
                        display: f.to_owned(),
                        replacement: f.to_owned(),
                    })
                } else {
                    None
                }
            })
            .collect();
        if match_pair.is_empty() {
            match_pair = vec![Pair {
                display: "Postgres".to_owned(),
                replacement: "Postgres".to_owned(),
            }]
        }
        Ok((line.len() - last_token.len() - 1, match_pair))
    }
}

pub fn generate_connection(connection_name: &str) -> Connection {
    match connection_name {
        "Snowflake" | "snowflake" | "S" | "s" => {
            let snowflake_config = SnowflakeConfig {
                server: "<account_name>.<region_id>.snowflakecomputing.com".to_owned(),
                port: "443".to_owned(),
                user: "bob".to_owned(),
                password: "password".to_owned(),
                database: "database".to_owned(),
                schema: "schema".to_owned(),
                warehouse: "warehouse".to_owned(),
                driver: Some("SnowflakeDSIIDriver".to_owned()),
                role: "role".to_owned(),
            };
            let connection: Connection = Connection {
                name: "snowflake".to_owned(),
                config: Some(ConnectionConfig::Snowflake(snowflake_config)),
            };
            connection
        }
        "Ethereum" | "ethereum" | "E" | "e" => {
            let eth_filter = EthFilter {
                from_block: Some(0),
                to_block: None,
                addresses: vec![],
                topics: vec![],
            };
            let ethereum_config = EthConfig {
                provider: Some(EthProviderConfig::Log(EthLogConfig {
                    wss_url: "wss://link".to_owned(),
                    filter: Some(eth_filter),
                    contracts: vec![],
                })),
            };
            let connection: Connection = Connection {
                name: "ethereum".to_owned(),
                config: Some(ConnectionConfig::Ethereum(ethereum_config)),
            };
            connection
        }
        "MySQL" | "MYSQL" | "mysql" | "Mysql" | "My" => {
            let mysql_config = MySQLConfig {
                url: "mysql://<user>:<password>@localhost:3306/<database>".to_owned(),
                server_id: Some((1).to_owned()),
            };
            let connection: Connection = Connection {
                name: "mysql".to_owned(),
                config: Some(ConnectionConfig::MySQL(mysql_config)),
            };
            connection
        }
        "S3" | "s3" => {
            let s3_details = S3Details {
                access_key_id: "<your_access_key_id>".to_owned(),
                secret_access_key: "<your_secret_access_key>".to_owned(),
                region: "<your_region>".to_owned(),
                bucket_name: "<your_bucket_name>".to_owned(),
            };
            let s3_config = S3Storage {
                details: Some(s3_details),
                tables: vec![],
            };
            let connection: Connection = Connection {
                name: "s3".to_owned(),
                config: Some(ConnectionConfig::S3Storage(s3_config)),
            };
            connection
        }
        "MongoDB" | "mongodb" | "MONGODB" | "Mongodb" | "Mo" | "MO" => {
            let mongo_config = MongodbConfig {
                connection_string:
                    "mongodb://<username>:<password>@localhost:27017/<database_name>".to_owned(),
            };
            let connection: Connection = Connection {
                name: "mongodb".to_owned(),
                config: Some(ConnectionConfig::MongoDB(mongo_config)),
            };
            connection
        }
        _ => {
            let postgres_config = PostgresConfig {
                user: Some("postgres".to_owned()),
                password: Some("postgres".to_owned()),
                host: Some("localhost".to_owned()),
                port: Some(5432),
                database: Some("users".to_owned()),
                sslmode: None,
                connection_url: None,
            };
            let connection: Connection = Connection {
                name: "postgres".to_owned(),
                config: Some(ConnectionConfig::Postgres(postgres_config)),
            };
            connection
        }
    }
}
type Question = (
    String,
    Box<dyn Fn((String, &mut Config)) -> Result<(), OrchestrationError>>,
);
pub fn generate_config_repl() -> Result<(), OrchestrationError> {
    let mut rl = Editor::<InitHelper, DefaultHistory>::new()
        .map_err(|e| OrchestrationError::CliError(CliError::ReadlineError(e)))?;
    rl.set_helper(Some(InitHelper {}));
    let mut default_config = Config::default();
    let default_app_name = "quick-start-app";
    let questions: Vec<Question> = vec![
        (
            format!("question: App name ({:}): ", default_app_name),
            Box::new(move |(app_name, config)| {
                let app_name = app_name.trim();
                if app_name.is_empty() {
                    config.app_name = default_app_name.to_string();
                } else {
                    config.app_name = app_name.to_string();
                }
                Ok(())
            }),
        ),
        (
            format!("question: Home directory ({:}): ", default_home_dir()),
            Box::new(move |(home_dir, config)| {
                if home_dir.is_empty() {
                    config.home_dir = default_home_dir();
                    config.cache_dir = default_cache_dir();
                } else {
                    config.home_dir = home_dir;
                    config.cache_dir = get_cache_dir(&config.home_dir);
                }
                Ok(())
            }),
        ),
        (
            "question: Connection Type - one of: [P]ostgres, [E]thereum, [S]nowflake, [My]SQL, [S3]Storage, [Mo]ngoDB: "
                .to_string(),
            Box::new(move |(connection, config)| {
                let sample_connection = generate_connection(&connection);
                config.connections.push(sample_connection);

                Ok(())
            }),
        ),
        (
            format!("question: Config path ({:}): ", DEFAULT_CONFIG_PATH),
            Box::new(move |(yaml_path, config)| {
                let mut yaml_path = yaml_path.trim();
                if yaml_path.is_empty() {
                    yaml_path = DEFAULT_CONFIG_PATH;
                }
                let f = std::fs::OpenOptions::new()
                    .create(true)
                    .write(true)
                    .open(yaml_path)
                    .map_err(|e| {
                        OrchestrationError::CliError(CliError::FileSystem(
                            yaml_path.to_string().into(),
                            e,
                        ))
                    })?;
                serde_yaml::to_writer(f, &config)
                    .map_err(OrchestrationError::FailedToWriteConfigYaml)?;

                info!("Generating workspace: \n\
                \n- {} (main configuration)\n- ./queries (folder for sql queries)\n- ./lambdas (folder for lambda functions)
                \n• More details about our config: https://getdozer.io/docs/reference/configuration/introduction\
                \n• Connector & Sources: https://getdozer.io/docs/reference/configuration/connectors\
                \n• Endpoints: https://getdozer.io/docs/reference/configuration/endpoints/",
                   yaml_path.to_owned());

                let path = PathBuf::from(yaml_path);
                if let Some(dir) = path.parent() {
                    let queries_path = Path::new(dir).join(DEFAULT_QUERIES_DIRECTORY);
                    if let Err(_e) = std::fs::create_dir(queries_path) {
                        warn!("Cannot create queries directory");
                    }

                    let lambdas_path = Path::new(dir).join(DEFAULT_LAMBDAS_DIRECTORY);
                    if let Err(_e) = std::fs::create_dir(lambdas_path) {
                        warn!("Cannot create lambdas directory");
                    }
                }

                Ok(())
            }),
        ),
    ];
    let result = questions.iter().try_for_each(|(question, func)| {
        let readline = rl.readline(question);
        match readline {
            Ok(input) => func((input, &mut default_config)),
            Err(err) => Err(OrchestrationError::CliError(CliError::ReadlineError(err))),
        }
    });

    match result {
        Ok(_) => Ok(()),
        Err(e) => match e {
            OrchestrationError::CliError(CliError::ReadlineError(ReadlineError::Interrupted)) => {
                info!("Exiting..");
                Ok(())
            }
            OrchestrationError::CliError(CliError::ReadlineError(ReadlineError::Eof)) => {
                info!("CTRL-D - exiting...");
                Ok(())
            }
            _ => Err(e),
        },
    }
}

1	use crate::errors::{CliError, OrchestrationError};
2	use dozer_types::constants::{DEFAULT_LAMBDAS_DIRECTORY, DEFAULT_QUERIES_DIRECTORY};
3	use dozer_types::log::warn;
4	use dozer_types::models::config::{default_cache_dir, default_home_dir, get_cache_dir};
5	use dozer_types::{
6	constants::DEFAULT_CONFIG_PATH,
7	ingestion_types::{
8	EthConfig, EthFilter, EthLogConfig, EthProviderConfig, MongodbConfig, MySQLConfig,
9	S3Details, S3Storage, SnowflakeConfig,
10	},
11	log::info,
12	models::{
13	config::Config,
14	connection::{Connection, ConnectionConfig, PostgresConfig},
15	},
16	serde_yaml,
17	};
18	use rustyline::history::DefaultHistory;
19	use rustyline::{
20	completion::{Completer, Pair},
21	Context,
22	};
23	use rustyline::{error::ReadlineError, Editor};
24	use rustyline_derive::{Helper, Highlighter, Hinter, Validator};
25	use std::path::{Path, PathBuf};
26
27	#[derive(Helper, Highlighter, Hinter, Validator)]
28	pub struct InitHelper {}
29		×
30	impl Completer for InitHelper {	×
31	type Candidate = Pair;	×
32	fn complete(	×
33	&self,	×
34	line: &str,	×
35	_pos: usize,	×
36	_ctx: &Context,	×
37	) -> rustyline::Result<(usize, Vec<Self::Candidate>)> {	×
38	let line = format!("{line}_");	×
39	let mut tokens = line.split_whitespace();	×
40	let mut last_token = String::from(tokens.next_back().unwrap());	×
41	last_token.pop();	×
42	let candidates: Vec<String> = vec![	×
43	"Postgres".to_owned(),	×
44	"Ethereum".to_owned(),	×
45	"Snowflake".to_owned(),	×
46	"MySQL".to_owned(),	×
47	"S3".to_owned(),	×
48	"MongoDB".to_owned(),	×
49	];	×
50	let mut match_pair: Vec<Pair> = candidates	×
51	.iter()	×
52	.filter_map(\|f\| {	×
53	if f.to_lowercase().starts_with(&last_token.to_lowercase()) {	×
54	Some(Pair {	×
55	display: f.to_owned(),	×
56	replacement: f.to_owned(),	×
57	})	×
58	} else {	×
59	None	×
60	}	×
61	})	×
62	.collect();	×
63	if match_pair.is_empty() {	×
64	match_pair = vec![Pair {	×
65	display: "Postgres".to_owned(),	×
66	replacement: "Postgres".to_owned(),	×
67	}]	×
68	}	×
69	Ok((line.len() - last_token.len() - 1, match_pair))	×
70	}	×
71	}	×
72		×
73	pub fn generate_connection(connection_name: &str) -> Connection {	×
74	match connection_name {	×
75	"Snowflake" \| "snowflake" \| "S" \| "s" => {	×
76	let snowflake_config = SnowflakeConfig {	×
77	server: "<account_name>.<region_id>.snowflakecomputing.com".to_owned(),	×
78	port: "443".to_owned(),	×
79	user: "bob".to_owned(),	×
80	password: "password".to_owned(),	×
81	database: "database".to_owned(),	×
82	schema: "schema".to_owned(),	×
83	warehouse: "warehouse".to_owned(),	×
84	driver: Some("SnowflakeDSIIDriver".to_owned()),	×
85	role: "role".to_owned(),	×
86	};	×
87	let connection: Connection = Connection {	×
88	name: "snowflake".to_owned(),	×
89	config: Some(ConnectionConfig::Snowflake(snowflake_config)),	×
90	};	×
91	connection	×
92	}	×
93	"Ethereum" \| "ethereum" \| "E" \| "e" => {	×
94	let eth_filter = EthFilter {	×
95	from_block: Some(0),	×
96	to_block: None,	×
97	addresses: vec![],	×
98	topics: vec![],	×
99	};	×
100	let ethereum_config = EthConfig {	×
101	provider: Some(EthProviderConfig::Log(EthLogConfig {	×
102	wss_url: "wss://link".to_owned(),	×
103	filter: Some(eth_filter),	×
104	contracts: vec![],	×
105	})),	×
106	};	×
107	let connection: Connection = Connection {	×
108	name: "ethereum".to_owned(),	×
109	config: Some(ConnectionConfig::Ethereum(ethereum_config)),	×
110	};	×
111	connection	×
112	}	×
113	"MySQL" \| "MYSQL" \| "mysql" \| "Mysql" \| "My" => {	×
114	let mysql_config = MySQLConfig {	×
115	url: "mysql://<user>:<password>@localhost:3306/<database>".to_owned(),	×
116	server_id: Some((1).to_owned()),	×
117	};	×
118	let connection: Connection = Connection {	×
119	name: "mysql".to_owned(),	×
120	config: Some(ConnectionConfig::MySQL(mysql_config)),	×
121	};	×
122	connection	×
123	}
124	"S3" \| "s3" => {	×
125	let s3_details = S3Details {	×
126	access_key_id: "<your_access_key_id>".to_owned(),	×
127	secret_access_key: "<your_secret_access_key>".to_owned(),	×
128	region: "<your_region>".to_owned(),	×
129	bucket_name: "<your_bucket_name>".to_owned(),	×
130	};	×
131	let s3_config = S3Storage {	×
132	details: Some(s3_details),	×
133	tables: vec![],	×
134	};	×
135	let connection: Connection = Connection {	×
136	name: "s3".to_owned(),	×
137	config: Some(ConnectionConfig::S3Storage(s3_config)),	×
138	};	×
139	connection	×
140	}	×
141	"MongoDB" \| "mongodb" \| "MONGODB" \| "Mongodb" \| "Mo" \| "MO" => {	×
142	let mongo_config = MongodbConfig {	×
143	connection_string:	×
144	"mongodb://<username>:<password>@localhost:27017/<database_name>".to_owned(),	×
145	};	×
146	let connection: Connection = Connection {	×
147	name: "mongodb".to_owned(),	×
148	config: Some(ConnectionConfig::MongoDB(mongo_config)),	×
149	};	×
150	connection	×
151	}	×
152	_ => {	×
153	let postgres_config = PostgresConfig {	×
154	user: Some("postgres".to_owned()),	×
155	password: Some("postgres".to_owned()),	×
156	host: Some("localhost".to_owned()),	×
157	port: Some(5432),	×
158	database: Some("users".to_owned()),	×
159	sslmode: None,	×
160	connection_url: None,	×
161	};	×
162	let connection: Connection = Connection {	×
163	name: "postgres".to_owned(),	×
164	config: Some(ConnectionConfig::Postgres(postgres_config)),	×
165	};	×
166	connection	×
167	}	×
168	}	×
169	}	×
170	type Question = (	×
171	String,	×
172	Box<dyn Fn((String, &mut Config)) -> Result<(), OrchestrationError>>,	×
173	);	×
174	pub fn generate_config_repl() -> Result<(), OrchestrationError> {	×
175	let mut rl = Editor::<InitHelper, DefaultHistory>::new()	×
176	.map_err(\|e\| OrchestrationError::CliError(CliError::ReadlineError(e)))?;	×
177	rl.set_helper(Some(InitHelper {}));	×
178	let mut default_config = Config::default();	×
179	let default_app_name = "quick-start-app";	×
180	let questions: Vec<Question> = vec![	×
181	(	×
182	format!("question: App name ({:}): ", default_app_name),	×
183	Box::new(move \|(app_name, config)\| {	×
184	let app_name = app_name.trim();	×
185	if app_name.is_empty() {	×
186	config.app_name = default_app_name.to_string();	×
187	} else {	×
188	config.app_name = app_name.to_string();	×
189	}	×
190	Ok(())	×
191	}),	×
192	),	×
193	(	×
194	format!("question: Home directory ({:}): ", default_home_dir()),	×
195	Box::new(move \|(home_dir, config)\| {	×
196	if home_dir.is_empty() {	×
197	config.home_dir = default_home_dir();	×
198	config.cache_dir = default_cache_dir();	×
199	} else {	×
200	config.home_dir = home_dir;	×
201	config.cache_dir = get_cache_dir(&config.home_dir);	×
202	}	×
203	Ok(())	×
204	}),	×
205	),	×
206	(	×
207	"question: Connection Type - one of: [P]ostgres, [E]thereum, [S]nowflake, [My]SQL, [S3]Storage, [Mo]ngoDB: "	×
208	.to_string(),	×
209	Box::new(move \|(connection, config)\| {	×
210	let sample_connection = generate_connection(&connection);	×
211	config.connections.push(sample_connection);	×
212		×
213	Ok(())	×
214	}),	×
215	),	×
216	(	×
217	format!("question: Config path ({:}): ", DEFAULT_CONFIG_PATH),	×
218	Box::new(move \|(yaml_path, config)\| {	×
219	let mut yaml_path = yaml_path.trim();	×
220	if yaml_path.is_empty() {	×
221	yaml_path = DEFAULT_CONFIG_PATH;	×
222	}	×
223	let f = std::fs::OpenOptions::new()	×
224	.create(true)	×
225	.write(true)	×
226	.open(yaml_path)	×
227	.map_err(\|e\| {	×
228	OrchestrationError::CliError(CliError::FileSystem(	×
229	yaml_path.to_string().into(),	×
230	e,	×
231	))	×
232	})?;	×
233	serde_yaml::to_writer(f, &config)	×
234	.map_err(OrchestrationError::FailedToWriteConfigYaml)?;	×
235
236	info!("Generating workspace: \n\	×
237	\n- {} (main configuration)\n- ./queries (folder for sql queries)\n- ./lambdas (folder for lambda functions)	×
238	\n• More details about our config: https://getdozer.io/docs/reference/configuration/introduction\	×
239	\n• Connector & Sources: https://getdozer.io/docs/reference/configuration/connectors\	×
240	\n• Endpoints: https://getdozer.io/docs/reference/configuration/endpoints/",	×
241	yaml_path.to_owned());	×
242
243	let path = PathBuf::from(yaml_path);	×
244	if let Some(dir) = path.parent() {	×
245	let queries_path = Path::new(dir).join(DEFAULT_QUERIES_DIRECTORY);	×
246	if let Err(_e) = std::fs::create_dir(queries_path) {	×
247	warn!("Cannot create queries directory");	×
248	}	×
249
250	let lambdas_path = Path::new(dir).join(DEFAULT_LAMBDAS_DIRECTORY);	×
251	if let Err(_e) = std::fs::create_dir(lambdas_path) {	×
252	warn!("Cannot create lambdas directory");	×
253	}	×
254	}	×
255
256	Ok(())	×
257	}),	×
258	),	×
259	];	×
260	let result = questions.iter().try_for_each(\|(question, func)\| {	×
261	let readline = rl.readline(question);	×
262	match readline {	×
263	Ok(input) => func((input, &mut default_config)),	×
264	Err(err) => Err(OrchestrationError::CliError(CliError::ReadlineError(err))),	×
265	}
266	});	×
267		×
268	match result {	×
269	Ok(_) => Ok(()),	×
270	Err(e) => match e {	×
271	OrchestrationError::CliError(CliError::ReadlineError(ReadlineError::Interrupted)) => {
272	info!("Exiting..");	×
273	Ok(())	×
274	}
275	OrchestrationError::CliError(CliError::ReadlineError(ReadlineError::Eof)) => {
276	info!("CTRL-D - exiting...");	×
277	Ok(())	×
278	}
279	_ => Err(e),	×
280	},
281	}
282	}	×

getdozer / dozer / 6009657516

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