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

81.61

/dozer-cli/src/pipeline/tests/builder.rs

use std::sync::Arc;

use crate::pipeline::source_builder::SourceBuilder;
use crate::pipeline::PipelineBuilder;
use dozer_types::ingestion_types::{GrpcConfig, GrpcConfigSchemas};
use dozer_types::models::config::Config;

use dozer_types::indicatif::MultiProgress;
use dozer_types::models::connection::{Connection, ConnectionConfig};
use dozer_types::models::source::Source;
use tokio::runtime::Runtime;

fn get_default_config() -> Config {
    let schema_str = include_str!("./schemas.json");
    let grpc_conn = Connection {
        config: Some(ConnectionConfig::Grpc(GrpcConfig {
            schemas: Some(GrpcConfigSchemas::Inline(schema_str.to_string())),
            ..Default::default()
        })),
        name: "grpc_conn".to_string(),
    };

    Config {
        app_name: "multi".to_string(),
        api: Default::default(),
        flags: Default::default(),
        connections: vec![grpc_conn.clone()],
        sources: vec![
            Source {
                name: "grpc_conn_users".to_string(),
                table_name: "users".to_string(),
                columns: vec!["id".to_string(), "name".to_string()],
                connection: grpc_conn.name.clone(),
                schema: None,
                refresh_config: None,
            },
            Source {
                name: "grpc_conn_customers".to_string(),
                table_name: "customers".to_string(),
                columns: vec!["id".to_string(), "name".to_string()],
                connection: grpc_conn.name,
                schema: None,
                refresh_config: None,
            },
        ],
        ..Default::default()
    }
}

#[test]
fn load_multi_sources() {
    let config = get_default_config();

    let used_sources = config
        .sources
        .iter()
        .map(|s| s.name.clone())
        .collect::<Vec<_>>();

    let builder = PipelineBuilder::new(
        &config.connections,
        &config.sources,
        config.sql.as_deref(),
        config
            .endpoints
            .into_iter()
            .map(|endpoint| (endpoint, None))
            .collect(),
        MultiProgress::new(),
    );

    let runtime = Runtime::new().unwrap();
    let grouped_connections = runtime
        .block_on(builder.get_grouped_tables(&used_sources))
        .unwrap();

    let source_builder = SourceBuilder::new(grouped_connections, None);
    let asm = source_builder
        .build_source_manager(Arc::new(runtime))
        .unwrap();

    asm.get_endpoint(&config.sources[0].name).unwrap();
    asm.get_endpoint(&config.sources[1].name).unwrap();
}

1	use std::sync::Arc;
2
3	use crate::pipeline::source_builder::SourceBuilder;
4	use crate::pipeline::PipelineBuilder;
5	use dozer_types::ingestion_types::{GrpcConfig, GrpcConfigSchemas};
6	use dozer_types::models::config::Config;
7
8	use dozer_types::indicatif::MultiProgress;
9	use dozer_types::models::connection::{Connection, ConnectionConfig};
10	use dozer_types::models::source::Source;
11	use tokio::runtime::Runtime;
12
13	fn get_default_config() -> Config {	1✔
14	let schema_str = include_str!("./schemas.json");	1✔
15	let grpc_conn = Connection {	1✔
16	config: Some(ConnectionConfig::Grpc(GrpcConfig {	1✔
17	schemas: Some(GrpcConfigSchemas::Inline(schema_str.to_string())),	1✔
18	..Default::default()	1✔
19	})),	1✔
20	name: "grpc_conn".to_string(),	1✔
21	};	1✔
22		1✔
23	Config {	1✔
24	app_name: "multi".to_string(),	1✔
25	api: Default::default(),	1✔
26	flags: Default::default(),	1✔
27	connections: vec![grpc_conn.clone()],	1✔
28	sources: vec![	1✔
29	Source {	1✔
30	name: "grpc_conn_users".to_string(),	1✔
31	table_name: "users".to_string(),	1✔
32	columns: vec!["id".to_string(), "name".to_string()],	1✔
33	connection: grpc_conn.name.clone(),	1✔
34	schema: None,	1✔
35	refresh_config: None,	1✔
36	},	1✔
37	Source {	1✔
38	name: "grpc_conn_customers".to_string(),	1✔
39	table_name: "customers".to_string(),	1✔
40	columns: vec!["id".to_string(), "name".to_string()],	1✔
41	connection: grpc_conn.name,	1✔
42	schema: None,	1✔
43	refresh_config: None,	1✔
44	},	1✔
45	],	1✔
46	..Default::default()	1✔
47	}	1✔
48	}	1✔
49		×
50	#[test]	1✔
51	fn load_multi_sources() {	1✔
52	let config = get_default_config();	1✔
53		1✔
54	let used_sources = config	1✔
55	.sources	1✔
56	.iter()	1✔
57	.map(\|s\| s.name.clone())	2✔
58	.collect::<Vec<_>>();	1✔
59		1✔
60	let builder = PipelineBuilder::new(	1✔
61	&config.connections,	1✔
62	&config.sources,	1✔
63	config.sql.as_deref(),	1✔
64	config	1✔
65	.endpoints	1✔
66	.into_iter()	1✔
67	.map(\|endpoint\| (endpoint, None))	1✔
68	.collect(),	1✔
69	MultiProgress::new(),	1✔
70	);	1✔
71		1✔
72	let runtime = Runtime::new().unwrap();	1✔
73	let grouped_connections = runtime	1✔
74	.block_on(builder.get_grouped_tables(&used_sources))	1✔
75	.unwrap();	1✔
76		1✔
77	let source_builder = SourceBuilder::new(grouped_connections, None);	1✔
78	let asm = source_builder	1✔
79	.build_source_manager(Arc::new(runtime))	1✔
80	.unwrap();	1✔
81		1✔
82	asm.get_endpoint(&config.sources[0].name).unwrap();	1✔
83	asm.get_endpoint(&config.sources[1].name).unwrap();	1✔
84	}	1✔

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