winci67

Author: jet-logic

runce/cli.py

@@ -3,6 +3,7 @@
 from sys import stderr, stdout
 from shutil import copyfileobj
 from subprocess import Popen, PIPE, run
+import sys
 from typing import Dict, Any
 from .utils import check_pid, filesizepu, kill_pid, tail_bytes, tail_file
 from .main import Main, flag, arg
@@ -131,6 +132,26 @@ def start(self) -> None:
                     sp.drop(x)
 
 
+def _tail(n: float, u="", out="", tab=None):
+    if u:
+        stdout.buffer.write(tail_bytes(out, int(n)))
+    elif n > 0:
+        if sys.platform.startswith("win"):
+            cmd = [
+                "powershell",
+                "-c",
+                f"Get-Content -Tail {int(n)} '{out}'",
+            ]
+        else:
+            cmd = ["tail", "-n", str(int(n)), out]
+        if tab:
+            with Popen(cmd, stdout=PIPE).stdout as o:
+                for line in o:
+                    stdout.buffer.write(b"\t" + line)
+        else:
+            run(cmd)
+
+
 class Tail(Main):
     """Tail process output."""
 
@@ -152,37 +173,18 @@ def start(self) -> None:
             hf = None
         else:
             hf = format_prep(self.format or r"{pid?}: {name}")
-        lines = filesizepu(self.lines or "10")
+        n, u = filesizepu(self.lines or "10")
         j = 0
         out = "err" if self.err else "out"
 
         for x in Manager().find_names(self.ids, ambiguous, no_record):
             if self.existing and not check_pid(x["pid"]):
                 continue
 
-            j > 1 and lines[0] > 0 and print()
+            j > 1 and n > 0 and print()
             if hf:
                 print(f"{self.p_open}{hf(x)}{self.p_close}", flush=True)
-
-            if lines[1]:
-                stdout.buffer.write(tail_bytes(x[out], int(lines[0])))
-            elif lines[0] > 0:
-                # TODO: pythonify
-                if sys.platform.startswith("win"):
-                    cmd = [
-                        "powershell",
-                        "-c",
-                        f"Get-Content -Tail {int(lines[0])} '{x[out]}'",
-                    ]
-                else:
-                    cmd = ["tail", "-n", str(int(lines[0])), x[out]]
-                if self.tab:
-                    with Popen(cmd, stdout=PIPE).stdout as o:
-                        for line in o:
-                            stdout.buffer.write(b"\t" + line)
-                else:
-                    run(cmd)
-                stdout.flush()
+            _tail(n, u, x[out], self.tab)
             j += 1
 
 
@@ -193,11 +195,11 @@ class Run(Main):
     tail: int = flag("t", "tail", "tail the output with n lines", default=0)
     run_id: str = flag("id", "Unique run identifier", default="")
     cwd: str = flag("Working directory for the command")
-    tail: int = flag(
+    tail: str = flag(
         "t",
         "tail",
         "Tail the output (n lines). Use `-t -1` to print the entire output",
-        default=0,
+        # default=0,
     )
     overwrite: bool = flag("overwrite", "Overwrite existing entry", default=False)
     cmd_after: str = flag("run-after", "Run command after", metavar="command")
@@ -228,13 +230,8 @@ def start(self) -> None:
 
         # Handle tail output
         if self.tail:
-            if self.tail < 0:
-                with open(e["out"], "rb") as f:
-                    copyfileobj(f, stdout.buffer)
-            elif self.tail > 0:
-                for x in tail_file(e["out"], self.tail):
-                    stdout.buffer.write(x)
-                    stdout.buffer.write(b"\n")
+            n, u = filesizepu(self.tail)
+            _tail(n, u, e["out"])
 
         # Run post-command if specified
         if self.cmd_after:

runce/utils.py

@@ -256,17 +256,17 @@ def tail_file(filename="", n=10):
         return f.read().splitlines()[-n:]
 
 
-def filesizepu(s: str) -> tuple[float, str]:
+def filesizepu(s: str) -> tuple[int, str]:
     u = ""
     if s[0].isnumeric():
         q = s.lower()
         if q.endswith("b"):
             q, u = q[0:-1], q[-1]
         for i, v in enumerate("kmgtpezy"):
             if q[-1].endswith(v):
-                return float(q[0:-1]) * (2 ** (10 * (i + 1))), v
-        return float(q), u
-    return float(s), u
+                return int(float(q[0:-1]) * (2 ** (10 * (i + 1)))), v
+        return int(q), u
+    return int(s), u
 
 
 def tail_bytes(filename: str, num_bytes: int):

tests/test_cli.py

@@ -109,7 +109,7 @@ def test_cli(self):
             "run", "--id", "apple", "--", "python", "tests/echo.py", "apple"
         )
 
-        self.assertRegex(o, r"(?xim) \W+ started: \W+ .+ \W+ apple \W+")
+        self.assertRegex(o, r"(?xim) (?:\W+|\A) started: \W+ .+ \W+ apple \W+")
 
         o = self.run_runce(
             "run",
@@ -122,12 +122,12 @@ def test_cli(self):
             "banana",
             "err",
         )
-        self.assertRegex(o, r"(?xim) \W+ started: \W+ .+ \W+ banana \W+")
+        self.assertRegex(o, r"(?xim) (?:\W+|\A) started: \W+ .+ \W+ banana \W+")
 
         o = self.run_runce(
             "run", "--id", "pineapple", "--", "python", "tests/echo.py", "pineapple"
         )
-        self.assertRegex(o, r"(?xim) \W+ started: \W+ .+ \W+ pineapple \W+")
+        self.assertRegex(o, r"(?xim) (?:\W+|\A) started: \W+ .+ \W+ pineapple \W+")
 
         a = [x for x in self.run_runce("status").strip().splitlines()]
 

tests/test_filesizepu.py

@@ -41,7 +41,9 @@ def test_zettabytes(self):
 
     def test_no_unit(self):
         self.assertEqual(filesizepu("1024"), (1024.0, ""))
-        self.assertEqual(filesizepu("3.14"), (3.14, ""))
+        self.assertEqual(
+            filesizepu("1208925819614629174706176"), (1208925819614629174706176, "")
+        )
 
     def test_invalid_input(self):
         with self.assertRaises(ValueError):
@@ -54,6 +56,8 @@ def test_invalid_input(self):
             filesizepu("1BB")
         with self.assertRaises(ValueError):
             filesizepu("9xb")
+        with self.assertRaises(ValueError):
+            filesizepu("3.14")
 
     def test_case_insensitive(self):
         # self.assertEqual(filesizepu("1K"), 1024)