refactor: move systemd functions to utils

selfprivacy_api/jobs/upgrade_system.py

@@ -4,10 +4,14 @@ After starting, track the status of the systemd unit and update the Job
 status accordingly.
 """
 import subprocess
-import time
 from selfprivacy_api.utils.huey import huey
 from selfprivacy_api.jobs import JobStatus, Jobs, Job
-from datetime import datetime
+from selfprivacy_api.utils.waitloop import wait_until_true
+from selfprivacy_api.utils.systemd import (
+    get_service_status,
+    get_last_log_lines,
+    ServiceStatus,
+)
 
 START_TIMEOUT = 60 * 5
 START_INTERVAL = 1
@@ -15,6 +19,49 @@ RUN_TIMEOUT = 60 * 60
 RUN_INTERVAL = 5
 
 
+def check_if_started(unit_name: str):
+    """Check if the systemd unit has started"""
+    try:
+        status = get_service_status(unit_name)
+        if status == ServiceStatus.ACTIVE:
+            return True
+        return False
+    except subprocess.CalledProcessError:
+        return False
+
+
+def check_running_status(job: Job, unit_name: str):
+    """Check if the systemd unit is running"""
+    try:
+        status = get_service_status(unit_name)
+        if status == ServiceStatus.INACTIVE:
+            Jobs.update(
+                job=job,
+                status=JobStatus.FINISHED,
+                result="System rebuilt.",
+                progress=100,
+            )
+            return True
+        if status == ServiceStatus.FAILED:
+            Jobs.update(
+                job=job,
+                status=JobStatus.ERROR,
+                error="System rebuild failed.",
+            )
+            return True
+        if status == ServiceStatus.ACTIVE:
+            log_lines = get_last_log_lines(unit_name, 1)
+            Jobs.update(
+                job=job,
+                status=JobStatus.RUNNING,
+                status_text=log_lines[0] if len(log_lines) > 0 else "",
+            )
+            return False
+        return False
+    except subprocess.CalledProcessError:
+        return False
+
+
 @huey.task()
 def rebuild_system_task(job: Job, upgrade: bool = False):
     """Rebuild the system"""
@@ -32,88 +79,39 @@ def rebuild_system_task(job: Job, upgrade: bool = False):
             status=JobStatus.RUNNING,
             status_text="Starting the system rebuild...",
         )
-        # Get current time to handle timeout
-        start_time = datetime.now()
         # Wait for the systemd unit to start
-        while True:
         try:
-                status = subprocess.run(
-                    ["systemctl", "is-active", unit_name],
-                    check=True,
-                    capture_output=True,
-                    text=True,
+            wait_until_true(
+                lambda: check_if_started(unit_name),
+                timeout_sec=START_TIMEOUT,
+                interval=START_INTERVAL,
             )
-                if status.stdout.strip() == "active":
-                    break
-                if (datetime.now() - start_time).total_seconds() > START_TIMEOUT:
+        except TimeoutError:
             Jobs.update(
                 job=job,
                 status=JobStatus.ERROR,
                 error="System rebuild timed out.",
             )
             return
-                time.sleep(START_INTERVAL)
-            except subprocess.CalledProcessError:
-                pass
         Jobs.update(
             job=job,
             status=JobStatus.RUNNING,
             status_text="Rebuilding the system...",
         )
         # Wait for the systemd unit to finish
-        while True:
         try:
-                status = subprocess.run(
-                    ["systemctl", "is-active", unit_name],
-                    check=False,
-                    capture_output=True,
-                    text=True,
+            wait_until_true(
+                lambda: check_running_status(job, unit_name),
+                timeout_sec=RUN_TIMEOUT,
+                interval=RUN_INTERVAL,
             )
-                if status.stdout.strip() == "inactive":
-                    Jobs.update(
-                        job=job,
-                        status=JobStatus.FINISHED,
-                        result="System rebuilt.",
-                        progress=100,
-                    )
-                    break
-                elif status.stdout.strip() == "failed":
-                    Jobs.update(
-                        job=job,
-                        status=JobStatus.ERROR,
-                        error="System rebuild failed.",
-                    )
-                    break
-                elif status.stdout.strip() == "active":
-                    log_line = subprocess.run(
-                        [
-                            "journalctl",
-                            "-u",
-                            unit_name,
-                            "-n",
-                            "1",
-                            "-o",
-                            "cat",
-                        ],
-                        check=False,
-                        capture_output=True,
-                        text=True,
-                    ).stdout.strip()
-                    Jobs.update(
-                        job=job,
-                        status=JobStatus.RUNNING,
-                        status_text=f"{log_line}",
-                    )
-            except subprocess.CalledProcessError:
-                pass
-            if (datetime.now() - start_time).total_seconds() > RUN_TIMEOUT:
+        except TimeoutError:
             Jobs.update(
                 job=job,
                 status=JobStatus.ERROR,
                 error="System rebuild timed out.",
             )
-                break
-            time.sleep(RUN_INTERVAL)
+            return
 
     except subprocess.CalledProcessError as e:
         Jobs.update(

selfprivacy_api/services/bitwarden/__init__.py

@@ -5,9 +5,9 @@ import typing
 
 from selfprivacy_api.jobs import Job, Jobs
 from selfprivacy_api.services.generic_service_mover import FolderMoveNames, move_service
-from selfprivacy_api.services.generic_status_getter import get_service_status
+from selfprivacy_api.utils.systemd import get_service_status
 from selfprivacy_api.services.service import Service, ServiceDnsRecord, ServiceStatus
-from selfprivacy_api.utils import ReadUserData, WriteUserData, get_domain
+from selfprivacy_api.utils import get_domain
 from selfprivacy_api.utils.block_devices import BlockDevice
 import selfprivacy_api.utils.network as network_utils
 from selfprivacy_api.services.bitwarden.icon import BITWARDEN_ICON

selfprivacy_api/services/gitea/__init__.py

@@ -5,9 +5,9 @@ import typing
 
 from selfprivacy_api.jobs import Job, Jobs
 from selfprivacy_api.services.generic_service_mover import FolderMoveNames, move_service
-from selfprivacy_api.services.generic_status_getter import get_service_status
+from selfprivacy_api.utils.systemd import get_service_status
 from selfprivacy_api.services.service import Service, ServiceDnsRecord, ServiceStatus
-from selfprivacy_api.utils import ReadUserData, WriteUserData, get_domain
+from selfprivacy_api.utils import get_domain
 from selfprivacy_api.utils.block_devices import BlockDevice
 import selfprivacy_api.utils.network as network_utils
 from selfprivacy_api.services.gitea.icon import GITEA_ICON

selfprivacy_api/services/jitsimeet/__init__.py

@@ -4,11 +4,11 @@ import subprocess
 import typing
 
 from selfprivacy_api.jobs import Job
-from selfprivacy_api.services.generic_status_getter import (
+from selfprivacy_api.utils.systemd import (
     get_service_status_from_several_units,
 )
 from selfprivacy_api.services.service import Service, ServiceDnsRecord, ServiceStatus
-from selfprivacy_api.utils import ReadUserData, WriteUserData, get_domain
+from selfprivacy_api.utils import get_domain
 from selfprivacy_api.utils.block_devices import BlockDevice
 import selfprivacy_api.utils.network as network_utils
 from selfprivacy_api.services.jitsimeet.icon import JITSI_ICON

selfprivacy_api/services/mailserver/__init__.py

@@ -6,7 +6,7 @@ import typing
 
 from selfprivacy_api.jobs import Job, Jobs
 from selfprivacy_api.services.generic_service_mover import FolderMoveNames, move_service
-from selfprivacy_api.services.generic_status_getter import (
+from selfprivacy_api.utils.systemd import (
     get_service_status_from_several_units,
 )
 from selfprivacy_api.services.service import Service, ServiceDnsRecord, ServiceStatus

selfprivacy_api/services/nextcloud/__init__.py

@@ -4,9 +4,9 @@ import subprocess
 import typing
 from selfprivacy_api.jobs import Job, Jobs
 from selfprivacy_api.services.generic_service_mover import FolderMoveNames, move_service
-from selfprivacy_api.services.generic_status_getter import get_service_status
+from selfprivacy_api.utils.systemd import get_service_status
 from selfprivacy_api.services.service import Service, ServiceDnsRecord, ServiceStatus
-from selfprivacy_api.utils import ReadUserData, WriteUserData, get_domain
+from selfprivacy_api.utils import get_domain
 from selfprivacy_api.utils.block_devices import BlockDevice
 import selfprivacy_api.utils.network as network_utils
 from selfprivacy_api.services.nextcloud.icon import NEXTCLOUD_ICON

selfprivacy_api/services/ocserv/__init__.py

@@ -3,9 +3,8 @@ import base64
 import subprocess
 import typing
 from selfprivacy_api.jobs import Job
-from selfprivacy_api.services.generic_status_getter import get_service_status
+from selfprivacy_api.utils.systemd import get_service_status
 from selfprivacy_api.services.service import Service, ServiceDnsRecord, ServiceStatus
-from selfprivacy_api.utils import ReadUserData, WriteUserData
 from selfprivacy_api.utils.block_devices import BlockDevice
 from selfprivacy_api.services.ocserv.icon import OCSERV_ICON
 import selfprivacy_api.utils.network as network_utils

selfprivacy_api/services/pleroma/__init__.py

@@ -4,10 +4,10 @@ import subprocess
 import typing
 from selfprivacy_api.jobs import Job, Jobs
 from selfprivacy_api.services.generic_service_mover import FolderMoveNames, move_service
-from selfprivacy_api.services.generic_status_getter import get_service_status
+from selfprivacy_api.utils.systemd import get_service_status
 from selfprivacy_api.services.service import Service, ServiceDnsRecord, ServiceStatus
 from selfprivacy_api.services.owned_path import OwnedPath
-from selfprivacy_api.utils import ReadUserData, WriteUserData, get_domain
+from selfprivacy_api.utils import get_domain
 from selfprivacy_api.utils.block_devices import BlockDevice
 import selfprivacy_api.utils.network as network_utils
 from selfprivacy_api.services.pleroma.icon import PLEROMA_ICON

selfprivacy_api/utils/systemd.py

@@ -1,5 +1,6 @@
 """Generic service status fetcher using systemctl"""
 import subprocess
+from typing import List
 
 from selfprivacy_api.services.service import ServiceStatus
 
@@ -58,3 +59,24 @@ def get_service_status_from_several_units(services: list[str]) -> ServiceStatus:
     if ServiceStatus.ACTIVE in service_statuses:
         return ServiceStatus.ACTIVE
     return ServiceStatus.OFF
+
+
+def get_last_log_lines(service: str, lines_count: int) -> List[str]:
+    if lines_count < 1:
+        raise ValueError("lines_count must be greater than 0")
+    try:
+        logs = subprocess.check_output(
+            [
+                "journalctl",
+                "-u",
+                service,
+                "-n",
+                str(lines_count),
+                "-o",
+                "cat",
+            ],
+            shell=False,
+        ).decode("utf-8")
+        return logs.splitlines()
+    except subprocess.CalledProcessError:
+        return []

tests/test_graphql/test_system_nixos_tasks.py

@@ -4,6 +4,7 @@
 import pytest
 
 from selfprivacy_api.jobs import JobStatus, Jobs
+from tests.test_graphql.common import assert_empty, assert_ok, get_data
 
 
 class ProcessMock:
@@ -92,8 +93,7 @@ def test_graphql_system_rebuild_unauthorized(client, fp, action):
             "query": query,
         },
     )
-    assert response.status_code == 200
-    assert response.json().get("data") is None
+    assert_empty(response)
     assert fp.call_count([fp.any()]) == 0
 
 
@@ -111,23 +111,23 @@ def test_graphql_system_rebuild(authorized_client, fp, action, mock_sleep_interv
     fp.register(["systemctl", "start", unit_name])
 
     # Wait for it to start
-    fp.register(["systemctl", "is-active", unit_name], stdout="inactive")
-    fp.register(["systemctl", "is-active", unit_name], stdout="inactive")
-    fp.register(["systemctl", "is-active", unit_name], stdout="active")
+    fp.register(["systemctl", "show", unit_name], stdout="ActiveState=inactive")
+    fp.register(["systemctl", "show", unit_name], stdout="ActiveState=inactive")
+    fp.register(["systemctl", "show", unit_name], stdout="ActiveState=active")
 
     # Check its exectution
-    fp.register(["systemctl", "is-active", unit_name], stdout="active")
+    fp.register(["systemctl", "show", unit_name], stdout="ActiveState=active")
     fp.register(
         ["journalctl", "-u", unit_name, "-n", "1", "-o", "cat"],
         stdout="Starting rebuild...",
     )
 
-    fp.register(["systemctl", "is-active", unit_name], stdout="active")
+    fp.register(["systemctl", "show", unit_name], stdout="ActiveState=active")
     fp.register(
         ["journalctl", "-u", unit_name, "-n", "1", "-o", "cat"], stdout="Rebuilding..."
     )
 
-    fp.register(["systemctl", "is-active", unit_name], stdout="inactive")
+    fp.register(["systemctl", "show", unit_name], stdout="ActiveState=inactive")
 
     response = authorized_client.post(
         "/graphql",
@@ -135,23 +135,11 @@ def test_graphql_system_rebuild(authorized_client, fp, action, mock_sleep_interv
             "query": query,
         },
     )
-    assert response.status_code == 200
-    assert response.json().get("data") is not None
-    assert (
-        response.json()["data"]["system"][f"runSystem{action.capitalize()}"]["success"]
-        is True
-    )
-    assert (
-        response.json()["data"]["system"][f"runSystem{action.capitalize()}"]["message"]
-        is not None
-    )
-    assert (
-        response.json()["data"]["system"][f"runSystem{action.capitalize()}"]["code"]
-        == 200
-    )
+    data = get_data(response)["system"][f"runSystem{action.capitalize()}"]
+    assert_ok(data)
 
     assert fp.call_count(["systemctl", "start", unit_name]) == 1
-    assert fp.call_count(["systemctl", "is-active", unit_name]) == 6
+    assert fp.call_count(["systemctl", "show", unit_name]) == 6
 
     job_id = response.json()["data"]["system"][f"runSystem{action.capitalize()}"][
         "job"
@@ -176,23 +164,23 @@ def test_graphql_system_rebuild_failed(
     fp.register(["systemctl", "start", unit_name])
 
     # Wait for it to start
-    fp.register(["systemctl", "is-active", unit_name], stdout="inactive")
-    fp.register(["systemctl", "is-active", unit_name], stdout="inactive")
-    fp.register(["systemctl", "is-active", unit_name], stdout="active")
+    fp.register(["systemctl", "show", unit_name], stdout="ActiveState=inactive")
+    fp.register(["systemctl", "show", unit_name], stdout="ActiveState=inactive")
+    fp.register(["systemctl", "show", unit_name], stdout="ActiveState=active")
 
     # Check its exectution
-    fp.register(["systemctl", "is-active", unit_name], stdout="active")
+    fp.register(["systemctl", "show", unit_name], stdout="ActiveState=active")
     fp.register(
         ["journalctl", "-u", unit_name, "-n", "1", "-o", "cat"],
         stdout="Starting rebuild...",
     )
 
-    fp.register(["systemctl", "is-active", unit_name], stdout="active")
+    fp.register(["systemctl", "show", unit_name], stdout="ActiveState=active")
     fp.register(
         ["journalctl", "-u", unit_name, "-n", "1", "-o", "cat"], stdout="Rebuilding..."
     )
 
-    fp.register(["systemctl", "is-active", unit_name], stdout="failed")
+    fp.register(["systemctl", "show", unit_name], stdout="ActiveState=failed")
 
     response = authorized_client.post(
         "/graphql",
@@ -200,23 +188,11 @@ def test_graphql_system_rebuild_failed(
             "query": query,
         },
     )
-    assert response.status_code == 200
-    assert response.json().get("data") is not None
-    assert (
-        response.json()["data"]["system"][f"runSystem{action.capitalize()}"]["success"]
-        is True
-    )
-    assert (
-        response.json()["data"]["system"][f"runSystem{action.capitalize()}"]["message"]
-        is not None
-    )
-    assert (
-        response.json()["data"]["system"][f"runSystem{action.capitalize()}"]["code"]
-        == 200
-    )
+    data = get_data(response)["system"][f"runSystem{action.capitalize()}"]
+    assert_ok(data)
 
     assert fp.call_count(["systemctl", "start", unit_name]) == 1
-    assert fp.call_count(["systemctl", "is-active", unit_name]) == 6
+    assert fp.call_count(["systemctl", "show", unit_name]) == 6
 
     job_id = response.json()["data"]["system"][f"runSystem{action.capitalize()}"][
         "job"