configurablility of router side storage

Cargo.lock

@@ -2497,6 +2497,15 @@ dependencies = [
  "winapi-util",
 ]
 
+[[package]]
+name = "scc"
+version = "2.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "46e6f046b7fef48e2660c57ed794263155d713de679057f2d0c169bfc6e756cc"
+dependencies = [
+ "sdd",
+]
+
 [[package]]
 name = "schannel"
 version = "0.1.29"
@@ -2536,6 +2545,12 @@ version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
 
+[[package]]
+name = "sdd"
+version = "3.0.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "490dcfcbfef26be6800d11870ff2df8774fa6e86d047e3e8c8a76b25655e41ca"
+
 [[package]]
 name = "security-framework"
 version = "3.7.0"
@@ -2671,6 +2686,32 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "serial_test"
+version = "3.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "911bd979bf1070a3f3aa7b691a3b3e9968f339ceeec89e08c280a8a22207a32f"
+dependencies = [
+ "futures-executor",
+ "futures-util",
+ "log",
+ "once_cell",
+ "parking_lot 0.12.5",
+ "scc",
+ "serial_test_derive",
+]
+
+[[package]]
+name = "serial_test_derive"
+version = "3.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0a7d91949b85b0d2fb687445e448b40d322b6b3e4af6b44a29b21d9a5f33e6d9"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "sha1"
 version = "0.10.6"
@@ -3752,6 +3793,7 @@ dependencies = [
  "macaddr",
  "serde",
  "serde_json",
+ "serial_test",
  "tokio",
  "wakey-core",
 ]

src/lib.rs

@@ -3,6 +3,7 @@
 
 pub mod service;
 pub mod utils;
