5888798292

Committed 17 Aug 2023 08:51AM UTC coverage: 76.025% (-1.4%) from 77.415%

Build # 5888798292

Build Type

push

github

Committed by

web-flow

Commit Message

feat: implement graph on live ui (#1847)

* feat: implement progress

* feat: implement enable progress flag

* feat: implement progress in live

* chore: fix clippy

* chore: always use telemetry metrics

* fix: Only run build once

---------

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

Run Details

536 of 536 new or added lines in 21 files covered. (100.0%)

46101 of 60639 relevant lines covered (76.03%)

40410.07 hits per line

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

98.68

/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;

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 = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    let runtime = Arc::new(runtime);
    let grouped_connections = runtime
        .block_on(builder.get_grouped_tables(&used_sources))
        .unwrap();

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

getdozer / dozer / 5888798292

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