From d86b9c815706186af5b33fab1ed379f2214bffe3 Mon Sep 17 00:00:00 2001
From: Arne Beer <contact@arne.beer>
Date: Tue, 21 Jan 2025 15:31:14 +0100
Subject: [PATCH] refactor: Clean up test code and don't wait if unnecessary

---
 CHANGELOG.md                                  |  3 +-
 pueue/src/daemon/callbacks.rs                 |  2 +-
 .../daemon/network/message_handler/kill.rs    | 16 +++---
 .../daemon/network/message_handler/restart.rs | 14 +++--
 pueue/src/daemon/network/response_helper.rs   |  2 +-
 pueue/src/daemon/process_handler/kill.rs      | 16 +++---
 pueue/tests/client/integration/env.rs         |  5 +-
 pueue/tests/client/integration/follow.rs      |  8 ++-
 pueue/tests/client/integration/log.rs         | 13 +++--
 pueue/tests/client/integration/restart.rs     | 22 ++++----
 pueue/tests/client/integration/status.rs      | 14 ++---
 pueue/tests/daemon/integration/add.rs         | 18 +++---
 pueue/tests/daemon/integration/aliases.rs     |  9 ++-
 pueue/tests/daemon/integration/clean.rs       | 10 ++--
 .../integration/environment_variables.rs      |  3 +-
 pueue/tests/daemon/integration/group.rs       |  4 +-
 pueue/tests/daemon/integration/kill.rs        | 20 +++++--
 pueue/tests/daemon/integration/log.rs         | 13 +++--
 .../daemon/integration/parallel_tasks.rs      |  8 +--
 pueue/tests/daemon/integration/pause.rs       | 46 +++++++++------
 pueue/tests/daemon/integration/priority.rs    | 18 +++---
 pueue/tests/daemon/integration/remove.rs      |  6 +-
 pueue/tests/daemon/integration/reset.rs       |  8 +--
 pueue/tests/daemon/integration/restart.rs     |  8 +--
 pueue/tests/daemon/integration/restore.rs     | 27 ++++-----
 pueue/tests/daemon/integration/start.rs       | 10 ++--
 pueue/tests/daemon/integration/stashed.rs     | 22 ++++----
 .../worker_environment_variables.rs           | 10 ++--
 pueue/tests/helper/asserts.rs                 | 56 ++++++++++++++++++-
 pueue_lib/src/task.rs                         |  5 ++
 30 files changed, 257 insertions(+), 159 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index dd2f6c58d..0a0246416 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,7 +10,8 @@ This release aims to further improve Pueue and to rectify some old design decisi
 
 ### Removing internal channel communication
 
-TLDR: Commands that start/stop/kill/pause tasks now only return when the task is actually started/stopped/killed/paused.
+TLDR: Commands that start/stop/pause tasks now only return when the task is actually started/stopped/paused.
+`kill`ing commands still takes a short while, as the process needs to be cleaned up properly.
 
 ---
 
