5831539996

pending completion

Build # 5831539996

Build Type

Pull #1824

github

Committed by

chubei

Commit Message

feat: Publish DAG to JSON

Pull Request Pull Request #1824: feat: Publish Source contracts as JSON file

Run Details

249 of 249 new or added lines in 14 files covered. (100.0%)

45567 of 61724 relevant lines covered (73.82%)

56287.38 hits per line

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

53.13

/dozer-log/src/home_dir.rs

use camino::{ReadDirUtf8, Utf8PathBuf};

#[derive(Debug, Clone)]
pub struct HomeDir {
    home_dir: Utf8PathBuf,
    cache_dir: Utf8PathBuf,
}

pub type Error = (Utf8PathBuf, std::io::Error);

impl HomeDir {
    pub fn new(home_dir: String, cache_dir: String) -> Self {
        Self {
            home_dir: home_dir.into(),
            cache_dir: cache_dir.into(),
        }
    }

    pub fn create_build_dir_all(&self, build_id: BuildId) -> Result<BuildPath, Error> {
        std::fs::create_dir_all(&self.cache_dir).map_err(|e| (self.cache_dir.clone(), e))?;

        let build_path = self.get_build_path(build_id);

        std::fs::create_dir_all(&build_path.contracts_dir)
            .map_err(|e| (build_path.contracts_dir.clone(), e))?;
        std::fs::create_dir_all(&build_path.log_dir)
            .map_err(|e: std::io::Error| (build_path.log_dir.clone(), e))?;

        Ok(build_path)
    }

    pub fn find_build_path(&self, build_id: u32) -> Option<BuildPath> {
        let build_path = self.get_build_path(BuildId::from_id(build_id));
        if build_path.exists() {
            Some(build_path)
        } else {
            None
        }
    }

    pub fn find_latest_build_path(&self) -> Result<Option<BuildPath>, Error> {
        Ok(find_latest_build_id(self.home_dir.clone())?
            .map(|build_id| self.get_build_path(build_id)))
    }

    fn get_build_path(&self, build_id: BuildId) -> BuildPath {
        let build_dir = self.home_dir.join(&build_id.name);

        let contracts_dir = build_dir.join("contracts");
        let dag_path = contracts_dir.join("__dozer_pipeline.json");
        let descriptor_path = contracts_dir.join("file_descriptor_set.bin");

        let data_dir = build_dir.join("data");
        let log_dir = data_dir.join("log");

        BuildPath {
            id: build_id,
            contracts_dir,
            dag_path,
            descriptor_path,
            data_dir,
            log_dir,
        }
    }
}

#[derive(Debug, Clone)]
pub struct BuildId {
    id: u32,
    name: String,
}

impl BuildId {
    fn from_id(id: u32) -> Self {
        Self {
            id,
            name: format!("v{id:04}"),
        }
    }

    fn from_name(name: &str) -> Option<Self> {
        let id = name.strip_prefix('v').and_then(|s| s.parse::<u32>().ok())?;
        Some(Self {
            id,
            name: name.to_string(),
        })
    }

    pub fn first() -> Self {
        Self::from_id(1)
    }

    pub fn id(&self) -> u32 {
        self.id
    }

    pub fn name(&self) -> &str {
        &self.name
    }

    pub fn next(&self) -> Self {
        Self::from_id(self.id + 1)
    }
}

struct ListSubDir {
    parent_dir: Utf8PathBuf,
    read_dir: ReadDirUtf8,
}

struct SubDir {
    _path: Utf8PathBuf,
    name: String,
}

impl Iterator for ListSubDir {
    type Item = Result<SubDir, Error>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            let entry = self.read_dir.next()?;
            let entry = match entry {
                Err(e) => return Some(Err((self.parent_dir.clone(), e))),
                Ok(entry) => entry,
            };

            let path = entry.path();
            if path.is_dir() {
                return Some(Ok(SubDir {
                    _path: path.to_path_buf(),
                    name: entry.file_name().to_string(),
                }));
            }
        }
    }
}

fn list_sub_dir(dir: Utf8PathBuf) -> Result<ListSubDir, Error> {
    let read_dir = dir.read_dir_utf8().map_err(|e| (dir.clone(), e))?;
    Ok(ListSubDir {
        parent_dir: dir,
        read_dir,
    })
}

