refactor: nix-collect-garbage is now pure

selfprivacy_api/jobs/nix_collect_garbage.py

@@ -5,8 +5,73 @@ from selfprivacy_api.jobs import Job, JobStatus, Jobs
 from selfprivacy_api.utils.huey import huey
 
 
+def run_nix_store_print_dead():
+    return subprocess.check_output(["nix-store", "--gc", "--print-dead"])
+
+
+def run_nix_collect_garbage():
+    return subprocess.Popen(
+        ["nix-collect-garbage", "-d"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT
+    )
+
+
+def parse_line(line, job: Job):
+    pattern = re.compile(r"[+-]?\d+\.\d+ \w+ freed")
+    match = re.search(
+        pattern,
+        line,
+    )
+
+    if match is None:
+        Jobs.update(
+            job=job,
+            status=JobStatus.FINISHED,
+            progress=100,
+            status_text="Completed with an error",
+            result="We are sorry, result was not found :(",
+        )
+
+    else:
+        Jobs.update(
+            job=job,
+            status=JobStatus.FINISHED,
+            progress=100,
+            status_text="Сleaning completed.",
+            result=f"{match.group(0)} have been cleared",
+        )
+
+
+def stream_process(
+    process,
+    package_equal_to_percent,
+    job: Job,
+):
+    go = process.poll() is None
+    percent = 0
+
+    for line in process.stdout:
+        if "deleting '/nix/store/" in line:
+            percent += package_equal_to_percent
+
+            Jobs.update(
+                job=job,
+                status=JobStatus.RUNNING,
+                progress=int(percent),
+                status_text="Сleaning...",
+            )
+
+        elif "store paths deleted," in line:
+            parse_line(line, job)
+
+    return go
+
+
 @huey.task()
-def nix_collect_garbage(job: Job):
+def nix_collect_garbage(
+    job: Job,
+    run_nix_store=run_nix_store_print_dead,
+    run_nix_collect=run_nix_collect_garbage,
+):  # innocent as a pure function
 
     Jobs.update(
         job=job,
@@ -15,11 +80,19 @@ def nix_collect_garbage(job: Job):
         status_text="Сalculate the number of dead packages...",
     )
 
-    output = subprocess.check_output(
-        ["nix-store --gc --print-dead", "--gc", "--print-dead"]
-    )
+    output = run_nix_store()
 
     dead_packages = len(re.findall("/nix/store/", output.decode("utf-8")))
+
+    if dead_packages == 0:
+        Jobs.update(
+            job=job,
+            status=JobStatus.FINISHED,
+            progress=100,
+            status_text="Nothing to clear",
+            result="System is clear",
+        )
+
     package_equal_to_percent = 100 / dead_packages
 
     Jobs.update(
@@ -29,52 +102,4 @@ def nix_collect_garbage(job: Job):
         status_text=f"Found {dead_packages} packages to remove!",
     )
 
-    def _parse_line(line):
-        pattern = re.compile(r"[+-]?\d+\.\d+ \w+ freed")
-        match = re.search(
-            pattern,
-            line,
-        )
-
-        if match is None:
-            Jobs.update(
-                job=job,
-                status=JobStatus.FINISHED,
-                progress=100,
-                status_text="Completed with an error",
-                result="We are sorry, result was not found :(",
-            )
-
-        else:
-            Jobs.update(
-                job=job,
-                status=JobStatus.FINISHED,
-                progress=100,
-                status_text="Сleaning completed.",
-                result=f"{match.group(0)} have been cleared",
-            )
-
-    def _stream_process(process):
-        go = process.poll() is None
-        percent = 0
-
-        for line in process.stdout:
-            if "deleting '/nix/store/" in line:
-                percent += package_equal_to_percent
-
-                Jobs.update(
-                    job=job,
-                    status=JobStatus.RUNNING,
-                    progress=int(percent),
-                    status_text="Сleaning...",
-                )
-
-            elif "store paths deleted," in line:
-                _parse_line(line)
-
-        return go
-
-    process = subprocess.Popen(
-        ["nix-collect-garbage", "-d"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT
-    )
-    _stream_process(process)
+    stream_process(run_nix_collect, package_equal_to_percent, job)

tests/test_graphql/test_nix_collect_garbage.py

@@ -0,0 +1,32 @@
+# pylint: disable=redefined-outer-name
+# pylint: disable=unused-argument
+# pylint: disable=missing-function-docstring
+
+import pytest
+
+from selfprivacy_api.jobs.nix_collect_garbage import nix_collect_garbage
+
+
+
+
+
+    created_at: datetime.datetime
+    updated_at: datetime.datetime
+    uid: UUID
+    type_id: str
+    name: str
+    description: str
+    status: JobStatus
+
+
+def test_nix_collect_garbage(job(
+    created_at = "2019-12-04",
+    updated_at = "2019-12-04",
+    uid = UUID,
+    type_id = "typeid",
+    name = "name",
+    description: "desc",
+    status = status(CREATED = "CREATED"),
+)):
+
+    assert nix_collect_garbage() is not None