diff --git a/pueue/src/daemon/callbacks.rs b/pueue/src/daemon/callbacks.rs
index 0884488b0..6ef20509a 100644
--- a/pueue/src/daemon/callbacks.rs
+++ b/pueue/src/daemon/callbacks.rs
@@ -68,7 +68,7 @@ pub fn build_callback_command(
     // This includes how many stashed and queued tasks are left in the group.
     parameters.insert("group", task.group.clone());
     let queued_tasks = state
-        .filter_tasks_of_group(|task| task.is_queued(), &task.group)
+        .filter_tasks_of_group(Task::is_queued, &task.group)
         .matching_ids
         .len();
     parameters.insert("queued_count", queued_tasks.to_string());
diff --git a/pueue/src/daemon/network/message_handler/kill.rs b/pueue/src/daemon/network/message_handler/kill.rs
index 1ca460b03..acb5994a5 100644
--- a/pueue/src/daemon/network/message_handler/kill.rs
+++ b/pueue/src/daemon/network/message_handler/kill.rs
@@ -1,9 +1,11 @@
-use pueue_lib::state::SharedState;
-use pueue_lib::success_msg;
-use pueue_lib::{network::message::*, settings::Settings};
+use pueue_lib::{
+    network::message::*, settings::Settings, state::SharedState, success_msg, task::Task,
+};
 
-use crate::daemon::network::response_helper::{ensure_group_exists, task_action_response_helper};
-use crate::daemon::process_handler;
+use crate::daemon::{
+    network::response_helper::{ensure_group_exists, task_action_response_helper},
+    process_handler,
+};
 
 /// Invoked when calling `pueue kill`.
 /// Forward the kill message to the task handler, which then kills the process.
@@ -23,7 +25,7 @@ pub fn kill(settings: &Settings, state: &SharedState, message: KillMessage) -> M
             TaskSelection::TaskIds(task_ids) => task_action_response_helper(
                 "Tasks are being killed",
                 task_ids.clone(),
-                |task| task.is_running(),
+                Task::is_running,
                 &state,
             ),
             TaskSelection::Group(group) => {
@@ -38,7 +40,7 @@ pub fn kill(settings: &Settings, state: &SharedState, message: KillMessage) -> M
             TaskSelection::TaskIds(task_ids) => task_action_response_helper(
                 "Tasks are being killed",
                 task_ids.clone(),
-                |task| task.is_running(),
+                Task::is_running,
                 &state,
             ),
             TaskSelection::Group(group) => success_msg!(
diff --git a/pueue/src/daemon/network/message_handler/restart.rs b/pueue/src/daemon/network/message_handler/restart.rs
index 51eefef10..885fc50c9 100644
--- a/pueue/src/daemon/network/message_handler/restart.rs
+++ b/pueue/src/daemon/network/message_handler/restart.rs
@@ -1,11 +1,13 @@
 use chrono::Local;
-use pueue_lib::settings::Settings;
 use std::sync::MutexGuard;
 
-use pueue_lib::aliasing::insert_alias;
-use pueue_lib::network::message::*;
-use pueue_lib::state::{SharedState, State};
-use pueue_lib::task::TaskStatus;
+use pueue_lib::{
+    aliasing::insert_alias,
+    network::message::*,
+    settings::Settings,
+    state::{SharedState, State},
+    task::{Task, TaskStatus},
+};
 
 use crate::daemon::process_handler;
 
@@ -28,7 +30,7 @@ pub fn restart_multiple(
     let response = task_action_response_helper(
         "Tasks has restarted",
         task_ids.clone(),
-        |task| task.is_done(),
+        Task::is_done,
         &state,
     );
 
diff --git a/pueue/src/daemon/network/response_helper.rs b/pueue/src/daemon/network/response_helper.rs
index 55fa38a24..186de668f 100644
--- a/pueue/src/daemon/network/response_helper.rs
+++ b/pueue/src/daemon/network/response_helper.rs
@@ -32,7 +32,7 @@ pub fn ensure_group_exists<'state>(
 /// task_action_response_helper(
 ///     "Tasks are being killed",
 ///     task_ids.clone(),
-///     |task| task.is_running(),
+///     Task::is_running,
 ///     &state,
 /// ),
 /// ```
diff --git a/pueue/src/daemon/process_handler/kill.rs b/pueue/src/daemon/process_handler/kill.rs
index d8bdec854..504801e34 100644
--- a/pueue/src/daemon/process_handler/kill.rs
+++ b/pueue/src/daemon/process_handler/kill.rs
@@ -1,10 +1,12 @@
 use log::{error, info, warn};
 
-use pueue_lib::network::message::{Signal, TaskSelection};
-use pueue_lib::process_helper::*;
-use pueue_lib::settings::Settings;
-use pueue_lib::state::GroupStatus;
-use pueue_lib::task::TaskStatus;
+use pueue_lib::{
+    network::message::{Signal, TaskSelection},
+    process_helper::*,
+    settings::Settings,
+    state::GroupStatus,
+    task::{Task, TaskStatus},
+};
 
 use crate::daemon::state_helper::{save_state, LockedState};
 use crate::ok_or_shutdown;
@@ -124,7 +126,7 @@ fn should_pause_group(state: &LockedState, issued_by_user: bool, group: &str) ->
         return false;
     }
 
-    // Check if there're tasks that're queued or enqueued.
-    let filtered_tasks = state.filter_tasks_of_group(|task| task.is_queued(), group);
+    // Check if there're tasks that're queued or scheduled to be enqueued.
+    let filtered_tasks = state.filter_tasks_of_group(Task::is_queued, group);
     !filtered_tasks.matching_ids.is_empty()
 }
diff --git a/pueue/tests/client/integration/env.rs b/pueue/tests/client/integration/env.rs
index 0b50e82b3..1fde1e2b0 100644
--- a/pueue/tests/client/integration/env.rs
+++ b/pueue/tests/client/integration/env.rs
@@ -1,4 +1,5 @@
 use anyhow::Result;
+use pueue_lib::task::Task;
 
 use crate::client::helper::*;
 
@@ -16,7 +17,7 @@ async fn set_environment() -> Result<()> {
 
     // Now start the command and wait for it to finish
     run_client_command(shared, &["enqueue", "0"])?;
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     let state = get_state(shared).await?;
     println!("{:#?}", state.tasks[&0].envs);
@@ -46,7 +47,7 @@ async fn unset_environment() -> Result<()> {
 
     // Now start the command and wait for it to finish
     run_client_command(shared, &["enqueue", "0"])?;
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     let state = get_state(shared).await?;
     println!("{:#?}", state.tasks[&0].envs);
diff --git a/pueue/tests/client/integration/follow.rs b/pueue/tests/client/integration/follow.rs
index aef1cda91..96481c5eb 100644
--- a/pueue/tests/client/integration/follow.rs
+++ b/pueue/tests/client/integration/follow.rs
@@ -1,6 +1,8 @@
 use anyhow::{Context, Result};
 use rstest::rstest;
 
+use pueue_lib::task::Task;
+
 use crate::client::helper::*;
 
 pub fn set_read_local_logs(daemon: &mut PueueDaemon, read_local_logs: bool) -> Result<()> {
@@ -27,7 +29,7 @@ async fn default(#[case] read_local_logs: bool) -> Result<()> {
 
     // Add a task and wait until it started.
     assert_success(add_task(shared, "sleep 1 && echo test").await?);
-    wait_for_task_condition(shared, 0, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 0, Task::is_running).await?;
 
     // Execute `follow`.
     // This will result in the client receiving the streamed output until the task finished.
@@ -51,7 +53,7 @@ async fn last_lines(#[case] read_local_logs: bool) -> Result<()> {
 
     // Add a task which echos 8 lines of output
     assert_success(add_task(shared, "echo \"1\n2\n3\n4\n5\n6\n7\n8\" && sleep 1").await?);
-    wait_for_task_condition(shared, 0, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 0, Task::is_running).await?;
 
     // Follow the task, but only print the last 4 lines of the output.
     let output = run_client_command(shared, &["follow", "--lines=4"])?;
@@ -123,7 +125,7 @@ async fn fail_on_non_existing(#[case] read_local_logs: bool) -> Result<()> {
 //
 //     // Add a task echoes something and waits for a while
 //     assert_success(add_task(shared, "echo test && sleep 20").await?);
-//     wait_for_task_condition(shared, 0, |task| task.is_running()).await?;
+//     wait_for_task_condition(shared, 0, Task::is_running).await?;
 //
 //     // Reset the daemon after 2 seconds. At this point, the client will already be following the
 //     // output and should notice that the task went away..
diff --git a/pueue/tests/client/integration/log.rs b/pueue/tests/client/integration/log.rs
index 69fa5d093..e86d2e0d3 100644
--- a/pueue/tests/client/integration/log.rs
+++ b/pueue/tests/client/integration/log.rs
@@ -1,10 +1,11 @@
 use std::collections::{BTreeMap, HashMap};
 
 use anyhow::{Context, Result};
-use pueue_lib::task::Task;
 use rstest::rstest;
 use serde::Deserialize;
 
+use pueue_lib::task::Task;
+
 use crate::client::helper::*;
 
 /// Test that the `log` command works for both:
@@ -28,7 +29,7 @@ async fn read(#[case] read_local_logs: bool) -> Result<()> {
 
     // Add a task and wait until it finishes.
     assert_success(add_task(shared, "echo test").await?);
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     let output = run_client_command(shared, &["log"])?;
 
@@ -59,7 +60,7 @@ async fn read_truncated(#[case] read_local_logs: bool) -> Result<()> {
 
     // Add a task and wait until it finishes.
     assert_success(add_task(shared, "echo '1\n2\n3\n4\n5\n6\n7\n8\n9\n10'").await?);
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     let output = run_client_command(shared, &["log", "--lines=5"])?;
 
@@ -77,7 +78,7 @@ async fn task_with_label() -> Result<()> {
 
     // Add a task and wait until it finishes.
     run_client_command(shared, &["add", "--label", "test_label", "echo test"])?;
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     let output = run_client_command(shared, &["log"])?;
 
@@ -95,7 +96,7 @@ async fn colored() -> Result<()> {
 
     // Add a task and wait until it finishes.
     assert_success(add_task(shared, "echo test").await?);
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     let output = run_client_command(shared, &["--color", "always", "log"])?;
 
@@ -122,7 +123,7 @@ async fn json() -> Result<()> {
 
     // Add a task and wait until it finishes.
     assert_success(add_task(shared, "echo test").await?);
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     let output = run_client_command(shared, &["log", "--json"])?;
 
diff --git a/pueue/tests/client/integration/restart.rs b/pueue/tests/client/integration/restart.rs
index 0454d014e..2358baffc 100644
--- a/pueue/tests/client/integration/restart.rs
+++ b/pueue/tests/client/integration/restart.rs
@@ -3,7 +3,7 @@ use std::collections::HashMap;
 use anyhow::Result;
 use assert_matches::assert_matches;
 
-use pueue_lib::task::{TaskResult, TaskStatus};
+use pueue_lib::task::{Task, TaskResult, TaskStatus};
 
 use crate::client::helper::*;
 
@@ -15,7 +15,7 @@ async fn restart_and_edit_task_command() -> Result<()> {
 
     // Create a task and wait for it to finish.
     assert_success(add_task(shared, "ls").await?);
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     // Set the editor to a command which replaces the temporary file's content.
     let mut envs = HashMap::new();
@@ -26,7 +26,7 @@ async fn restart_and_edit_task_command() -> Result<()> {
 
     // Restart the command, edit its command and wait for it to start.
     run_client_command_with_env(shared, &["restart", "--in-place", "--edit", "0"], envs)?;
-    wait_for_task_condition(shared, 0, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 0, Task::is_running).await?;
 
     // Make sure that both the command has been updated.
     let state = get_state(shared).await?;
@@ -49,7 +49,7 @@ async fn restart_and_edit_task_path() -> Result<()> {
 
     // Create a task and wait for it to finish.
     assert_success(add_task(shared, "ls").await?);
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     // Set the editor to a command which replaces the temporary file's content.
     let mut envs = HashMap::new();
@@ -57,7 +57,7 @@ async fn restart_and_edit_task_path() -> Result<()> {
 
     // Restart the command, edit its command and wait for it to start.
     run_client_command_with_env(shared, &["restart", "--in-place", "--edit", "0"], envs)?;
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     // Make sure that both the path has been updated.
     let state = get_state(shared).await?;
@@ -75,7 +75,7 @@ async fn restart_and_edit_task_path_and_command() -> Result<()> {
 
     // Create a task and wait for it to finish.
     assert_success(add_task(shared, "ls").await.unwrap());
-    wait_for_task_condition(shared, 0, |task| task.is_done())
+    wait_for_task_condition(shared, 0, Task::is_done)
         .await
         .unwrap();
 
@@ -92,7 +92,7 @@ echo '5' > ${PUEUE_EDIT_PATH}/0/priority || ",
     // Restart the command, edit its command and path and wait for it to start.
     // The task will fail afterwards, but it should still be edited.
     run_client_command_with_env(shared, &["restart", "--in-place", "--edit", "0"], envs)?;
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     // Make sure that both the path has been updated.
     let state = get_state(shared).await?;
@@ -123,7 +123,7 @@ async fn restart_and_edit_task_priority() -> Result<()> {
 
     // Create a task and wait for it to finish.
     assert_success(add_task(shared, "ls").await?);
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     // Set the editor to a command which replaces the temporary file's content.
     let mut envs = HashMap::new();
@@ -131,7 +131,7 @@ async fn restart_and_edit_task_priority() -> Result<()> {
 
     // Restart the command, edit its priority and wait for it to start.
     run_client_command_with_env(shared, &["restart", "--in-place", "--edit", "0"], envs)?;
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     // Make sure that the priority has been updated.
     let state = get_state(shared).await?;
@@ -149,7 +149,7 @@ async fn normal_restart_with_edit() -> Result<()> {
 
     // Create a task and wait for it to finish.
     assert_success(add_task(shared, "ls").await?);
-    let original_task = wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    let original_task = wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     // Set the editor to a command which replaces the temporary file's content.
     let mut envs = HashMap::new();
@@ -160,7 +160,7 @@ async fn normal_restart_with_edit() -> Result<()> {
 
     // Restart the command, edit its command and wait for it to start.
     run_client_command_with_env(shared, &["restart", "--edit", "0"], envs)?;
-    wait_for_task_condition(shared, 1, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 1, Task::is_running).await?;
 
     // Make sure that both the command has been updated.
     let state = get_state(shared).await?;
diff --git a/pueue/tests/client/integration/status.rs b/pueue/tests/client/integration/status.rs
index 1564cc997..ca28dcaff 100644
--- a/pueue/tests/client/integration/status.rs
+++ b/pueue/tests/client/integration/status.rs
@@ -1,6 +1,6 @@
-use anyhow::Context;
-use anyhow::Result;
-use pueue_lib::state::State;
+use anyhow::{Context, Result};
+
+use pueue_lib::{state::State, task::Task};
 
 use crate::client::helper::*;
 
@@ -42,7 +42,7 @@ async fn full() -> Result<()> {
 //
 //    // Add a task and wait until it finishes.
 //    assert_success(add_task(shared, "ls").await?);
-//    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+//    wait_for_task_condition(shared, 0, Task::is_done).await?;
 //
 //    let output = run_status_without_path(shared, &["--color", "always"]).await?;
 //
@@ -67,7 +67,7 @@ async fn single_group() -> Result<()> {
     run_client_command(shared, &["add", "--stashed", "ls"])?;
 
     // Make sure the first task finished.
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     let output = run_status_without_path(shared, &["--group", "testgroup"]).await?;
 
@@ -94,7 +94,7 @@ async fn multiple_groups() -> Result<()> {
     run_client_command(shared, &["add", "--group", "testgroup2", "ls"])?;
 
     // Make sure the second task finished.
-    wait_for_task_condition(shared, 1, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 1, Task::is_done).await?;
 
     let output = run_status_without_path(shared, &[]).await?;
 
@@ -113,7 +113,7 @@ async fn json() -> Result<()> {
 
     // Add a task and wait until it finishes.
     assert_success(add_task(shared, "ls").await?);
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     let output = run_client_command(shared, &["status", "--json"])?;
 
diff --git a/pueue/tests/daemon/integration/add.rs b/pueue/tests/daemon/integration/add.rs
index 7ff5c027f..0c10a3092 100644
--- a/pueue/tests/daemon/integration/add.rs
+++ b/pueue/tests/daemon/integration/add.rs
@@ -2,8 +2,7 @@ use anyhow::Result;
 use assert_matches::assert_matches;
 use chrono::Local;
 
-use pueue_lib::network::message::TaskSelection;
-use pueue_lib::task::*;
+use pueue_lib::{network::message::TaskSelection, task::*};
 
 use crate::helper::*;
 
@@ -19,7 +18,7 @@ async fn test_normal_add() -> Result<()> {
     assert_success(add_task(shared, "sleep 0.01").await?);
 
     // Wait until the task finished and get state
-    let task = wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    let task = wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     let post_addition_time = Local::now();
 
@@ -53,10 +52,7 @@ async fn test_stashed_add() -> Result<()> {
     assert_success(send_message(shared, message).await?);
 
     // Make sure the task is actually stashed.
-    wait_for_task_condition(shared, 0, |task| {
-        matches!(task.status, TaskStatus::Stashed { .. })
-    })
-    .await?;
+    assert_task_condition(shared, 0, Task::is_stashed, "The task should be stashed.").await?;
 
     Ok(())
 }
@@ -74,7 +70,13 @@ async fn test_add_with_immediate_start() -> Result<()> {
     assert_success(add_and_start_task(shared, "sleep 60").await?);
 
     // Make sure the task is actually being started.
-    wait_for_task_condition(shared, 0, |task| task.is_running()).await?;
+    assert_task_condition(
+        shared,
+        0,
+        Task::is_running,
+        "Tasks should start immediately",
+    )
+    .await?;
 
     Ok(())
 }
diff --git a/pueue/tests/daemon/integration/aliases.rs b/pueue/tests/daemon/integration/aliases.rs
index c36b6ed63..e19db3bbe 100644
--- a/pueue/tests/daemon/integration/aliases.rs
+++ b/pueue/tests/daemon/integration/aliases.rs
@@ -3,8 +3,7 @@ use std::collections::HashMap;
 use anyhow::Result;
 use assert_matches::assert_matches;
 
-use pueue_lib::network::message::*;
-use pueue_lib::task::*;
+use pueue_lib::{network::message::*, task::*};
 
 use crate::helper::*;
 
@@ -22,7 +21,7 @@ async fn test_add_with_alias() -> Result<()> {
     assert_success(add_task(shared, "non_existing_cmd test").await?);
 
     // Wait until the task finished and get state
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     let task = get_task(shared, 0).await?;
 
@@ -56,7 +55,7 @@ async fn test_restart_with_alias() -> Result<()> {
 
     // Add a task whose command that should fail and wait for it to finish.
     assert_success(add_task(shared, "non_existing_cmd test").await?);
-    let task = wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    let task = wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     // Ensure the command hasn't been mutated and the task failed.
     assert_eq!(task.command, "non_existing_cmd test");
@@ -87,7 +86,7 @@ async fn test_restart_with_alias() -> Result<()> {
         stashed: false,
     };
     send_message(shared, message).await?;
-    let task = wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    let task = wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     // The task finished successfully and its command has replaced the alias.
     assert_eq!(task.original_command, "replaced_cmd test");
diff --git a/pueue/tests/daemon/integration/clean.rs b/pueue/tests/daemon/integration/clean.rs
index e0eac5f83..75791767d 100644
--- a/pueue/tests/daemon/integration/clean.rs
+++ b/pueue/tests/daemon/integration/clean.rs
@@ -1,5 +1,5 @@
 use anyhow::Result;
-use pueue_lib::network::message::*;
+use pueue_lib::{network::message::*, task::Task};
 
 use crate::helper::*;
 
@@ -14,7 +14,7 @@ async fn test_normal_clean() -> Result<()> {
         assert_success(add_task(shared, command).await?);
     }
     // Wait for task2 to start. This implies that task[0,1] are done.
-    wait_for_task_condition(shared, 2, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 2, Task::is_running).await?;
 
     // Send the clean message
     let clean_message = CleanMessage {
@@ -42,7 +42,7 @@ async fn test_successful_only_clean() -> Result<()> {
         assert_success(add_task(shared, command).await?);
     }
     // Wait for task2 to start. This implies task[0,1] being finished.
-    wait_for_task_condition(shared, 1, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 1, Task::is_done).await?;
 
     // Send the clean message
     let clean_message = CleanMessage {
@@ -75,7 +75,7 @@ async fn test_clean_in_selected_group() -> Result<()> {
     }
 
     // Wait for task6 to start. This implies task[4,5] in the 'other' group being finished.
-    wait_for_task_condition(shared, 6, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 6, Task::is_running).await?;
 
     // Send the clean message
     let clean_message = CleanMessage {
@@ -113,7 +113,7 @@ async fn test_clean_successful_only_in_selected_group() -> Result<()> {
     }
 
     // Wait for task6 to start. This implies task[4,5] in the 'other' group being finished.
-    wait_for_task_condition(shared, 6, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 6, Task::is_running).await?;
 
     // Send the clean message
     let clean_message = CleanMessage {
diff --git a/pueue/tests/daemon/integration/environment_variables.rs b/pueue/tests/daemon/integration/environment_variables.rs
index 9e7583056..c3b2377a5 100644
--- a/pueue/tests/daemon/integration/environment_variables.rs
+++ b/pueue/tests/daemon/integration/environment_variables.rs
@@ -1,4 +1,5 @@
 use anyhow::Result;
+use pueue_lib::task::Task;
 
 use crate::helper::*;
 
@@ -14,7 +15,7 @@ async fn test_isolated_task_environment() -> Result<()> {
     // This environment variable is injected into the daemon's environment.
     // It shouldn't show up in the task's environment, as the task should be isolated!
     assert_success(add_and_start_task(shared, "echo $PUEUED_TEST_ENV_VARIABLE").await?);
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     let log = get_task_log(shared, 0, None).await?;
 
diff --git a/pueue/tests/daemon/integration/group.rs b/pueue/tests/daemon/integration/group.rs
index dd04c8ad3..fabe11408 100644
--- a/pueue/tests/daemon/integration/group.rs
+++ b/pueue/tests/daemon/integration/group.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
 
-use pueue_lib::network::message::*;
+use pueue_lib::{network::message::*, task::Task};
 
 use crate::helper::*;
 
@@ -65,7 +65,7 @@ async fn test_cannot_delete_group_with_tasks() -> Result<()> {
 
     // Add a task
     assert_success(add_task_to_group(shared, "ls", "testgroup").await?);
-    wait_for_task_condition(&daemon.settings.shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(&daemon.settings.shared, 0, Task::is_done).await?;
 
     // We shouldn't be capable of removing that group
     let message = GroupMessage::Remove("testgroup".to_string());
diff --git a/pueue/tests/daemon/integration/kill.rs b/pueue/tests/daemon/integration/kill.rs
index 78d6620a3..8a4cf9aad 100644
--- a/pueue/tests/daemon/integration/kill.rs
+++ b/pueue/tests/daemon/integration/kill.rs
@@ -49,9 +49,15 @@ async fn test_kill_tasks_with_pause(
     for _ in 0..3 {
         assert_success(add_and_start_task(shared, "sleep 60").await?);
     }
-    // Wait until all tasks are running
+    // Wait until all tasks are running, they should be start `immediately`.
     for id in 0..3 {
-        wait_for_task_condition(shared, id, |task| task.is_running()).await?;
+        assert_task_condition(
+            shared,
+            id,
+            Task::is_running,
+            "Tasks should start immediately.",
+        )
+        .await?;
     }
 
     // Add another task that will be normally enqueued.
@@ -124,9 +130,15 @@ async fn test_kill_tasks_without_pause(#[case] kill_message: KillMessage) -> Res
     for _ in 0..3 {
         assert_success(add_and_start_task(shared, "sleep 60").await?);
     }
-    // Wait until all tasks are running
+    // Wait until all tasks are running, they should be start `immediately`.
     for id in 0..3 {
-        wait_for_task_condition(shared, id, |task| task.is_running()).await?;
+        assert_task_condition(
+            shared,
+            id,
+            Task::is_running,
+            "Tasks should start immediately",
+        )
+        .await?;
     }
 
     // Add a dummy group that also shouldn't be paused.
diff --git a/pueue/tests/daemon/integration/log.rs b/pueue/tests/daemon/integration/log.rs
index 8ca557222..bc41be206 100644
--- a/pueue/tests/daemon/integration/log.rs
+++ b/pueue/tests/daemon/integration/log.rs
@@ -3,9 +3,10 @@ use std::fs::File;
 use std::path::Path;
 
 use anyhow::{bail, Context, Result};
-use pueue_lib::network::message::*;
 use tempfile::TempDir;
 
+use pueue_lib::{network::message::*, task::Task};
+
 use crate::helper::*;
 
 /// This function creates files `[1-20]` in the specified directory.
@@ -60,7 +61,7 @@ async fn test_full_log() -> Result<()> {
     // Add a task that lists those files and wait for it to finish.
     let command = format!("ls {tempdir_path:?}");
     assert_success(add_task(shared, &command).await?);
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     // Request all log lines
     let output = get_task_log(shared, 0, None).await?;
@@ -86,7 +87,7 @@ async fn test_partial_log() -> Result<()> {
     // Add a task that lists those files and wait for it to finish.
     let command = format!("ls {tempdir_path:?}");
     assert_success(add_task(shared, &command).await?);
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     // Debug output to see what the file actually looks like:
     let real_log_path = shared.pueue_directory().join("task_logs").join("0.log");
@@ -128,7 +129,7 @@ async fn test_correct_log_order() -> Result<()> {
     // Add a task that lists those files and wait for it to finish.
     let command = "echo 'test' && echo 'error' && echo 'test'";
     assert_success(add_task(shared, command).await?);
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     // Request all log lines
     let log_message = LogRequestMessage {
@@ -171,7 +172,7 @@ async fn logs_of_group() -> Result<()> {
     assert_success(add_task_to_group(shared, command, "test_2").await?);
 
     // Wait for both to finish
-    wait_for_task_condition(shared, 1, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 1, Task::is_done).await?;
 
     // Request the task's logs.
     let message = LogRequestMessage {
@@ -205,7 +206,7 @@ async fn logs_for_all() -> Result<()> {
     assert_success(add_task_to_group(shared, command, "test_2").await?);
 
     // Wait for both to finish
-    wait_for_task_condition(shared, 1, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 1, Task::is_done).await?;
 
     // Request the task's logs.
     let message = LogRequestMessage {
diff --git a/pueue/tests/daemon/integration/parallel_tasks.rs b/pueue/tests/daemon/integration/parallel_tasks.rs
index 2e1d5b428..5f744a479 100644
--- a/pueue/tests/daemon/integration/parallel_tasks.rs
+++ b/pueue/tests/daemon/integration/parallel_tasks.rs
@@ -24,7 +24,7 @@ async fn test_parallel_tasks() -> Result<()> {
 
     // Ensure those three tasks are started.
     for task_id in 0..3 {
-        wait_for_task_condition(shared, task_id, |task| task.is_running()).await?;
+        wait_for_task_condition(shared, task_id, Task::is_running).await?;
     }
 
     // Tasks 4-5 should still be queued
@@ -49,7 +49,7 @@ async fn test_parallel_tasks() -> Result<()> {
 
     // Ensure only two tasks are started.
     for task_id in 5..7 {
-        wait_for_task_condition(shared, task_id, |task| task.is_running()).await?;
+        wait_for_task_condition(shared, task_id, Task::is_running).await?;
     }
 
     // Tasks 8-10 should still be queued
@@ -79,7 +79,7 @@ async fn test_unlimited_parallel_tasks() -> Result<()> {
         assert_success(add_task_to_group(shared, "sleep 600", "testgroup").await?);
     }
     // Ensure the first tasks is started.
-    wait_for_task_condition(shared, 0, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 0, Task::is_running).await?;
 
     // Update the parallel limit of the group to 0
     let message = ParallelMessage {
@@ -90,7 +90,7 @@ async fn test_unlimited_parallel_tasks() -> Result<()> {
 
     // Make sure all other tasks are started as well in quick succession.
     for task_id in 1..10 {
-        wait_for_task_condition(shared, task_id, |task| task.is_running()).await?;
+        wait_for_task_condition(shared, task_id, Task::is_running).await?;
     }
 
     Ok(())
diff --git a/pueue/tests/daemon/integration/pause.rs b/pueue/tests/daemon/integration/pause.rs
index 18e88c498..3a4a3b2dc 100644
--- a/pueue/tests/daemon/integration/pause.rs
+++ b/pueue/tests/daemon/integration/pause.rs
@@ -40,21 +40,26 @@ async fn test_pause_running_task() -> Result<()> {
 
     // Start a long running task and make sure it's started
     add_task(shared, "sleep 60").await?;
-    wait_for_task_condition(shared, 0, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 0, Task::is_running).await?;
 
     // This pauses the daemon
     pause_tasks(shared, TaskSelection::All).await?;
 
     // Make sure the task as well as the default group get paused
-    wait_for_task_condition(shared, 0, |task| {
-        matches!(task.status, TaskStatus::Paused { .. })
-    })
+    assert_group_status(
+        shared,
+        PUEUE_DEFAULT_GROUP,
+        GroupStatus::Paused,
+        "Default group should be paused.",
+    )
+    .await?;
+    assert_task_condition(
+        shared,
+        0,
+        Task::is_paused,
+        "All default groups should be paused.",
+    )
     .await?;
-    let state = get_state(shared).await?;
-    assert_eq!(
-        state.groups.get(PUEUE_DEFAULT_GROUP).unwrap().status,
-        GroupStatus::Paused
-    );
 
     Ok(())
 }
@@ -67,7 +72,7 @@ async fn test_pause_with_wait() -> Result<()> {
 
     // Start a long running task and make sure it's started
     add_task(shared, "sleep 60").await?;
-    wait_for_task_condition(shared, 0, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 0, Task::is_running).await?;
 
     // Pauses the default queue while waiting for tasks
     let message = PauseMessage {
@@ -79,13 +84,20 @@ async fn test_pause_with_wait() -> Result<()> {
         .context("Failed to send message")?;
 
     // Make sure the default group gets paused, but the task is still running
-    wait_for_group_status(shared, PUEUE_DEFAULT_GROUP, GroupStatus::Paused).await?;
-    let state = get_state(shared).await?;
-    assert_matches!(
-        state.tasks.get(&0).unwrap().status,
-        TaskStatus::Running { .. },
-        "Task should continue running after group is paused."
-    );
+    assert_group_status(
+        shared,
+        PUEUE_DEFAULT_GROUP,
+        GroupStatus::Paused,
+        "Default group should be paused.",
+    )
+    .await?;
+    assert_task_condition(
+        shared,
+        0,
+        Task::is_running,
+        "Task should continue running after group is paused.",
+    )
+    .await?;
 
     Ok(())
 }
diff --git a/pueue/tests/daemon/integration/priority.rs b/pueue/tests/daemon/integration/priority.rs
index 557329567..2929d7d60 100644
--- a/pueue/tests/daemon/integration/priority.rs
+++ b/pueue/tests/daemon/integration/priority.rs
@@ -1,7 +1,7 @@
 use anyhow::Result;
 use rstest::rstest;
 
-use pueue_lib::network::message::TaskSelection;
+use pueue_lib::{network::message::TaskSelection, task::Task};
 
 use crate::helper::*;
 
@@ -26,8 +26,8 @@ async fn test_default_ordering(#[case] priority: i32) -> Result<()> {
     start_tasks(shared, TaskSelection::All).await?;
 
     // Make sure task 0 is being started and task 1 is still waiting.
-    wait_for_task_condition(shared, 0, |task| task.is_running()).await?;
-    wait_for_task_condition(shared, 1, |task| task.is_queued()).await?;
+    wait_for_task_condition(shared, 0, Task::is_running).await?;
+    wait_for_task_condition(shared, 1, Task::is_queued).await?;
 
     Ok(())
 }
@@ -50,9 +50,9 @@ async fn test_highest_priority_first() -> Result<()> {
     start_tasks(shared, TaskSelection::All).await?;
 
     // Make sure task 0 is being started and task 1 is still waiting.
-    wait_for_task_condition(shared, 2, |task| task.is_running()).await?;
-    wait_for_task_condition(shared, 1, |task| task.is_queued()).await?;
-    wait_for_task_condition(shared, 0, |task| task.is_queued()).await?;
+    wait_for_task_condition(shared, 2, Task::is_running).await?;
+    wait_for_task_condition(shared, 1, Task::is_queued).await?;
+    wait_for_task_condition(shared, 0, Task::is_queued).await?;
 
     Ok(())
 }
@@ -75,9 +75,9 @@ async fn test_default_priority_over_negative_priority() -> Result<()> {
     start_tasks(shared, TaskSelection::All).await?;
 
     // Make sure task 0 is being started and task 1 is still waiting.
-    wait_for_task_condition(shared, 2, |task| task.is_running()).await?;
-    wait_for_task_condition(shared, 0, |task| task.is_queued()).await?;
-    wait_for_task_condition(shared, 1, |task| task.is_queued()).await?;
+    wait_for_task_condition(shared, 2, Task::is_running).await?;
+    wait_for_task_condition(shared, 0, Task::is_queued).await?;
+    wait_for_task_condition(shared, 1, Task::is_queued).await?;
 
     Ok(())
 }
diff --git a/pueue/tests/daemon/integration/remove.rs b/pueue/tests/daemon/integration/remove.rs
index 6b4e27696..548c4b9f7 100644
--- a/pueue/tests/daemon/integration/remove.rs
+++ b/pueue/tests/daemon/integration/remove.rs
@@ -1,5 +1,5 @@
 use anyhow::Result;
-use pueue_lib::network::message::*;
+use pueue_lib::{network::message::*, task::Task};
 
 use crate::helper::*;
 
@@ -22,11 +22,11 @@ async fn test_normal_remove() -> Result<()> {
         assert_success(add_task(shared, command).await?);
     }
     // Wait for task2 to start. This implies task[0,1] being finished.
-    wait_for_task_condition(shared, 2, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 2, Task::is_running).await?;
 
     // Explicitly start task3, wait for it to start and directly pause it.
     start_tasks(shared, TaskSelection::TaskIds(vec![3])).await?;
-    wait_for_task_condition(shared, 3, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 3, Task::is_running).await?;
 
     pause_tasks(shared, TaskSelection::TaskIds(vec![3])).await?;
 
diff --git a/pueue/tests/daemon/integration/reset.rs b/pueue/tests/daemon/integration/reset.rs
index 9e5bfa904..13d3422ff 100644
--- a/pueue/tests/daemon/integration/reset.rs
+++ b/pueue/tests/daemon/integration/reset.rs
@@ -1,5 +1,5 @@
 use anyhow::{Context, Result};
-use pueue_lib::{network::message::*, state::GroupStatus};
+use pueue_lib::{network::message::*, state::GroupStatus, task::Task};
 
 use crate::helper::*;
 
@@ -14,7 +14,7 @@ async fn test_reset() -> Result<()> {
     add_task(shared, "failed").await?;
     add_task_to_group(shared, "sleep 60", "test_2").await?;
     add_task(shared, "ls").await?;
-    wait_for_task_condition(shared, 2, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 2, Task::is_running).await?;
 
     // Reset all groups of the daemon
     send_message(
@@ -60,7 +60,7 @@ async fn test_reset_single_group() -> Result<()> {
     add_task(shared, "failed").await?;
     add_task_to_group(shared, "sleep 60", "test_2").await?;
     add_task_to_group(shared, "sleep 60", "test_3").await?;
-    wait_for_task_condition(shared, 2, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 2, Task::is_running).await?;
 
     // Reset only the test_2 of the daemon.
     send_message(
@@ -102,7 +102,7 @@ async fn test_reset_multiple_groups() -> Result<()> {
     add_task(shared, "failed").await?;
     add_task_to_group(shared, "sleep 60", "test_2").await?;
     add_task_to_group(shared, "sleep 60", "test_3").await?;
-    wait_for_task_condition(shared, 2, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 2, Task::is_running).await?;
 
     // Reset only the test_2 of the daemon.
     send_message(
diff --git a/pueue/tests/daemon/integration/restart.rs b/pueue/tests/daemon/integration/restart.rs
index 667db82dd..256ed9c38 100644
--- a/pueue/tests/daemon/integration/restart.rs
+++ b/pueue/tests/daemon/integration/restart.rs
@@ -1,7 +1,7 @@
 use std::path::PathBuf;
 
 use anyhow::Result;
-use pueue_lib::network::message::*;
+use pueue_lib::{network::message::*, task::Task};
 
 use crate::helper::*;
 
@@ -16,7 +16,7 @@ async fn test_restart_in_place() -> Result<()> {
     assert_success(add_task(shared, "sleep 0.1").await?);
 
     // Wait for task 0 to finish.
-    let original_task = wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    let original_task = wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     // Restart task 0 with an extended sleep command with a different path.
     let restart_message = RestartMessage {
@@ -36,7 +36,7 @@ async fn test_restart_in_place() -> Result<()> {
     assert_eq!(state.tasks.len(), 1, "No new task should be created");
 
     // Task 0 should soon be started again
-    let task = wait_for_task_condition(shared, 0, |task| task.is_running()).await?;
+    let task = wait_for_task_condition(shared, 0, Task::is_running).await?;
 
     // The created_at time should be the same, as we updated in place
     assert_eq!(
@@ -64,7 +64,7 @@ async fn test_cannot_restart_running() -> Result<()> {
     assert_success(add_task(shared, "sleep 60").await?);
 
     // Wait for task 0 to finish.
-    let task = wait_for_task_condition(shared, 0, |task| task.is_running()).await?;
+    let task = wait_for_task_condition(shared, 0, Task::is_running).await?;
 
     // Restart task 0 with an extended sleep command.
     let restart_message = RestartMessage {
diff --git a/pueue/tests/daemon/integration/restore.rs b/pueue/tests/daemon/integration/restore.rs
index c79ba3c6a..3993706f7 100644
--- a/pueue/tests/daemon/integration/restore.rs
+++ b/pueue/tests/daemon/integration/restore.rs
@@ -1,5 +1,4 @@
 use anyhow::Result;
-use pretty_assertions::assert_eq;
 
 use pueue_lib::network::message::TaskSelection;
 use pueue_lib::state::GroupStatus;
@@ -21,12 +20,13 @@ async fn test_start_running() -> Result<()> {
     // Boot it up again
     let mut child = standalone_daemon(&settings.shared).await?;
 
-    // Assert that the group is still running.
-    let state = get_state(shared).await?;
-    assert_eq!(
-        state.groups.get(PUEUE_DEFAULT_GROUP).unwrap().status,
-        GroupStatus::Running
-    );
+    assert_group_status(
+        shared,
+        PUEUE_DEFAULT_GROUP,
+        GroupStatus::Running,
+        "Default group should still be running.",
+    )
+    .await?;
 
     child.kill()?;
     Ok(())
@@ -50,12 +50,13 @@ async fn test_start_paused() -> Result<()> {
     // Boot it up again
     let mut child = standalone_daemon(&settings.shared).await?;
 
-    // Assert that the group is still paused.
-    let state = get_state(shared).await?;
-    assert_eq!(
-        state.groups.get(PUEUE_DEFAULT_GROUP).unwrap().status,
-        GroupStatus::Paused
-    );
+    assert_group_status(
+        shared,
+        PUEUE_DEFAULT_GROUP,
+        GroupStatus::Paused,
+        "Default group should still be paused.",
+    )
+    .await?;
 
     child.kill()?;
     Ok(())
diff --git a/pueue/tests/daemon/integration/start.rs b/pueue/tests/daemon/integration/start.rs
index 9fc8dd29a..bfca3375a 100644
--- a/pueue/tests/daemon/integration/start.rs
+++ b/pueue/tests/daemon/integration/start.rs
@@ -1,8 +1,8 @@
 use anyhow::Result;
-use pueue_lib::network::message::*;
-use pueue_lib::task::*;
 use rstest::rstest;
 
+use pueue_lib::{network::message::*, task::*};
+
 use crate::helper::*;
 
 /// Test if explicitly starting tasks and resuming tasks works as intended.
@@ -39,14 +39,14 @@ async fn test_start_tasks(#[case] start_message: StartMessage) -> Result<()> {
     // Wait for task 0 to start on its own.
     // We have to do this, otherwise we'll start task 1/2 beforehand, which prevents task 0 to be
     // started on its own.
-    wait_for_task_condition(shared, 0, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 0, Task::is_running).await?;
 
     // Start tasks 1 and 2 manually
     start_tasks(shared, TaskSelection::TaskIds(vec![1, 2])).await?;
 
     // Wait until all tasks are running
     for id in 0..3 {
-        wait_for_task_condition(shared, id, |task| task.is_running()).await?;
+        wait_for_task_condition(shared, id, Task::is_running).await?;
     }
 
     // Pause the whole daemon and wait until all tasks are paused
@@ -63,7 +63,7 @@ async fn test_start_tasks(#[case] start_message: StartMessage) -> Result<()> {
 
     // Ensure all tasks are running
     for id in 0..3 {
-        wait_for_task_condition(shared, id, |task| task.is_running()).await?;
+        wait_for_task_condition(shared, id, Task::is_running).await?;
     }
     Ok(())
 }
diff --git a/pueue/tests/daemon/integration/stashed.rs b/pueue/tests/daemon/integration/stashed.rs
index 915e81d3a..5dcbacaa4 100644
--- a/pueue/tests/daemon/integration/stashed.rs
+++ b/pueue/tests/daemon/integration/stashed.rs
@@ -1,10 +1,8 @@
 use anyhow::{Context, Result};
 use chrono::{DateTime, Local, TimeDelta};
-use pueue_lib::state::GroupStatus;
 use rstest::rstest;
 
-use pueue_lib::network::message::*;
-use pueue_lib::task::*;
+use pueue_lib::{network::message::*, state::GroupStatus, task::*};
 
 use crate::helper::*;
 
@@ -44,7 +42,7 @@ async fn test_enqueued_tasks(#[case] enqueue_at: Option<DateTime<Local>>) -> Res
         .context("Failed to to add task message")?;
 
     // Make sure the task is started after being enqueued
-    wait_for_task_condition(shared, 0, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 0, Task::is_running).await?;
 
     Ok(())
 }
@@ -64,12 +62,17 @@ async fn test_delayed_tasks() -> Result<()> {
     .await?;
     assert_success(response);
 
-    // The task should be added in stashed state for about 1 second.
-    wait_for_task_condition(shared, 0, |task| task.is_stashed()).await?;
+    assert_task_condition(
+        shared,
+        0,
+        Task::is_stashed,
+        "Task should be stashed for about 1 second.",
+    )
+    .await?;
 
     // Make sure the task is started after being automatically enqueued.
     sleep_ms(800).await;
-    wait_for_task_condition(shared, 0, |task| task.is_running()).await?;
+    wait_for_task_condition(shared, 0, Task::is_running).await?;
 
     Ok(())
 }
@@ -96,10 +99,9 @@ async fn test_stash_queued_task() -> Result<()> {
         },
     )
     .await
-    .context("Failed to send STash message")?;
+    .context("Failed to send Stash message")?;
 
-    let task = get_task(shared, 0).await?;
-    assert_eq!(task.status, TaskStatus::Stashed { enqueue_at: None });
+    assert_task_condition(shared, 0, Task::is_stashed, "Task has just been stashed.").await?;
 
     Ok(())
 }
diff --git a/pueue/tests/daemon/integration/worker_environment_variables.rs b/pueue/tests/daemon/integration/worker_environment_variables.rs
index ebc42ae1c..6635e4db2 100644
--- a/pueue/tests/daemon/integration/worker_environment_variables.rs
+++ b/pueue/tests/daemon/integration/worker_environment_variables.rs
@@ -1,6 +1,6 @@
 use anyhow::Result;
 
-use pueue_lib::{network::message::TaskSelection, state::PUEUE_DEFAULT_GROUP};
+use pueue_lib::{network::message::TaskSelection, state::PUEUE_DEFAULT_GROUP, task::Task};
 
 use crate::helper::*;
 
@@ -19,7 +19,7 @@ async fn test_single_worker() -> Result<()> {
     sleep_ms(1000).await;
 
     // Wait for the last task to finish.
-    wait_for_task_condition(shared, 2, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 2, Task::is_done).await?;
 
     // All tasks should have the worker id 0, as the tasks are processed sequentially.
     let state = get_state(shared).await?;
@@ -55,7 +55,7 @@ async fn test_multiple_worker() -> Result<()> {
 
     // Start and wait for the tasks
     start_tasks(shared, TaskSelection::Group("test_3".to_string())).await?;
-    wait_for_task_condition(shared, 2, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 2, Task::is_done).await?;
 
     // The first three tasks should have the same worker id's as the task ids.
     // They ran in parallel and each should have their own worker id assigned.
@@ -71,7 +71,7 @@ async fn test_multiple_worker() -> Result<()> {
         assert_success(add_env_task_to_group(shared, "sleep 0.1", "test_3").await?);
     }
     start_tasks(shared, TaskSelection::Group("test_3".to_string())).await?;
-    wait_for_task_condition(shared, 4, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 4, Task::is_done).await?;
 
     let state = get_state(shared).await?;
     // Task3 gets worker0
@@ -93,7 +93,7 @@ async fn test_worker_for_new_pool() -> Result<()> {
 
     // Add a tasks that finishes instantly.
     assert_success(add_env_task_to_group(shared, "sleep 0.1", "testgroup").await?);
-    wait_for_task_condition(shared, 0, |task| task.is_done()).await?;
+    wait_for_task_condition(shared, 0, Task::is_done).await?;
 
     // The task should have the correct worker id + group.
     let state = get_state(shared).await?;
diff --git a/pueue/tests/helper/asserts.rs b/pueue/tests/helper/asserts.rs
index f05a638ef..1aa578e1e 100644
--- a/pueue/tests/helper/asserts.rs
+++ b/pueue/tests/helper/asserts.rs
@@ -1,11 +1,12 @@
 use anyhow::{bail, Result};
 use assert_matches::assert_matches;
 
-use pueue_lib::network::message::*;
 use pueue_lib::settings::Shared;
 use pueue_lib::state::State;
+use pueue_lib::task::Task;
+use pueue_lib::{network::message::*, state::GroupStatus};
 
-use super::send_message;
+use super::{get_state, send_message};
 
 /// Assert that a message is a successful message.
 pub fn assert_success(message: Message) {
@@ -25,6 +26,57 @@ pub fn assert_failure(message: Message) {
     );
 }
 
+/// A small helper script which pulls the newest state and asserts that a certain condition on a
+/// specific task is given.
+pub async fn assert_task_condition<F>(
+    shared: &Shared,
+    task_id: usize,
+    condition: F,
+    message: &str,
+) -> Result<Task>
+where
+    F: Fn(&Task) -> bool,
+{
+    let state = get_state(shared).await?;
+    match state.tasks.get(&task_id) {
+        Some(task) => {
+            if !condition(task) {
+                bail!("Condition check for task {task_id} failed: {message}");
+            }
+            Ok(task.clone())
+        }
+        None => {
+            bail!("Couldn't find task {task_id} while checking for condition: {message}")
+        }
+    }
+}
+
+/// Make sure a specific group has the expected status.
+pub async fn assert_group_status(
+    shared: &Shared,
+    group_name: &str,
+    expected_status: GroupStatus,
+    message: &str,
+) -> Result<()> {
+    let state = get_state(shared).await?;
+    match state.groups.get(group_name) {
+        Some(group) => {
+            if group.status != expected_status {
+                bail!(
+                    "Group {group_name} doesn't have expected status {expected_status:?}. Found {:?}: {message}",
+                    group.status
+                );
+            }
+            Ok(())
+        }
+        None => {
+            bail!(
+                "Couldn't find group {group_name} while asserting status {expected_status:?}: {message}"
+            )
+        }
+    }
+}
+
 /// Make sure the expected environment variables are set.
 /// This also makes sure, the variables have properly been injected into the processes'
 /// environment.
diff --git a/pueue_lib/src/task.rs b/pueue_lib/src/task.rs
index f4176d9e7..a32154ae7 100644
--- a/pueue_lib/src/task.rs
+++ b/pueue_lib/src/task.rs
@@ -134,6 +134,11 @@ impl Task {
         )
     }
 
+    /// Whether the task is a running, but paused process managed by the TaskHandler.
+    pub fn is_paused(&self) -> bool {
+        matches!(self.status, TaskStatus::Paused { .. })
+    }
+
     /// Whether the task's process finished.
     pub fn is_done(&self) -> bool {
         matches!(self.status, TaskStatus::Done { .. })