fn find_latest_build_id(dir: Utf8PathBuf) -> Result<Option<BuildId>, Error> {
    if !dir.exists() {
        return Ok(None);
    }

    let mut result = None;
    for sub_dir in list_sub_dir(dir)? {
        let sub_dir = sub_dir?;
        if let Some(build) = BuildId::from_name(&sub_dir.name) {
            if let Some(BuildId { id, .. }) = result {
                if build.id > id {
                    result = Some(build);
                }
            } else {
                result = Some(build);
            }
        }
    }
    Ok(result)
}

#[derive(Debug, Clone)]
pub struct BuildPath {
    pub id: BuildId,
    pub contracts_dir: Utf8PathBuf,
    pub dag_path: Utf8PathBuf,
    pub descriptor_path: Utf8PathBuf,
    data_dir: Utf8PathBuf,
    log_dir: Utf8PathBuf,
}

impl BuildPath {
    pub fn get_endpoint_path(&self, endpoint_name: &str) -> EndpointPath {
        let schema_path = self.contracts_dir.join(format!("{}.json", endpoint_name));
        let log_dir = self.log_dir.join(endpoint_name);
        EndpointPath {
            build_id: self.id.clone(),
            schema_path,
            log_dir,
        }
    }

    fn exists(&self) -> bool {
        self.contracts_dir.exists() && self.data_dir.exists()
    }
}

#[derive(Debug, Clone)]
pub struct EndpointPath {
    pub build_id: BuildId,
    pub schema_path: Utf8PathBuf,
    pub log_dir: Utf8PathBuf,
}

