crates/nornir-identity-directory/src/import.rs
//! **The CSV import, planned whole and applied in one write** (A34-17, A34-9).
//!
//! The import was a loop of one store write per row inside one RPC: up to
//! 10 000 writes on the async runtime, and a failure half way left half the
//! file applied. Now [`plan`] decides every row against the directory as it
//! stands (pure), and [`run`] hands the whole plan to
//! [`DirectoryStore::apply_import`], which writes all of it or nothing. A dry
//! run is the same plan through the same checks with `commit: false`, so it
//! also says when the account has no room for the new employees.
//!
//! [`run`] is the body of the product's background job (Gunnar runs it as a
//! jera ChainQueue job, the console polls its row); it reports its progress
//! through the callback.
use crate::csv::parse_csv;
use crate::model::{DirectoryStore, NewUser, Source, StoreError, User, UserPatch};
/// What an update does to the person.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Change {
/// Names only.
Updated,
/// `active` goes false: gated, sessions ended.
Deactivated,
/// `active` goes true.
Reactivated,
}
/// One decided row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ImportOp {
/// A NEW employee.
Create { line: usize, user: NewUser },
/// A change to the directory user of that name.
Update {
line: usize,
principal_id: String,
user_name: String,
patch: UserPatch,
change: Change,
},
}
/// What an import came to.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Imported {
pub created: u32,
pub updated: u32,
pub deactivated: u32,
pub reactivated: u32,
pub unchanged: u32,
/// One sentence per row that was not read or not applied.
pub errors: Vec<String>,
/// **The principals whose sessions this import ended** (A34-9): every
/// deactivated person, for the console to sign out as SCIM's are.
pub ended_sessions: Vec<String>,
/// Nothing was written.
pub dry_run: bool,
}
/// The whole import, decided.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ImportPlan {
pub ops: Vec<ImportOp>,
/// The counts the plan comes to, and the rows that could not be read.
pub outcome: Imported,
}
impl ImportPlan {
/// The principals the plan deactivates.
pub fn deactivating(&self) -> Vec<String> {
self.ops
.iter()
.filter_map(|op| match op {
ImportOp::Update {
principal_id,
change: Change::Deactivated,
..
} => Some(principal_id.clone()),
_ => None,
})
.collect()
}
}
/// **Decide every row** of `csv` against `directory` (deleted rows
/// included, as the store lists them). A row whose user name the directory
/// holds is an update of that person; any other row a NEW employee;
/// `active=false` deactivates. Pure: nothing is written.
pub fn plan(csv: &str, directory: &[User]) -> Result<ImportPlan, String> {
let (rows, errors) = parse_csv(csv)?;
let mut out = ImportPlan {
outcome: Imported {
errors,
..Imported::default()
},
..ImportPlan::default()
};
for row in rows {
let known = directory
.iter()
.find(|u| !u.deleted && u.user_name == row.user_name);
match known {
None => {
out.outcome.created += 1;
out.ops.push(ImportOp::Create {
line: row.line,
user: NewUser {
user_name: row.user_name,
display_name: row.display_name,
email: row.email,
external_id: String::new(),
active: row.active,
source: Source::Csv,
},
});
}
Some(u) => {
let names_differ = (!row.display_name.is_empty()
&& row.display_name != u.display_name)
|| (!row.email.is_empty() && row.email != u.email);
let active_differs = row.active != u.active;
if !names_differ && !active_differs {
out.outcome.unchanged += 1;
continue;
}
let change = match (active_differs, row.active) {
(true, false) => Change::Deactivated,
(true, true) => Change::Reactivated,
_ => Change::Updated,
};
match change {
Change::Deactivated => out.outcome.deactivated += 1,
Change::Reactivated => out.outcome.reactivated += 1,
Change::Updated => out.outcome.updated += 1,
}
out.ops.push(ImportOp::Update {
line: row.line,
principal_id: u.principal_id.clone(),
user_name: row.user_name,
patch: UserPatch {
display_name: (!row.display_name.is_empty()).then_some(row.display_name),
email: (!row.email.is_empty()).then_some(row.email),
active: Some(row.active),
..UserPatch::default()
},
change,
});
}
}
}
Ok(out)
}
/// Where a running import is: `(done, total)` rows.
pub type Progress<'a> = &'a dyn Fn(u64, u64);
/// **Run one import**: read the directory, plan, and apply the plan in one
/// write (or, `dry_run`, check it and write nothing). A refusal of the write
/// is the whole import's: nothing was applied, and the sentence says why.
pub fn run(
store: &dyn DirectoryStore,
csv: &str,
dry_run: bool,
progress: Progress<'_>,
) -> Result<Imported, String> {
let directory = store.users().map_err(|e| e.to_string())?;
let plan = plan(csv, &directory)?;
let total = plan.ops.len() as u64;
progress(0, total);
store
.apply_import(&plan, !dry_run)
.map_err(|e: StoreError| format!("nothing was imported: {e}"))?;
progress(total, total);
let ended = if dry_run { Vec::new() } else { plan.deactivating() };
Ok(Imported {
ended_sessions: ended,
dry_run,
..plan.outcome
})
}
#[cfg(test)]
mod tests {
use super::*;
use crate::model::Group;
use std::cell::RefCell;
/// A store in memory whose `apply_import` is all-or-nothing, and which
/// refuses a create of the user name `full` (no room).
#[derive(Default)]
struct Mem {
users: RefCell<Vec<User>>,
writes: RefCell<u32>,
}
impl DirectoryStore for Mem {
fn users(&self) -> Result<Vec<User>, StoreError> {
Ok(self.users.borrow().clone())
}
fn provision(&self, u: &NewUser) -> Result<User, StoreError> {
let user = User {
principal_id: format!("acme.{}", u.user_name),
user_name: u.user_name.clone(),
display_name: u.display_name.clone(),
email: u.email.clone(),
external_id: u.external_id.clone(),
active: u.active,
..User::default()
};
self.users.borrow_mut().push(user.clone());
Ok(user)
}
fn update_user(&self, id: &str, p: &UserPatch) -> Result<(User, bool), StoreError> {
let mut users = self.users.borrow_mut();
let u = users
.iter_mut()
.find(|u| u.principal_id == id)
.ok_or_else(|| StoreError::NotFound(id.into()))?;
let changed = p.active.is_some_and(|a| a != u.active);
if let Some(a) = p.active {
u.active = a;
}
if let Some(d) = &p.display_name {
u.display_name = d.clone();
}
Ok((u.clone(), changed))
}
fn delete_user(&self, id: &str) -> Result<User, StoreError> {
Err(StoreError::NotFound(id.into()))
}
fn groups(&self) -> Result<Vec<Group>, StoreError> {
Ok(Vec::new())
}
fn put_group(&self, g: Group) -> Result<Group, StoreError> {
Ok(g)
}
fn remove_group(&self, _: &str) -> Result<bool, StoreError> {
Ok(false)
}
fn new_group_id(&self) -> String {
"g1".into()
}
fn now_unix_ms(&self) -> i64 {
0
}
fn apply_import(&self, plan: &ImportPlan, commit: bool) -> Result<(), StoreError> {
let saved = self.users.borrow().clone();
for op in &plan.ops {
let done = match op {
ImportOp::Create { user, .. } if user.user_name == "full" => Err(
StoreError::Precondition("the account has no room for full".into()),
),
ImportOp::Create { user, .. } => self.provision(user).map(|_| ()),
ImportOp::Update {
principal_id,
patch,
..
} => self.update_user(principal_id, patch).map(|_| ()),
};
if let Err(e) = done {
*self.users.borrow_mut() = saved;
return Err(e);
}
}
if commit {
*self.writes.borrow_mut() += 1;
} else {
*self.users.borrow_mut() = saved;
}
Ok(())
}
}
fn bob() -> User {
User {
principal_id: "acme.bob".into(),
user_name: "bob".into(),
display_name: "Bob".into(),
active: true,
..User::default()
}
}
fn no_progress(_: u64, _: u64) {}
/// ★ **A34-9: a CSV import that deactivates people names them**, so the
/// console ends their sessions as SCIM's deactivation does.
#[test]
fn a_deactivating_import_names_the_ended_sessions() {
let store = Mem::default();
store.users.borrow_mut().push(bob());
let done = run(
&store,
"user_name,display_name,active\nbob,Bob,false\ncarol,Carol,true\n",
false,
&no_progress,
)
.unwrap();
assert_eq!((done.created, done.deactivated), (1, 1));
assert_eq!(done.ended_sessions, vec!["acme.bob".to_owned()]);
assert!(!store.users.borrow()[0].active);
let dry = run(&store, "user_name,active\ncarol,false\n", true, &no_progress).unwrap();
assert_eq!(dry.deactivated, 1);
assert!(dry.ended_sessions.is_empty(), "a dry run ends nobody's session");
assert!(store.users.borrow()[1].active, "a dry run writes nothing");
}
/// ★ **A34-17: the import is ONE write, all or nothing.** A row the store
/// refuses half way (no room) leaves the directory exactly as it was; the
/// dry run says the same refusal.
#[test]
fn an_import_that_fails_half_way_applies_nothing() {
let store = Mem::default();
store.users.borrow_mut().push(bob());
let csv = "user_name,active\nbob,false\ndave,true\nfull,true\nerin,true\n";
let before = store.users.borrow().clone();
let err = run(&store, csv, false, &no_progress).unwrap_err();
assert!(err.contains("nothing was imported") && err.contains("no room"), "{err}");
assert_eq!(*store.users.borrow(), before);
assert_eq!(*store.writes.borrow(), 0);
let dry = run(&store, csv, true, &no_progress).unwrap_err();
assert!(dry.contains("no room"), "the dry run checks room: {dry}");
let ok = run(&store, "user_name\ndave\nerin\n", false, &no_progress).unwrap();
assert_eq!((ok.created, *store.writes.borrow()), (2, 1), "two creates, one write");
}
#[test]
fn the_plan_decides_each_row_once() {
let mut gone = bob();
gone.deleted = true;
let p = plan(
"user_name,display_name,active\nbob,Bob,true\nbob2,,\n",
&[bob(), gone],
)
.unwrap();
assert_eq!(p.outcome.unchanged, 1);
assert_eq!(p.outcome.created, 1);
assert!(p.deactivating().is_empty());
}
}