+pub use wakey_linux;
 
 pub use service::{
     broadcast_wake_targets, get_interface_summaries, get_interface_summary, get_ips, get_leases,

wakey-agent/src/cli.rs

@@ -106,6 +106,10 @@ pub struct ReloadArgs {
 
 #[derive(Args)]
 pub struct ObserveArgs {
+    /// Path to the agent config file. If present, local path settings are passed through.
+    #[arg(long, default_value = config::DEFAULT_CONFIG_PATH)]
+    pub config: PathBuf,
+
     #[command(subcommand)]
     pub command: ObserveCommand,
 }

wakey-agent/src/config.rs

@@ -4,6 +4,12 @@ use std::fmt;
 use std::path::{Path, PathBuf};
 
 pub const DEFAULT_CONFIG_PATH: &str = "/etc/wakey-agent/config.toml";
+const WAKEY_DHCP_LEASES_ENV: &str = "WAKEY_DHCP_LEASES";
+const WAKEY_MAC_NAME_CACHE_ENV: &str = "WAKEY_MAC_NAME_CACHE";
+const WAKEY_OBSERVATION_STORE_ENV: &str = "WAKEY_OBSERVATION_STORE";
+const DEFAULT_DHCP_LEASES_PATH: &str = "/tmp/dhcp.leases";
+const DEFAULT_MAC_NAME_CACHE_PATH: &str = "/tmp/wakey_mac_names.json";
+const DEFAULT_OBSERVATION_STORE_PATH: &str = "/tmp/wakey_observations.json";
 
 #[derive(Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct AgentConfig {
@@ -14,6 +20,14 @@ pub struct AgentConfig {
     pub reconnect_base_ms: u64,
     #[serde(default = "default_reconnect_max_ms")]
     pub reconnect_max_ms: u64,
+    #[serde(default = "default_observation_sync_interval_seconds")]
+    pub observation_sync_interval_seconds: u64,
+    #[serde(default = "default_dhcp_leases_path")]
+    pub dhcp_leases_path: PathBuf,
+    #[serde(default = "default_mac_name_cache_path")]
+    pub mac_name_cache_path: PathBuf,
+    #[serde(default = "default_observation_store_path")]
+    pub observation_store_path: PathBuf,
 }
 
 impl fmt::Debug for AgentConfig {
@@ -24,6 +38,13 @@ impl fmt::Debug for AgentConfig {
             .field("agent_token", &"<redacted>")
             .field("reconnect_base_ms", &self.reconnect_base_ms)
             .field("reconnect_max_ms", &self.reconnect_max_ms)
+            .field(
+                "observation_sync_interval_seconds",
+                &self.observation_sync_interval_seconds,
+            )
+            .field("dhcp_leases_path", &self.dhcp_leases_path)
+            .field("mac_name_cache_path", &self.mac_name_cache_path)
+            .field("observation_store_path", &self.observation_store_path)
             .finish()
     }
 }
@@ -36,6 +57,41 @@ const fn default_reconnect_max_ms() -> u64 {
     30_000
 }
 
+const fn default_observation_sync_interval_seconds() -> u64 {
+    60
+}
+
+fn default_dhcp_leases_path() -> PathBuf {
+    DEFAULT_DHCP_LEASES_PATH.into()
+}
+
+fn default_mac_name_cache_path() -> PathBuf {
+    DEFAULT_MAC_NAME_CACHE_PATH.into()
+}
+
+fn default_observation_store_path() -> PathBuf {
+    DEFAULT_OBSERVATION_STORE_PATH.into()
+}
+
+impl AgentConfig {
+    pub fn local_path_envs(&self) -> Vec<(&'static str, &Path)> {
+        vec![
+            (WAKEY_DHCP_LEASES_ENV, self.dhcp_leases_path.as_path()),
+            (WAKEY_MAC_NAME_CACHE_ENV, self.mac_name_cache_path.as_path()),
+            (
+                WAKEY_OBSERVATION_STORE_ENV,
+                self.observation_store_path.as_path(),
+            ),
+        ]
+    }
+}
+
+pub fn apply_local_path_env_to_command(cmd: &mut std::process::Command, config: &AgentConfig) {
+    for (key, path) in config.local_path_envs() {
+        cmd.env(key, path);
+    }
+}
+
 pub fn load_config(path: &Path) -> Result<AgentConfig> {
     let content = std::fs::read_to_string(path)
         .with_context(|| format!("failed to read agent config {}", path.display()))?;
@@ -118,12 +174,22 @@ mod tests {
             agent_token: "secret".into(),
             reconnect_base_ms: 123,
             reconnect_max_ms: 456,
+            observation_sync_interval_seconds: 7,
+            dhcp_leases_path: "/tmp/test-dhcp.leases".into(),
+            mac_name_cache_path: "/tmp/test-names.json".into(),
+            observation_store_path: "/tmp/test-observations.json".into(),
         };
 
         save_config(&path, &config).expect("save");
         let loaded = load_config(&path).expect("load");
         assert_eq!(loaded, config);
         assert!(format!("{:?}", loaded).contains("<redacted>"));
+        assert!(
+            loaded
+                .local_path_envs()
+                .iter()
+                .any(|(key, _)| *key == WAKEY_OBSERVATION_STORE_ENV)
+        );
         let _ = std::fs::remove_file(&path);
         let _ = std::fs::remove_dir_all(&dir);
     }

wakey-agent/src/dispatch.rs

@@ -1,27 +1,38 @@
 use anyhow::Result;
 use tracing::{debug, info, instrument};
 
+use crate::config::AgentConfig;
 use crate::protocol::{
     AgentCommand, CommandResult, DevsRequest, InventoryRequest, LeasesRequest, WakeRequest,
 };
 
 #[instrument(skip_all)]
-pub async fn dispatch_command(command: AgentCommand) -> Result<CommandResult> {
+pub async fn dispatch_command(
+    command: AgentCommand,
+    config: &AgentConfig,
+) -> Result<CommandResult> {
     let kind = command_kind(&command);
     info!(command = %kind, "dispatching command into local wakey services");
     match command {
-        AgentCommand::Leases(req) => dispatch_leases(req).await,
+        AgentCommand::Leases(req) => dispatch_leases(req, config).await,
         AgentCommand::Devs(req) => dispatch_devs(req).await,
-        AgentCommand::Inventory(req) => dispatch_inventory(req).await,
+        AgentCommand::Inventory(req) => dispatch_inventory(req, config).await,
         AgentCommand::Wake(req) => dispatch_wake(req).await,
     }
 }
 
-async fn dispatch_leases(req: LeasesRequest) -> Result<CommandResult> {
+async fn dispatch_leases(req: LeasesRequest, config: &AgentConfig) -> Result<CommandResult> {
-    let leases = wakey::get_leases(wakey_core::LeaseQuery {
+    let leases = wakey::wakey_linux::dhcp::read_dhcp_leases_with_names_from_paths(
-        include_state: req.include_state,
+        &config.dhcp_leases_path,
-    })
+        &config.observation_store_path,
+        &config.mac_name_cache_path,
+    )
     .await?;
+    let leases = if req.include_state {
+        wakey::wakey_linux::dhcp::enrich_leases_with_nud_state(leases).await
+    } else {
+        wakey::leases_without_state(leases)
+    };
     debug!(
         rows = leases.len(),
         include_state = req.include_state,
@@ -50,8 +61,18 @@ async fn dispatch_devs(req: DevsRequest) -> Result<CommandResult> {
     Ok(CommandResult::Devs { rows: devs })
 }
 
-async fn dispatch_inventory(req: InventoryRequest) -> Result<CommandResult> {
+async fn dispatch_inventory(req: InventoryRequest, config: &AgentConfig) -> Result<CommandResult> {
-    let inventory = wakey::inventory(req.into_inventory_query()).await?;
+    let query = req.into_inventory_query();
+    let neighbors = wakey::wakey_linux::devices::query_neighbors(&query).await?;
+    let leases = wakey::wakey_linux::dhcp::read_dhcp_leases_with_names_from_paths(
+        &config.dhcp_leases_path,
+        &config.observation_store_path,
+        &config.mac_name_cache_path,
+    )
+    .await?;
+    let inventory = wakey_core::DeviceInventory {
+        devices: wakey::merge_devices(neighbors, wakey::leases_without_state(leases), &query),
+    };
     debug!(
         rows = inventory.devices.len(),
         "dispatched inventory command"

wakey-agent/src/enroll.rs

@@ -60,6 +60,10 @@ pub async fn enroll(
         agent_token: payload.agent_token,
         reconnect_base_ms: 1_000,
         reconnect_max_ms: 30_000,
+        observation_sync_interval_seconds: 60,
+        dhcp_leases_path: "/tmp/dhcp.leases".into(),
+        mac_name_cache_path: "/tmp/wakey_mac_names.json".into(),
+        observation_store_path: "/tmp/wakey_observations.json".into(),
     };
     let backup_path = save_config_with_backup(config_path, &config)?;
     info!(agent_id = %config.agent_id, config_path = %config_path.display(), "agent enrollment succeeded and config was written");

wakey-agent/src/main.rs

@@ -95,6 +95,9 @@ async fn main() -> Result<()> {
 fn observe(args: cli::ObserveArgs) -> Result<()> {
     let mut cmd = std::process::Command::new(resolve_wakey_binary());
     cmd.arg("observe");
+    if let Ok(config) = config::load_config(&args.config) {
+        config::apply_local_path_env_to_command(&mut cmd, &config);
+    }
 
     match args.command {
         ObserveCommand::Dhcp(args) => {
@@ -173,6 +176,10 @@ fn init_config(args: InitConfigArgs) -> Result<()> {
             .unwrap_or_else(|| "REPLACE_ME_AGENT_TOKEN".to_string()),
         reconnect_base_ms: 1_000,
         reconnect_max_ms: 30_000,
+        observation_sync_interval_seconds: 60,
+        dhcp_leases_path: "/tmp/dhcp.leases".into(),
+        mac_name_cache_path: "/tmp/wakey_mac_names.json".into(),
+        observation_store_path: "/tmp/wakey_observations.json".into(),
     };
 
     if let Some(path) = &args.config {

wakey-agent/src/session.rs

@@ -88,7 +88,9 @@ async fn run_once(config: &AgentConfig) -> Result<()> {
     let http_client = reqwest::Client::new();
     let mut heartbeat = interval(Duration::from_secs(30));
     heartbeat.set_missed_tick_behavior(MissedTickBehavior::Skip);
-    let mut observation_sync = interval(Duration::from_secs(60));
+    let mut observation_sync = interval(Duration::from_secs(
+        config.observation_sync_interval_seconds.max(1),
+    ));
     observation_sync.set_missed_tick_behavior(MissedTickBehavior::Skip);
 
     loop {
@@ -117,7 +119,7 @@ async fn run_once(config: &AgentConfig) -> Result<()> {
                     Message::Text(text) => {
                         match serde_json::from_str::<ServerMessage>(&text) {
                             Ok(message) => {
-                                handle_server_message(&mut sink, message).await?;
+                                handle_server_message(config, &mut sink, message).await?;
                             }
                             Err(err) => {
                                 // Allow the server to introduce extra frame types without
@@ -163,9 +165,10 @@ struct AgentObservationRequest {
 }
 
 async fn send_agent_observations(client: &reqwest::Client, config: &AgentConfig) -> Result<()> {
-    let observations = wakey::list_local_observations()
+    let observations =
-        .await
+        wakey::wakey_linux::dhcp::list_local_observations_from_path(&config.observation_store_path)
-        .context("failed to read local observations")?;
+            .await
+            .context("failed to read local observations")?;
     if observations.is_empty() {
         return Ok(());
     }
@@ -215,7 +218,11 @@ pub fn next_backoff_ms(current_ms: u64, max_ms: u64) -> u64 {
     current_ms.saturating_mul(2).min(cap)
 }
 
-async fn handle_server_message<S>(sink: &mut S, message: ServerMessage) -> Result<()>
+async fn handle_server_message<S>(
+    config: &AgentConfig,
+    sink: &mut S,
+    message: ServerMessage,
+) -> Result<()>
 where
     S: SinkExt<Message> + Unpin,
     <S as futures_util::Sink<Message>>::Error: std::error::Error + Send + Sync + 'static,
@@ -227,7 +234,7 @@ where
         } => {
             let kind = command_kind(&command);
             info!(request_id = %request_id, command = %kind, "received command from control-plane");
-            match dispatch_command(command).await {
+            match dispatch_command(command, config).await {
                 Ok(result) => {
                     info!(request_id = %request_id, command = %kind, "command execution completed");
                     send_command_result(sink, request_id, kind, result).await?;

wakey-linux/Cargo.toml

@@ -13,6 +13,9 @@ serde_json = "1"
 tokio = { version = "1", features = ["fs", "net", "rt", "sync"] }
 wakey-core = { path = "../wakey-core", registry = "gitea", version = "0"}
 
+[dev-dependencies]
+serial_test = "3"
+
 [dependencies.lda-ipjs]
 path = "../ipjs"
 registry = "gitea"

wakey-linux/src/dhcp.rs

@@ -1,13 +1,18 @@
 use std::io::{self, ErrorKind};
 use std::net::IpAddr;
+use std::path::PathBuf;
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use macaddr::MacAddr;
 use serde::{Deserialize, Serialize};
 use wakey_core::{DhcpLease, DhcpLeaseWithState};
 
-const MAC_NAME_CACHE: &str = "/tmp/wakey_mac_names.json";
+const DEFAULT_DHCP_LEASES: &str = "/tmp/dhcp.leases";
-const OBSERVATION_STORE: &str = "/tmp/wakey_observations.json";
+const DEFAULT_MAC_NAME_CACHE: &str = "/tmp/wakey_mac_names.json";
+const DEFAULT_OBSERVATION_STORE: &str = "/tmp/wakey_observations.json";
+const DHCP_LEASES_ENV: &str = "WAKEY_DHCP_LEASES";
+const MAC_NAME_CACHE_ENV: &str = "WAKEY_MAC_NAME_CACHE";
+const OBSERVATION_STORE_ENV: &str = "WAKEY_OBSERVATION_STORE";
 
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct LocalObservationStore {
@@ -50,7 +55,13 @@ pub struct LocalDeviceObservation {
 
 /// Load the MAC-to-name cache used to preserve useful names across lease churn.
 pub async fn load_mac_name_cache() -> io::Result<std::collections::BTreeMap<String, String>> {
-    match tokio::fs::read_to_string(MAC_NAME_CACHE).await {
+    load_mac_name_cache_from_path(mac_name_cache_path()).await
+}
+
+pub async fn load_mac_name_cache_from_path(
+    path: impl AsRef<std::path::Path>,
+) -> io::Result<std::collections::BTreeMap<String, String>> {
+    match tokio::fs::read_to_string(path).await {
         Ok(s) => serde_json::from_str(&s).map_err(io::Error::other),
         Err(e) if e.kind() == ErrorKind::NotFound => Ok(Default::default()),
         Err(e) => Err(e),
@@ -59,13 +70,30 @@ pub async fn load_mac_name_cache() -> io::Result<std::collections::BTreeMap<Stri
 
 /// Persist the MAC-to-name cache back to disk.
 async fn save_mac_name_cache(map: &std::collections::BTreeMap<String, String>) -> io::Result<()> {
+    save_mac_name_cache_to_path(mac_name_cache_path(), map).await
+}
+
+async fn save_mac_name_cache_to_path(
+    path: impl AsRef<std::path::Path>,
+    map: &std::collections::BTreeMap<String, String>,
+) -> io::Result<()> {
     let s = serde_json::to_string(map).map_err(io::Error::other)?;
-    let _ = tokio::fs::write(MAC_NAME_CACHE, s).await;
+    let _ = tokio::fs::write(path, s).await;
     Ok(())
 }
 
 pub async fn load_observation_store() -> io::Result<LocalObservationStore> {
-    match tokio::fs::read_to_string(OBSERVATION_STORE).await {
+    match tokio::fs::read_to_string(observation_store_path()).await {
+        Ok(s) => serde_json::from_str(&s).map_err(io::Error::other),
+        Err(e) if e.kind() == ErrorKind::NotFound => Ok(Default::default()),
+        Err(e) => Err(e),
+    }
+}
+
+pub async fn load_observation_store_from_path(
+    path: impl AsRef<std::path::Path>,
+) -> io::Result<LocalObservationStore> {
+    match tokio::fs::read_to_string(path).await {
         Ok(s) => serde_json::from_str(&s).map_err(io::Error::other),
         Err(e) if e.kind() == ErrorKind::NotFound => Ok(Default::default()),
         Err(e) => Err(e),
@@ -74,11 +102,24 @@ pub async fn load_observation_store() -> io::Result<LocalObservationStore> {
 
 async fn save_observation_store(store: &LocalObservationStore) -> io::Result<()> {
     let s = serde_json::to_string(store).map_err(io::Error::other)?;
-    tokio::fs::write(OBSERVATION_STORE, s).await
+    tokio::fs::write(observation_store_path(), s).await
 }
 
 pub async fn list_local_observations() -> io::Result<Vec<LocalDeviceObservation>> {
     let store = load_observation_store().await?;
+    list_local_observations_from_store(store)
+}
+
+pub async fn list_local_observations_from_path(
+    path: impl AsRef<std::path::Path>,
+) -> io::Result<Vec<LocalDeviceObservation>> {
+    let store = load_observation_store_from_path(path).await?;
+    list_local_observations_from_store(store)
+}
+
+fn list_local_observations_from_store(
+    store: LocalObservationStore,
+) -> io::Result<Vec<LocalDeviceObservation>> {
     let mut out = Vec::with_capacity(store.dhcp_clients.len() + store.neighbors.len());
     out.extend(
         store
@@ -134,7 +175,7 @@ pub async fn observe_dhcp_event(
         .filter(|v| !v.is_empty() && *v != "*")
         .map(ToOwned::to_owned);
     let now = now_unix();
-    let mac_s = mac.to_string();
+    let mac_s = mac.to_string().to_ascii_lowercase();
 
     let mut store = load_observation_store().await.unwrap_or_default();
     let mut changed = false;
@@ -190,14 +231,14 @@ pub async fn observe_neighbor_event(
         return Ok(false);
     }
     let Some(key) = mac
-        .map(|value| format!("mac:{}", value))
+        .map(|value| format!("mac:{}", value.to_string().to_ascii_lowercase()))
         .or_else(|| ip.map(|value| format!("ip:{}", value)))
     else {
         return Ok(false);
     };
 
     let now = now_unix();
-    let mac = mac.map(|value| value.to_string());
+    let mac = mac.map(|value| value.to_string().to_ascii_lowercase());
     let mut store = load_observation_store().await.unwrap_or_default();
     let mut changed = false;
     store
@@ -248,9 +289,15 @@ pub fn parse_dhcp_lease_line(line: &str) -> Option<DhcpLease> {
     })
 }
 
-/// Read raw DHCP leases from `/tmp/dhcp.leases`.
+/// Read raw DHCP leases from the configured dnsmasq lease file.
 pub async fn read_dhcp_leases() -> io::Result<Vec<DhcpLease>> {
-    match tokio::fs::read_to_string("/tmp/dhcp.leases").await {
+    read_dhcp_leases_from_path(dhcp_leases_path()).await
+}
+
+pub async fn read_dhcp_leases_from_path(
+    path: impl AsRef<std::path::Path>,
+) -> io::Result<Vec<DhcpLease>> {
+    match tokio::fs::read_to_string(path).await {
         Ok(file) => Ok(file.lines().filter_map(parse_dhcp_lease_line).collect()),
         Err(e) if e.kind() == ErrorKind::NotFound => Ok(Vec::new()),
         Err(e) => Err(e),
@@ -259,9 +306,27 @@ pub async fn read_dhcp_leases() -> io::Result<Vec<DhcpLease>> {
 
 /// Read DHCP leases and fill missing names from the MAC-name cache.
 pub async fn read_dhcp_leases_with_names() -> io::Result<Vec<DhcpLease>> {
-    let leases = read_dhcp_leases().await?;
+    read_dhcp_leases_with_names_from_paths(
-    let observations = load_observation_store().await.unwrap_or_default();
+        dhcp_leases_path(),
-    let mut cache = load_mac_name_cache().await.unwrap_or_default();
+        observation_store_path(),
+        mac_name_cache_path(),
+    )
+    .await
+}
+
+pub async fn read_dhcp_leases_with_names_from_paths(
+    leases_path: impl AsRef<std::path::Path>,
+    observation_store_path: impl AsRef<std::path::Path>,
+    mac_name_cache_path: impl AsRef<std::path::Path>,
+) -> io::Result<Vec<DhcpLease>> {
+    let leases = read_dhcp_leases_from_path(leases_path).await?;
+    let observations = load_observation_store_from_path(observation_store_path)
+        .await
+        .unwrap_or_default();
+    let mac_name_cache_path = mac_name_cache_path.as_ref();
+    let mut cache = load_mac_name_cache_from_path(mac_name_cache_path)
+        .await
+        .unwrap_or_default();
     let mut changed = false;
     let mut leases_with_names = Vec::with_capacity(leases.len());
     for mut l in leases {
@@ -283,7 +348,7 @@ pub async fn read_dhcp_leases_with_names() -> io::Result<Vec<DhcpLease>> {
         leases_with_names.push(l);
     }
     if changed {
-        let _ = save_mac_name_cache(&cache).await;
+        let _ = save_mac_name_cache_to_path(mac_name_cache_path, &cache).await;
     }
     Ok(leases_with_names)
 }
@@ -295,6 +360,25 @@ fn now_unix() -> u64 {
         .unwrap_or(0)
 }
 
+fn dhcp_leases_path() -> PathBuf {
+    configured_path(DHCP_LEASES_ENV, DEFAULT_DHCP_LEASES)
+}
+
+fn mac_name_cache_path() -> PathBuf {
+    configured_path(MAC_NAME_CACHE_ENV, DEFAULT_MAC_NAME_CACHE)
+}
+
+fn observation_store_path() -> PathBuf {
+    configured_path(OBSERVATION_STORE_ENV, DEFAULT_OBSERVATION_STORE)
+}
+
+fn configured_path(env_key: &str, default: &str) -> PathBuf {
+    std::env::var_os(env_key)
+        .filter(|value| !value.is_empty())
+        .map(PathBuf::from)
+        .unwrap_or_else(|| PathBuf::from(default))
+}
+
 /// Enrich DHCP leases with the best currently known neighbor state per IP.
 pub async fn enrich_leases_with_nud_state(leases: Vec<DhcpLease>) -> Vec<DhcpLeaseWithState> {
     let ips: Vec<IpAddr> = leases.iter().map(|l| l.ip).collect();
@@ -323,3 +407,92 @@ pub async fn enrich_leases_with_nud_state(leases: Vec<DhcpLease>) -> Vec<DhcpLea
         })
         .collect()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serial_test::serial;
+
+    struct EnvGuard {
+        keys: Vec<&'static str>,
+    }
+
+    impl EnvGuard {
+        fn set(key: &'static str, value: &std::path::Path) -> Self {
+            let guard = Self { keys: vec![key] };
+            // SAFETY: these tests are serialized and do not spawn work that reads these
+            // environment variables outside the test body.
+            unsafe {
+                std::env::set_var(key, value);
+            }
+            guard
+        }
+    }
+
+    impl Drop for EnvGuard {
+        fn drop(&mut self) {
+            for key in &self.keys {
+                // SAFETY: these tests are serialized and do not spawn work that reads these
+                // environment variables outside the test body.
+                unsafe {
+                    std::env::remove_var(key);
+                }
+            }
+        }
+    }
+
+    fn temp_file(name: &str) -> PathBuf {
+        let nonce = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map(|duration| duration.as_nanos())
+            .unwrap_or(0);
+        std::env::temp_dir().join(format!("wakey-linux-{name}-{}-{nonce}", std::process::id()))
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn dhcp_lease_file_path_can_be_overridden() {
+        let path = temp_file("leases");
+        let _guard = EnvGuard::set(DHCP_LEASES_ENV, &path);
+        tokio::fs::write(&path, "1893456000 aa:bb:cc:dd:ee:ff 192.168.1.2 lda *\n")
+            .await
+            .expect("lease fixture should write");
+
+        let leases = read_dhcp_leases().await.expect("leases should read");
+        assert_eq!(leases.len(), 1);
+        assert_eq!(leases[0].name.as_deref(), Some("lda"));
+        assert_eq!(leases[0].ip.to_string(), "192.168.1.2");
+
+        let _ = tokio::fs::remove_file(path).await;
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn observation_and_name_cache_paths_can_be_overridden() {
+        let observation_path = temp_file("observations");
+        let cache_path = temp_file("names");
+        let _observation_guard = EnvGuard::set(OBSERVATION_STORE_ENV, &observation_path);
+        let _cache_guard = EnvGuard::set(MAC_NAME_CACHE_ENV, &cache_path);
+
+        let changed = observe_dhcp_event(
+            "add",
+            "aa:bb:cc:dd:ee:ff".parse().expect("mac should parse"),
+            Some("192.168.1.2".parse().expect("ip should parse")),
+            Some("lda"),
+        )
+        .await
+        .expect("observation should write");
+        assert!(changed);
+
+        let store = load_observation_store()
+            .await
+            .expect("observation store should read");
+        assert!(store.dhcp_clients.contains_key("aa:bb:cc:dd:ee:ff"));
+
+        let cache = load_mac_name_cache().await.expect("name cache should read");
+        assert_eq!(cache.get("aa:bb:cc:dd:ee:ff"), Some(&"lda".to_string()));
+
+        let _ = tokio::fs::remove_file(observation_path).await;
+        let _ = tokio::fs::remove_file(cache_path).await;
+    }
+}