1	use camino::{ReadDirUtf8, Utf8PathBuf};
2
3	#[derive(Debug, Clone)]	×
4	pub struct HomeDir {
5	home_dir: Utf8PathBuf,
6	cache_dir: Utf8PathBuf,
7	}
8
9	pub type Error = (Utf8PathBuf, std::io::Error);
10
11	impl HomeDir {
12	pub fn new(home_dir: String, cache_dir: String) -> Self {	36✔
13	Self {	36✔
14	home_dir: home_dir.into(),	36✔
15	cache_dir: cache_dir.into(),	36✔
16	}	36✔
17	}	36✔
18		×
19	pub fn create_build_dir_all(&self, build_id: BuildId) -> Result<BuildPath, Error> {	×
20	std::fs::create_dir_all(&self.cache_dir).map_err(\|e\| (self.cache_dir.clone(), e))?;	18✔
21		×
22	let build_path = self.get_build_path(build_id);	18✔
23		18✔
24	std::fs::create_dir_all(&build_path.contracts_dir)	18✔
25	.map_err(\|e\| (build_path.contracts_dir.clone(), e))?;	18✔
26	std::fs::create_dir_all(&build_path.log_dir)	18✔
27	.map_err(\|e: std::io::Error\| (build_path.log_dir.clone(), e))?;	18✔
28
29	Ok(build_path)	18✔
30	}	18✔
31		×
32	pub fn find_build_path(&self, build_id: u32) -> Option<BuildPath> {	×
33	let build_path = self.get_build_path(BuildId::from_id(build_id));	×
34	if build_path.exists() {	×
35	Some(build_path)	×
36	} else {	×
37	None	×
38	}	×
39	}	×
40
41	pub fn find_latest_build_path(&self) -> Result<Option<BuildPath>, Error> {	36✔
42	Ok(find_latest_build_id(self.home_dir.clone())?	36✔
43	.map(\|build_id\| self.get_build_path(build_id)))	36✔
44	}	36✔
45
46	fn get_build_path(&self, build_id: BuildId) -> BuildPath {	36✔
47	let build_dir = self.home_dir.join(&build_id.name);	36✔
48		36✔
49	let contracts_dir = build_dir.join("contracts");	36✔
50	let dag_path = contracts_dir.join("__dozer_pipeline.json");	36✔
51	let descriptor_path = contracts_dir.join("file_descriptor_set.bin");	36✔
52		36✔
53	let data_dir = build_dir.join("data");	36✔
54	let log_dir = data_dir.join("log");	36✔
55		36✔
56	BuildPath {	36✔
57	id: build_id,	36✔
58	contracts_dir,	36✔
59	dag_path,	36✔
60	descriptor_path,	36✔
61	data_dir,	36✔
62	log_dir,	36✔
63	}	36✔
64	}	36✔
65	}	×
66
67	#[derive(Debug, Clone)]	57✔
68	pub struct BuildId {
69	id: u32,
70	name: String,	×
71	}	×
72		×
73	impl BuildId {
74	fn from_id(id: u32) -> Self {	18✔
75	Self {	18✔
76	id,	18✔
77	name: format!("v{id:04}"),	18✔
78	}	18✔
79	}	18✔
80		×
81	fn from_name(name: &str) -> Option<Self> {	36✔
82	let id = name.strip_prefix('v').and_then(\|s\| s.parse::<u32>().ok())?;	36✔
83	Some(Self {	18✔
84	id,	18✔
85	name: name.to_string(),	18✔
86	})	18✔
87	}	36✔
88		×
89	pub fn first() -> Self {	18✔
90	Self::from_id(1)	18✔
91	}	18✔
92		×
93	pub fn id(&self) -> u32 {	×
94	self.id	×
95	}	×
96		×
97	pub fn name(&self) -> &str {	36✔
98	&self.name	36✔
99	}	36✔
100		×
101	pub fn next(&self) -> Self {	×
102	Self::from_id(self.id + 1)	×
103	}	×
104	}	×
105		×
106	struct ListSubDir {	×
107	parent_dir: Utf8PathBuf,	×
108	read_dir: ReadDirUtf8,	×
109	}	×
110		×
111	struct SubDir {	×
112	_path: Utf8PathBuf,	×
113	name: String,	×
114	}	×
115		×
116	impl Iterator for ListSubDir {	×
117	type Item = Result<SubDir, Error>;
118		×
119	fn next(&mut self) -> Option<Self::Item> {	72✔
120	loop {
121	let entry = self.read_dir.next()?;	72✔
122	let entry = match entry {	36✔
123	Err(e) => return Some(Err((self.parent_dir.clone(), e))),	×
124	Ok(entry) => entry,	36✔
125	};	36✔
126		36✔
127	let path = entry.path();	36✔
128	if path.is_dir() {	36✔
129	return Some(Ok(SubDir {	36✔
130	_path: path.to_path_buf(),	36✔
131	name: entry.file_name().to_string(),	36✔
132	}));	36✔
133	}	×
134	}	×
135	}	72✔
136	}	×
137		×
138	fn list_sub_dir(dir: Utf8PathBuf) -> Result<ListSubDir, Error> {	36✔
139	let read_dir = dir.read_dir_utf8().map_err(\|e\| (dir.clone(), e))?;	36✔
140	Ok(ListSubDir {	36✔
141	parent_dir: dir,	36✔
142	read_dir,	36✔
143	})	36✔
144	}	36✔
145		×
146	fn find_latest_build_id(dir: Utf8PathBuf) -> Result<Option<BuildId>, Error> {	36✔
147	if !dir.exists() {	36✔
148	return Ok(None);	×
149	}	36✔
150		36✔
151	let mut result = None;	36✔
152	for sub_dir in list_sub_dir(dir)? {	36✔
153	let sub_dir = sub_dir?;	36✔
154	if let Some(build) = BuildId::from_name(&sub_dir.name) {	36✔
155	if let Some(BuildId { id, .. }) = result {	18✔
156	if build.id > id {	×
157	result = Some(build);	×
158	}	×
159	} else {	18✔
160	result = Some(build);	18✔
161	}	18✔
162	}	18✔
163	}
164	Ok(result)	36✔
165	}	36✔
166
167	#[derive(Debug, Clone)]	×
168	pub struct BuildPath {
169	pub id: BuildId,
170	pub contracts_dir: Utf8PathBuf,
171	pub dag_path: Utf8PathBuf,
172	pub descriptor_path: Utf8PathBuf,
173	data_dir: Utf8PathBuf,
174	log_dir: Utf8PathBuf,	×
175	}
176		×
177	impl BuildPath {	×
178	pub fn get_endpoint_path(&self, endpoint_name: &str) -> EndpointPath {	36✔
179	let schema_path = self.contracts_dir.join(format!("{}.json", endpoint_name));	36✔
180	let log_dir = self.log_dir.join(endpoint_name);	36✔
181	EndpointPath {	36✔
182	build_id: self.id.clone(),	36✔
183	schema_path,	36✔
184	log_dir,	36✔
185	}	36✔
186	}	36✔
187		×
188	fn exists(&self) -> bool {	×
189	self.contracts_dir.exists() && self.data_dir.exists()	×
190	}	×
191	}
192
193	#[derive(Debug, Clone)]	×
194	pub struct EndpointPath {	×
195	pub build_id: BuildId,	×
196	pub schema_path: Utf8PathBuf,	×
197	pub log_dir: Utf8PathBuf,	×
198	}	×

getdozer / dozer / 5831539996

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