use crate::config::{ClaimHint, Config, EnvCapture};
use anyhow::{bail, Context, Result};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
#[derive(Debug, Clone, Default, Deserialize, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct JobFile {
pub name: Option<String>,
pub cwd: Option<String>,
pub command: Vec<String>,
pub timeout: Option<String>,
pub tags: Vec<String>,
pub priority: Option<i32>,
pub env_capture: Option<EnvCapture>,
pub no_limit_env_hints: Option<bool>,
pub needs: Vec<String>,
pub after: Vec<String>,
pub locks: Vec<String>,
pub retries: Option<u32>,
pub nice: Option<i32>,
pub dedupe_key: Option<String>,
pub dedupe_window: Option<String>,
pub resources: Resources,
pub env: BTreeMap<String, String>,
}
#[derive(Debug, Clone, Default, Deserialize, Serialize)]
#[serde(default, deny_unknown_fields)]
pub struct Resources {
pub cpu: Option<crate::claim::Claim>,
pub mem: Option<crate::claim::Claim>,
}
impl JobFile {
pub fn load(path: &Path) -> Result<Self> {
let text = std::fs::read_to_string(path)
.with_context(|| format!("reading job file {}", path.display()))?;
let ext = path
.extension()
.and_then(|e| e.to_str())
.unwrap_or("toml")
.to_ascii_lowercase();
let parsed: Self = match ext.as_str() {
"yaml" | "yml" => {
serde_yaml_ng::from_str(&text).map_err(|e| job_file_error(path, &e.to_string()))?
}
"json" => {
serde_json::from_str(&text).map_err(|e| job_file_error(path, &e.to_string()))?
}
_ => toml::from_str(&text).map_err(|e| job_file_error(path, &e.to_string()))?,
};
Ok(parsed)
}
}
fn job_file_error(path: &Path, detail: &str) -> anyhow::Error {
anyhow::anyhow!(
"incorrect job file {}: {detail}\n\n\
A correct job file has this form:\n\n\
\x20 command = [\"uv\", \"run\", \"train.py\"]\n\n\
\x20 [resources]\n\
\x20 cpu = 2\n\
\x20 mem = \"4GB\"\n\n\
For a list of all the fields, run `qex help job-file`.",
path.display()
)
}
#[derive(Debug, Clone, Default)]
pub struct SubmitOptions {
pub name: Option<String>,
pub cwd: Option<PathBuf>,
pub cpu: Option<crate::claim::Claim>,
pub mem: Option<crate::claim::Claim>,
pub timeout: Option<String>,
pub tags: Vec<String>,
pub priority: Option<i32>,
pub env: Vec<(String, String)>,
pub env_capture: Option<EnvCapture>,
pub command: Vec<String>,
pub job_file: Option<PathBuf>,
pub needs: Vec<String>,
pub after: Vec<String>,
pub locks: Vec<String>,
pub retries: Option<u32>,
pub nice: Option<i32>,
pub no_limit_env_hints: bool,
pub dedupe_key: Option<String>,
pub dedupe_window: Option<String>,
}
#[derive(Debug, Clone, Default)]
pub struct DependencyNames {
pub needs: Vec<String>,
pub after: Vec<String>,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct JobSpec {
pub id: uuid::Uuid,
pub name: String,
pub cwd: PathBuf,
pub command: Vec<String>,
pub env: BTreeMap<String, String>,
pub cpu: u64,
pub mem: u64,
pub timeout: Option<u64>,
pub tags: Vec<String>,
pub priority: i32,
pub env_capture: EnvCapture,
#[serde(default)]
pub claim_source: String,
#[serde(default)]
pub group: Option<uuid::Uuid>,
#[serde(default)]
pub group_name: Option<String>,
#[serde(default)]
pub needs: Vec<uuid::Uuid>,
#[serde(default)]
pub after: Vec<uuid::Uuid>,
#[serde(default)]
pub locks: Vec<String>,
#[serde(default)]
pub retries: u32,
#[serde(default)]
pub nice: Option<i32>,
#[serde(default)]
pub dedupe_key: Option<String>,
#[serde(default)]
pub dedupe_window: u64,
pub submitted_at: u64,
}
impl JobSpec {
#[cfg(test)]
pub fn resolve(opts: &SubmitOptions, cfg: &Config) -> Result<Self> {
Self::resolve_with_deps(opts, cfg).map(|(spec, _)| spec)
}
pub fn resolve_with_deps(
opts: &SubmitOptions,
cfg: &Config,
) -> Result<(Self, DependencyNames)> {
let file = match &opts.job_file {
Some(p) => JobFile::load(p)?,
None => JobFile::default(),
};
Self::resolve_from_file(opts, cfg, file)
}
pub fn resolve_from_file(
opts: &SubmitOptions,
cfg: &Config,
file: JobFile,
) -> Result<(Self, DependencyNames)> {
let command = if !opts.command.is_empty() {
opts.command.clone()
} else {
file.command.clone()
};
if command.is_empty() {
bail!(
"no command.\n\n\
Write the command after `--`:\n\
\x20 qex submit --cpu 2 --mem 4GB -- uv run train.py\n\n\
Or set `command` in a job file:\n\
\x20 qex submit --job train.toml"
);
}
if !opts.command.is_empty() && !file.command.is_empty() {
bail!(
"there is a command after `--` and a command in the job file. \
Delete one command. qex must have one command only."
);
}
let capture = opts
.env_capture
.or(file.env_capture)
.unwrap_or(cfg.submit.env_capture);
let mut env = capture_env(capture, &cfg.submit.minimal_env);
for (k, v) in &file.env {
env.insert(k.clone(), v.clone());
}
for (k, v) in &opts.env {
env.insert(k.clone(), v.clone());
}
let cwd = match (&opts.cwd, &file.cwd) {
(Some(p), _) => p.clone(),
(None, Some(p)) => PathBuf::from(p),
(None, None) => std::env::current_dir()
.context("cannot determine the current directory to capture as the job's cwd")?,
};
let cwd = cwd.canonicalize().with_context(|| {
format!(
"the job directory {} does not exist, or qex cannot read it",
cwd.display()
)
})?;
if !cwd.is_dir() {
bail!(
"the job directory {} is a file, not a directory",
cwd.display()
);
}
let asked_cpu = opts.cpu.as_ref().or(file.resources.cpu.as_ref());
let asked_mem = opts.mem.as_ref().or(file.resources.mem.as_ref());
let learned = if cfg.learn.enabled && (asked_cpu.is_none() || asked_mem.is_none()) {
crate::usage::suggest(&crate::usage::load(), &cwd, &command, cfg.learn.margin)
} else {
None
};
let mut source = "default";
let cpu = match asked_cpu {
Some(c) => {
source = "explicit";
c.cores(cfg)
}
None => match &learned {
Some(s) => {
source = "learned";
s.cpu
}
None => cfg.default_cpu(),
},
}
.max(1);
let mem = match asked_mem {
Some(c) => {
if source != "learned" {
source = "explicit";
}
c.bytes(cfg)
}
None => match &learned {
Some(s) => {
source = "learned";
s.mem
}
None => cfg.default_mem()?,
},
};
let hints = cfg.claims.export_env
&& !opts.no_limit_env_hints
&& !file.no_limit_env_hints.unwrap_or(false);
let chosen = asked_cpu.is_some() && asked_mem.is_some();
if hints && chosen && capture != EnvCapture::None {
export_claim(&mut env, cpu, mem, &cfg.claims.also);
}
let timeout = match opts.timeout.as_ref().or(file.timeout.as_ref()) {
Some(s) => {
crate::units::parse_duration(s).map_err(|e| anyhow::anyhow!("--timeout: {e}"))?
}
None => cfg.default_timeout()?,
};
let name = opts
.name
.clone()
.or(file.name)
.unwrap_or_else(|| default_name(&command));
if name.parse::<uuid::Uuid>().is_ok() {
bail!(
"the name `{name}` has the form of a job id, and qex does not accept it.\n\n\
qex reads a dependency value as an id when it has this form, and an id \
follows a different rule from a name. A name with this form would avoid \
that rule.\n\n\
Use a name that a person can read, such as `build` or `test`."
);
}
let nice = opts.nice.or(file.nice);
if let Some(n) = nice {
if !(-20..=19).contains(&n) {
bail!(
"the nice value {n} is outside the range -20 to 19. Use a number from \
-20 to 19. The system takes no other number, and it does not say so: \
above the range it uses 19, and below the range it refuses the change \
on a machine with no privilege. A larger number gives way to the work \
of a person, and 0 asks for the priority of a command that you type."
);
}
}
let mut tags = file.tags;
tags.extend(opts.tags.iter().cloned());
tags.sort();
tags.dedup();
let dedupe_key = match opts.dedupe_key.clone().or(file.dedupe_key) {
Some(k) if k.trim().is_empty() => bail!(
"--dedupe-key is empty.\n\n\
An empty key holds no job, so it makes no submission idempotent.\n\n\
Give a key that names the work and the place, such as \
`--dedupe-key build:$(pwd)`."
),
Some(k) => Some(k.trim().to_string()),
None => None,
};
let dedupe_window = match opts.dedupe_window.as_ref().or(file.dedupe_window.as_ref()) {
Some(s) => crate::units::parse_duration(s)
.map_err(|e| anyhow::anyhow!("--dedupe-window: {e}"))?
.map(|d| d.as_secs())
.unwrap_or(0),
None => 0,
};
if dedupe_window > 0 && dedupe_key.is_none() {
bail!(
"--dedupe-window needs --dedupe-key.\n\n\
The window says how long a job that succeeded keeps its key. \
With no key, qex has nothing to keep, and the option does nothing.\n\n\
Add a key: `--dedupe-key build:$(pwd)`."
);
}
let mut deps = DependencyNames {
needs: file.needs,
after: file.after,
};
deps.needs.extend(opts.needs.iter().cloned());
deps.after.extend(opts.after.iter().cloned());
Ok((
Self {
id: uuid::Uuid::new_v4(),
name,
cwd,
command,
env,
cpu,
mem,
timeout: timeout.map(|d| d.as_secs()),
tags,
priority: opts.priority.or(file.priority).unwrap_or(0),
env_capture: capture,
claim_source: source.to_string(),
group: None,
group_name: None,
locks: {
let mut all = file.locks.clone();
all.extend(opts.locks.iter().cloned());
all.sort();
all.dedup();
all
},
retries: opts.retries.or(file.retries).unwrap_or(0),
nice,
dedupe_key,
dedupe_window,
needs: Vec::new(),
after: Vec::new(),
submitted_at: crate::sys::now_secs(),
},
deps,
))
}
}
fn default_name(command: &[String]) -> String {
command
.first()
.map(|c| {
Path::new(c)
.file_name()
.and_then(|f| f.to_str())
.unwrap_or(c)
.to_string()
})
.unwrap_or_else(|| "job".to_string())
}
fn capture_env(mode: EnvCapture, minimal: &[String]) -> BTreeMap<String, String> {
match mode {
EnvCapture::All => std::env::vars().collect(),
EnvCapture::Minimal => minimal
.iter()
.filter_map(|k| std::env::var(k).ok().map(|v| (k.clone(), v)))
.collect(),
EnvCapture::None => BTreeMap::new(),
}
}
pub fn parse_env_pair(s: &str) -> Result<(String, String), String> {
match s.split_once('=') {
Some((k, v)) if !k.is_empty() => Ok((k.to_string(), v.to_string())),
_ => Err(format!(
"incorrect --env value `{s}`. Use the form KEY=VALUE. \
Example: --env RUST_LOG=debug"
)),
}
}
const HEAP_FLOOR_MB: u64 = 4;
fn export_claim(
env: &mut std::collections::BTreeMap<String, String>,
cpu: u64,
mem: u64,
also: &[ClaimHint],
) {
let cores = cpu.to_string();
let mut set = |key: &str, value: &str| {
env.entry(key.to_string())
.or_insert_with(|| value.to_string());
};
set("QEX_CPU", &cores);
set("QEX_MEM", &mem.to_string());
set("QEX_MEM_MB", &(mem / (1 << 20)).to_string());
for key in [
"GOMAXPROCS", "OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS", "NUMEXPR_NUM_THREADS", "VECLIB_MAXIMUM_THREADS", "RAYON_NUM_THREADS", "JULIA_NUM_THREADS", "DOTNET_PROCESSOR_COUNT", "POLARS_MAX_THREADS", "CARGO_BUILD_JOBS", ] {
set(key, &cores);
}
if mem > 0 {
set("GOMEMLIMIT", &mem.to_string());
}
let heap_mb = (mem / (1 << 20)) * 3 / 4;
if heap_mb >= HEAP_FLOOR_MB {
set("NODE_OPTIONS", &format!("--max-old-space-size={heap_mb}"));
}
for name in also {
match name {
ClaimHint::Java => {
if heap_mb >= HEAP_FLOOR_MB {
set(
"JAVA_TOOL_OPTIONS",
&format!("-XX:ActiveProcessorCount={cores} -Xmx{heap_mb}m"),
);
} else {
set(
"JAVA_TOOL_OPTIONS",
&format!("-XX:ActiveProcessorCount={cores}"),
);
}
}
ClaimHint::Make => {
set("MAKEFLAGS", &format!("-j{cores}"));
}
}
}
}
#[cfg(test)]
mod tests {
#[test]
fn the_claim_reaches_the_job_and_replaces_nothing() {
use std::collections::BTreeMap;
let mut env = BTreeMap::new();
export_claim(&mut env, 2, 2 << 30, &[]);
assert_eq!(env["QEX_CPU"], "2");
assert_eq!(env["QEX_MEM"], (2u64 << 30).to_string());
assert_eq!(env["QEX_MEM_MB"], "2048");
for key in [
"GOMAXPROCS",
"OMP_NUM_THREADS",
"OPENBLAS_NUM_THREADS",
"MKL_NUM_THREADS",
"NUMEXPR_NUM_THREADS",
"VECLIB_MAXIMUM_THREADS",
"RAYON_NUM_THREADS",
"JULIA_NUM_THREADS",
"DOTNET_PROCESSOR_COUNT",
"POLARS_MAX_THREADS",
"CARGO_BUILD_JOBS",
] {
assert_eq!(env.get(key).map(String::as_str), Some("2"), "{key}");
}
assert_eq!(env["GOMEMLIMIT"], (2u64 << 30).to_string());
assert_eq!(env["NODE_OPTIONS"], "--max-old-space-size=1536");
assert!(!env.contains_key("JAVA_TOOL_OPTIONS"));
assert!(!env.contains_key("MAKEFLAGS"));
let mut env = BTreeMap::new();
export_claim(&mut env, 3, 4 << 30, &[ClaimHint::Java, ClaimHint::Make]);
assert!(env["JAVA_TOOL_OPTIONS"].contains("-XX:ActiveProcessorCount=3"));
assert!(env["JAVA_TOOL_OPTIONS"].contains("-Xmx3072m"));
assert_eq!(env["MAKEFLAGS"], "-j3");
}
#[test]
fn a_claim_too_small_for_a_heap_gives_no_heap() {
use std::collections::BTreeMap;
for mem_mb in 0..=5u64 {
let mut env = BTreeMap::new();
export_claim(&mut env, 2, mem_mb << 20, &[ClaimHint::Java]);
assert!(
!env.contains_key("NODE_OPTIONS"),
"a claim of {mem_mb}MB gives a heap below the floor: {env:?}"
);
assert_eq!(
env["JAVA_TOOL_OPTIONS"], "-XX:ActiveProcessorCount=2",
"a claim of {mem_mb}MB must give no -Xmx"
);
assert_eq!(env["GOMAXPROCS"], "2");
}
let mut env = BTreeMap::new();
export_claim(&mut env, 2, 6 << 20, &[ClaimHint::Java]);
assert_eq!(env["NODE_OPTIONS"], "--max-old-space-size=4");
assert_eq!(
env["JAVA_TOOL_OPTIONS"],
"-XX:ActiveProcessorCount=2 -Xmx4m"
);
let mut env = BTreeMap::new();
export_claim(&mut env, 2, 100 * 1024, &[ClaimHint::Java]);
assert_eq!(env["QEX_MEM_MB"], "0");
assert_eq!(env["GOMEMLIMIT"], (100u64 * 1024).to_string());
let mut env = BTreeMap::new();
export_claim(&mut env, 2, 0, &[ClaimHint::Java]);
assert!(
!env.contains_key("GOMEMLIMIT"),
"a claim of zero must give no memory limit: {env:?}"
);
assert!(!env.contains_key("NODE_OPTIONS"), "{env:?}");
assert_eq!(env["JAVA_TOOL_OPTIONS"], "-XX:ActiveProcessorCount=2");
assert_eq!(env["QEX_MEM"], "0");
assert_eq!(env["GOMAXPROCS"], "2");
}
#[test]
fn capture_none_receives_no_claim_either() {
let _guard = env_lock();
let cfg = Config::default();
assert!(cfg.claims.export_env, "the default writes the claim");
let mut o = opts(&["true"]);
o.env_capture = Some(EnvCapture::None);
o.env = vec![("MINE".into(), "1".into())];
o.cpu = Some(crate::claim::Claim::Exact(2));
o.mem = Some(crate::claim::Claim::Exact(2 << 30));
let spec = JobSpec::resolve(&o, &cfg).unwrap();
assert_eq!(spec.env.len(), 1, "got: {:?}", spec.env);
assert_eq!(spec.env.get("MINE").unwrap(), "1");
}
#[test]
fn half_a_claim_tells_the_job_nothing() {
let _guard = env_lock();
let mut cfg = Config::default();
cfg.learn.enabled = false;
for (cpu, mem) in [
(Some(crate::claim::Claim::Exact(2)), None),
(None, Some(crate::claim::Claim::Exact(4 << 30))),
] {
let mut o = opts(&["true"]);
o.env_capture = Some(EnvCapture::Minimal);
o.cpu = cpu.clone();
o.mem = mem.clone();
let spec = JobSpec::resolve(&o, &cfg).unwrap();
assert!(
!spec.env.contains_key("QEX_CPU") && !spec.env.contains_key("GOMAXPROCS"),
"half a claim ({cpu:?}, {mem:?}) must write nothing: {:?}",
spec.env
);
}
let mut o = opts(&["true"]);
o.env_capture = Some(EnvCapture::Minimal);
o.cpu = Some(crate::claim::Claim::Exact(2));
o.mem = Some(crate::claim::Claim::Exact(4 << 30));
let spec = JobSpec::resolve(&o, &cfg).unwrap();
assert_eq!(spec.env.get("QEX_CPU").map(String::as_str), Some("2"));
assert_eq!(spec.env.get("GOMAXPROCS").map(String::as_str), Some("2"));
}
#[test]
fn the_config_file_turns_the_claim_off() {
let _guard = env_lock();
let mut cfg = Config::default();
cfg.claims.export_env = false;
let mut o = opts(&["true"]);
o.env_capture = Some(EnvCapture::Minimal);
o.cpu = Some(crate::claim::Claim::Exact(2));
o.mem = Some(crate::claim::Claim::Exact(4 << 30));
let spec = JobSpec::resolve(&o, &cfg).unwrap();
assert!(!spec.env.contains_key("GOMAXPROCS"), "got: {:?}", spec.env);
cfg.claims.export_env = true;
let mut o = opts(&["true"]);
o.env_capture = Some(EnvCapture::Minimal);
o.cpu = Some(crate::claim::Claim::Exact(2));
o.mem = Some(crate::claim::Claim::Exact(4 << 30));
o.no_limit_env_hints = true;
let spec = JobSpec::resolve(&o, &cfg).unwrap();
assert!(!spec.env.contains_key("GOMAXPROCS"), "got: {:?}", spec.env);
}
#[test]
fn the_claim_never_replaces_a_value_that_exists() {
use std::collections::BTreeMap;
let mut env = BTreeMap::new();
env.insert("GOMAXPROCS".to_string(), "9".to_string());
env.insert("QEX_CPU".to_string(), "mine".to_string());
export_claim(&mut env, 2, 1 << 30, &[]);
assert_eq!(env["GOMAXPROCS"], "9", "an explicit value must stay");
assert_eq!(env["QEX_CPU"], "mine");
assert_eq!(env["OMP_NUM_THREADS"], "2");
}
use super::*;
use crate::testutil::{env_lock, EnvVar};
fn cfg_without_learning() -> Config {
let mut cfg = Config::default();
cfg.learn.enabled = false;
cfg
}
fn opts(command: &[&str]) -> SubmitOptions {
SubmitOptions {
command: command.iter().map(|s| s.to_string()).collect(),
..Default::default()
}
}
fn job_file(dir: &Path, name: &str, contents: &str) -> PathBuf {
std::fs::create_dir_all(dir).unwrap();
let p = dir.join(name);
std::fs::write(&p, contents).unwrap();
p
}
fn tmpdir(tag: &str) -> PathBuf {
let d = std::env::temp_dir().join(format!("qex-spec-{tag}-{}", std::process::id()));
std::fs::create_dir_all(&d).unwrap();
d
}
#[test]
fn env_is_captured_by_default() {
let _guard = env_lock();
let _m = EnvVar::set("QEX_TEST_MARKER", "present");
let spec = JobSpec::resolve(&opts(&["true"]), &Config::default()).unwrap();
assert_eq!(
spec.env.get("QEX_TEST_MARKER").map(String::as_str),
Some("present"),
"default capture should inherit the invoking environment"
);
}
#[test]
fn capture_none_drops_everything_but_explicit_values() {
let _guard = env_lock();
let _m = EnvVar::set("QEX_TEST_MARKER", "present");
let mut o = opts(&["true"]);
o.env_capture = Some(EnvCapture::None);
o.env = vec![("ONLY".into(), "this".into())];
let spec = JobSpec::resolve(&o, &Config::default()).unwrap();
assert_eq!(spec.env.len(), 1);
assert_eq!(spec.env.get("ONLY").unwrap(), "this");
}
#[test]
fn capture_minimal_keeps_only_the_allowlist() {
let _guard = env_lock();
let _m = EnvVar::set("QEX_TEST_MARKER", "present");
let _h = EnvVar::set("HOME", "/home/example");
let mut o = opts(&["true"]);
o.env_capture = Some(EnvCapture::Minimal);
let spec = JobSpec::resolve(&o, &Config::default()).unwrap();
assert_eq!(
spec.env.get("HOME").map(String::as_str),
Some("/home/example")
);
assert!(
!spec.env.contains_key("QEX_TEST_MARKER"),
"minimal capture leaked a non-allowlisted variable"
);
}
#[test]
fn overrides_apply_on_top_of_every_capture_mode() {
let _guard = env_lock();
for mode in [EnvCapture::All, EnvCapture::Minimal, EnvCapture::None] {
let mut o = opts(&["true"]);
o.env_capture = Some(mode);
o.env = vec![("PATH".into(), "/only/here".into())];
let spec = JobSpec::resolve(&o, &Config::default()).unwrap();
assert_eq!(
spec.env.get("PATH").unwrap(),
"/only/here",
"override lost under capture mode {mode:?}"
);
}
}
#[test]
fn cli_env_overrides_captured_env() {
let _guard = env_lock();
let _m = EnvVar::set("QEX_TEST_OVERRIDE", "from-shell");
let mut o = opts(&["true"]);
o.env = vec![("QEX_TEST_OVERRIDE".into(), "from-cli".into())];
let spec = JobSpec::resolve(&o, &Config::default()).unwrap();
assert_eq!(spec.env.get("QEX_TEST_OVERRIDE").unwrap(), "from-cli");
}
#[test]
fn precedence_runs_captured_then_job_file_then_cli() {
let _guard = env_lock();
let _m = EnvVar::set("QEX_LAYER", "captured");
let dir = tmpdir("precedence");
let mut o = opts(&["true"]);
assert_eq!(
JobSpec::resolve(&o, &Config::default()).unwrap().env["QEX_LAYER"],
"captured"
);
let jf = job_file(
&dir,
"j.toml",
"command = [\"true\"]\n[env]\nQEX_LAYER = \"job-file\"\n",
);
o = SubmitOptions {
job_file: Some(jf.clone()),
..Default::default()
};
assert_eq!(
JobSpec::resolve(&o, &Config::default()).unwrap().env["QEX_LAYER"],
"job-file"
);
o.env = vec![("QEX_LAYER".into(), "cli".into())];
assert_eq!(
JobSpec::resolve(&o, &Config::default()).unwrap().env["QEX_LAYER"],
"cli"
);
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn the_nice_value_runs_job_file_then_command_line() {
let _guard = env_lock();
let dir = tmpdir("nice");
let o = opts(&["true"]);
assert_eq!(JobSpec::resolve(&o, &Config::default()).unwrap().nice, None);
let jf = job_file(&dir, "j.toml", "command = [\"true\"]\nnice = 5\n");
let mut o = SubmitOptions {
job_file: Some(jf.clone()),
..Default::default()
};
assert_eq!(
JobSpec::resolve(&o, &Config::default()).unwrap().nice,
Some(5)
);
o.nice = Some(0);
assert_eq!(
JobSpec::resolve(&o, &Config::default()).unwrap().nice,
Some(0)
);
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn a_nice_value_outside_the_range_is_refused() {
let _guard = env_lock();
for n in [-21, 20, 100] {
let mut o = opts(&["true"]);
o.nice = Some(n);
let err = JobSpec::resolve(&o, &Config::default())
.unwrap_err()
.to_string();
assert!(
err.contains("-20 to 19"),
"the message must give the range, and it said: {err}"
);
}
for n in [-20, 0, 19] {
let mut o = opts(&["true"]);
o.nice = Some(n);
JobSpec::resolve(&o, &Config::default()).unwrap();
}
}
#[test]
fn job_files_parse_as_toml_yaml_or_json() {
let _guard = env_lock();
let dir = tmpdir("formats");
let cases = [
(
"j.toml",
"command = [\"echo\", \"hi\"]\nname = \"t\"\n[resources]\ncpu = 3\nmem = \"8GB\"\n",
),
(
"j.yaml",
"command: [echo, hi]\nname: t\nresources:\n cpu: 3\n mem: 8GB\n",
),
(
"j.json",
r#"{"command":["echo","hi"],"name":"t","resources":{"cpu":3,"mem":"8GB"}}"#,
),
];
for (fname, body) in cases {
let p = job_file(&dir, fname, body);
let o = SubmitOptions {
job_file: Some(p),
..Default::default()
};
let spec = JobSpec::resolve(&o, &Config::default())
.unwrap_or_else(|e| panic!("{fname} failed to resolve: {e}"));
assert_eq!(spec.command, vec!["echo", "hi"], "{fname}");
assert_eq!(spec.cpu, 3, "{fname}");
assert_eq!(spec.mem, 8 << 30, "{fname}");
assert_eq!(spec.name, "t", "{fname}");
}
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn a_bad_job_file_error_carries_a_working_example() {
let dir = tmpdir("badfile");
let p = job_file(&dir, "bad.toml", "command = \"not an array\"\n");
let err = JobFile::load(&p).unwrap_err().to_string();
assert!(err.contains("qex help job-file"), "got: {err}");
assert!(err.contains("command = ["), "error lacks an example: {err}");
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn unknown_job_file_fields_are_rejected_rather_than_ignored() {
let dir = tmpdir("unknown");
let p = job_file(
&dir,
"typo.toml",
"command = [\"true\"]\ntimeoutt = \"5m\"\n",
);
let err = JobFile::load(&p).unwrap_err().to_string();
assert!(err.contains("timeoutt"), "got: {err}");
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn config_defaults_supply_the_job_size() {
let _guard = env_lock();
let mut cfg: Config =
toml::from_str("[defaults]\ncpu = 4\nmem = \"6GB\"\ntimeout = \"30m\"\n").unwrap();
cfg.learn.enabled = false;
let spec = JobSpec::resolve(&opts(&["true"]), &cfg).unwrap();
assert_eq!(spec.cpu, 4);
assert_eq!(spec.mem, 6 << 30);
assert_eq!(spec.timeout, Some(1800));
}
#[test]
fn explicit_sizes_replace_the_config_defaults() {
let _guard = env_lock();
let cfg: Config =
toml::from_str("[defaults]\ncpu = 4\nmem = \"6GB\"\ntimeout = \"30m\"\n").unwrap();
let mut o = opts(&["true"]);
o.cpu = Some(crate::claim::Claim::Exact(2));
o.mem = Some(crate::claim::Claim::Exact(1 << 30));
o.timeout = Some("0".into());
let spec = JobSpec::resolve(&o, &cfg).unwrap();
assert_eq!(spec.cpu, 2);
assert_eq!(spec.mem, 1 << 30);
assert_eq!(spec.timeout, None, "`--timeout 0` must remove the limit");
let dir = tmpdir("defaults");
let p = job_file(
&dir,
"j.toml",
"command = [\"true\"]\n[resources]\ncpu = 3\nmem = \"2GB\"\n",
);
let o = SubmitOptions {
job_file: Some(p),
..Default::default()
};
let spec = JobSpec::resolve(&o, &cfg).unwrap();
assert_eq!(spec.cpu, 3);
assert_eq!(spec.mem, 2 << 30);
assert_eq!(spec.timeout, Some(1800));
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn built_in_defaults_supply_a_size() {
let _guard = env_lock();
let spec = JobSpec::resolve(&opts(&["true"]), &cfg_without_learning()).unwrap();
assert_eq!(spec.cpu, 1, "the default job must claim 1 core");
let cores = crate::sys::cpu_count().max(1);
let expected = (crate::sys::total_memory() / cores).max(1 << 28);
assert_eq!(
spec.mem, expected,
"the default memory must be the machine memory divided by the cores"
);
assert_eq!(
spec.timeout, None,
"a job must have no time limit by default"
);
}
#[test]
fn a_command_in_both_places_is_ambiguous_and_rejected() {
let dir = tmpdir("dupcmd");
let p = job_file(&dir, "j.toml", "command = [\"from-file\"]\n");
let mut o = opts(&["from-cli"]);
o.job_file = Some(p);
let err = JobSpec::resolve(&o, &Config::default())
.unwrap_err()
.to_string();
assert!(
err.contains("job file"),
"the error must name both sources: {err}"
);
assert!(
err.contains("--"),
"the error must name both sources: {err}"
);
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn last_env_flag_wins() {
let mut o = opts(&["true"]);
o.env = vec![("K".into(), "first".into()), ("K".into(), "second".into())];
let spec = JobSpec::resolve(&o, &Config::default()).unwrap();
assert_eq!(spec.env.get("K").unwrap(), "second");
}
#[test]
fn cwd_is_captured_and_made_absolute() {
let spec = JobSpec::resolve(&opts(&["true"]), &Config::default()).unwrap();
assert!(spec.cwd.is_absolute());
assert_eq!(
spec.cwd,
std::env::current_dir().unwrap().canonicalize().unwrap()
);
}
#[test]
fn missing_cwd_is_rejected_at_submit_time_not_at_run_time() {
let mut o = opts(&["true"]);
o.cwd = Some(PathBuf::from("/nonexistent/qex/dir"));
let err = JobSpec::resolve(&o, &Config::default())
.unwrap_err()
.to_string();
assert!(err.contains("does not exist"), "got: {err}");
}
#[test]
fn a_command_is_required_and_the_error_shows_both_ways_to_give_one() {
let err = JobSpec::resolve(&opts(&[]), &Config::default())
.unwrap_err()
.to_string();
assert!(err.contains("qex submit"), "error should show usage: {err}");
assert!(
err.contains("--job"),
"error should mention job files: {err}"
);
}
#[test]
fn a_name_with_the_form_of_an_id_is_refused() {
let _guard = env_lock();
let mut o = opts(&["true"]);
o.name = Some("550e8400-e29b-41d4-a716-446655440000".into());
let err = JobSpec::resolve(&o, &Config::default())
.unwrap_err()
.to_string();
assert!(err.contains("form of a job id"), "got: {err}");
o.name = Some("build".into());
assert!(JobSpec::resolve(&o, &Config::default()).is_ok());
}
#[test]
fn a_dedupe_key_comes_from_the_command_line_or_the_job_file() {
let _guard = env_lock();
let mut o = opts(&["true"]);
o.dedupe_key = Some("build:/x".into());
let spec = JobSpec::resolve(&o, &Config::default()).unwrap();
assert_eq!(spec.dedupe_key.as_deref(), Some("build:/x"));
assert_eq!(spec.dedupe_window, 0, "the default window is zero");
let dir = tmpdir("dedupe");
let p = job_file(
&dir,
"j.toml",
"command = [\"true\"]\ndedupe_key = \"from-file\"\ndedupe_window = \"1h\"\n",
);
let mut o = SubmitOptions {
job_file: Some(p),
..Default::default()
};
let spec = JobSpec::resolve(&o, &Config::default()).unwrap();
assert_eq!(spec.dedupe_key.as_deref(), Some("from-file"));
assert_eq!(spec.dedupe_window, 3600);
o.dedupe_key = Some("from-cli".into());
o.dedupe_window = Some("0".into());
let spec = JobSpec::resolve(&o, &Config::default()).unwrap();
assert_eq!(spec.dedupe_key.as_deref(), Some("from-cli"));
assert_eq!(spec.dedupe_window, 0);
std::fs::remove_dir_all(&dir).ok();
}
#[test]
fn a_dedupe_option_that_holds_no_job_is_refused() {
let _guard = env_lock();
let mut o = opts(&["true"]);
o.dedupe_key = Some(" ".into());
let err = JobSpec::resolve(&o, &Config::default())
.unwrap_err()
.to_string();
assert!(err.contains("empty"), "got: {err}");
let mut o = opts(&["true"]);
o.dedupe_window = Some("1h".into());
let err = JobSpec::resolve(&o, &Config::default())
.unwrap_err()
.to_string();
assert!(
err.contains("--dedupe-key"),
"the message must name the option that is missing: {err}"
);
}
#[test]
fn name_defaults_to_the_program_basename() {
let spec =
JobSpec::resolve(&opts(&["/usr/bin/python3", "x.py"]), &Config::default()).unwrap();
assert_eq!(spec.name, "python3");
}
#[test]
fn env_pairs_parse_and_values_may_contain_equals() {
assert_eq!(
parse_env_pair("K=a=b").unwrap(),
("K".to_string(), "a=b".to_string())
);
assert_eq!(
parse_env_pair("K=").unwrap(),
("K".to_string(), String::new())
);
assert!(parse_env_pair("noequals").is_err());
assert!(parse_env_pair("=novalue").is_err());
}
}