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

71.64

/dozer-cli/src/pipeline/source_builder.rs

use crate::pipeline::connector_source::ConnectorSourceFactory;
use crate::shutdown::ShutdownReceiver;
use crate::OrchestrationError;
use dozer_core::appsource::{AppSourceManager, AppSourceMappings};
use dozer_ingestion::connectors::TableInfo;

use dozer_types::indicatif::MultiProgress;
use dozer_types::models::connection::Connection;
use dozer_types::models::source::Source;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::runtime::Runtime;

pub struct SourceBuilder<'a> {
    grouped_connections: HashMap<Connection, Vec<Source>>,
    progress: Option<&'a MultiProgress>,
}

const SOURCE_PORTS_RANGE_START: u16 = 1000;

impl<'a> SourceBuilder<'a> {
    pub fn new(
        grouped_connections: HashMap<Connection, Vec<Source>>,
        progress: Option<&'a MultiProgress>,
    ) -> Self {
        Self {
            grouped_connections,
            progress,
        }
    }

    pub fn get_ports(&self) -> HashMap<(&str, &str), u16> {
        let mut port: u16 = SOURCE_PORTS_RANGE_START;

        let mut ports = HashMap::new();
        for (conn, sources_group) in &self.grouped_connections {
            for source in sources_group {
                ports.insert((conn.name.as_str(), source.name.as_str()), port);
                port += 1;
            }
        }
        ports
    }

    pub async fn build_source_manager(
        &self,
        runtime: &Arc<Runtime>,
        shutdown: ShutdownReceiver,
    ) -> Result<AppSourceManager, OrchestrationError> {
        let mut asm = AppSourceManager::new();

        let mut port: u16 = SOURCE_PORTS_RANGE_START;

        for (connection, sources_group) in &self.grouped_connections {
            let mut ports = HashMap::new();
            let mut table_and_ports = vec![];
            for source in sources_group {
                ports.insert(source.name.clone(), port);

                table_and_ports.push((
                    TableInfo {
                        schema: source.schema.clone(),
                        name: source.table_name.clone(),
                        column_names: source.columns.clone(),
                    },
                    port,
                ));

                port += 1;
            }

            let source_factory = ConnectorSourceFactory::new(
                table_and_ports,
                connection.clone(),
                runtime.clone(),
                self.progress.cloned(),
                shutdown.clone(),
            )
            .await?;

            asm.add(
                Box::new(source_factory),
                AppSourceMappings::new(connection.name.to_string(), ports),
            )?;
        }

        Ok(asm)
    }
}

1	use crate::pipeline::connector_source::ConnectorSourceFactory;
2	use crate::shutdown::ShutdownReceiver;
3	use crate::OrchestrationError;
4	use dozer_core::appsource::{AppSourceManager, AppSourceMappings};
5	use dozer_ingestion::connectors::TableInfo;
6
7	use dozer_types::indicatif::MultiProgress;
8	use dozer_types::models::connection::Connection;
9	use dozer_types::models::source::Source;
10	use std::collections::HashMap;
11	use std::sync::Arc;
12	use tokio::runtime::Runtime;
13
14	pub struct SourceBuilder<'a> {
15	grouped_connections: HashMap<Connection, Vec<Source>>,
16	progress: Option<&'a MultiProgress>,
17	}
18
19	const SOURCE_PORTS_RANGE_START: u16 = 1000;
20
21	impl<'a> SourceBuilder<'a> {
22	pub fn new(	61✔
23	grouped_connections: HashMap<Connection, Vec<Source>>,	61✔
24	progress: Option<&'a MultiProgress>,	61✔
25	) -> Self {	61✔
26	Self {	61✔
27	grouped_connections,	61✔
28	progress,	61✔
29	}	61✔
30	}	61✔
31		×
32	pub fn get_ports(&self) -> HashMap<(&str, &str), u16> {	×
33	let mut port: u16 = SOURCE_PORTS_RANGE_START;	×
34		×
35	let mut ports = HashMap::new();	×
36	for (conn, sources_group) in &self.grouped_connections {	×
37	for source in sources_group {	×
38	ports.insert((conn.name.as_str(), source.name.as_str()), port);	×
39	port += 1;	×
40	}	×
41	}	×
42	ports	×
43	}	×
44		×
45	pub async fn build_source_manager(	61✔
46	&self,	61✔
47	runtime: &Arc<Runtime>,	61✔
48	shutdown: ShutdownReceiver,	61✔
49	) -> Result<AppSourceManager, OrchestrationError> {	61✔
50	let mut asm = AppSourceManager::new();	61✔
51		61✔
52	let mut port: u16 = SOURCE_PORTS_RANGE_START;	61✔
53		×
54	for (connection, sources_group) in &self.grouped_connections {	122✔
55	let mut ports = HashMap::new();	61✔
56	let mut table_and_ports = vec![];	61✔
57	for source in sources_group {	143✔
58	ports.insert(source.name.clone(), port);	82✔
59		82✔
60	table_and_ports.push((	82✔
61	TableInfo {	82✔
62	schema: source.schema.clone(),	82✔
63	name: source.table_name.clone(),	82✔
64	column_names: source.columns.clone(),	82✔
65	},	82✔
66	port,	82✔
67	));	82✔
68		82✔
69	port += 1;	82✔
70	}	82✔
71		×
72	let source_factory = ConnectorSourceFactory::new(	61✔
73	table_and_ports,	61✔
74	connection.clone(),	61✔
75	runtime.clone(),	61✔
76	self.progress.cloned(),	61✔
77	shutdown.clone(),	61✔
78	)	61✔
79	.await?;	120✔
80		×
81	asm.add(	61✔
82	Box::new(source_factory),	61✔
83	AppSourceMappings::new(connection.name.to_string(), ports),	61✔
84	)?;	61✔
85	}	×
86
87	Ok(asm)	61✔
88	}	61✔
89	}	×

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