Send raw command string instead of a dictionary

Author: Miauwkeru

flow/record/fieldtypes/__init__.py

@@ -736,10 +736,11 @@ class command(FieldType):
     executable: path | None = None
     args: list[str] | None = None
 
+    _raw: str | None = None
     _path_type: type[path] = None
-    _posix: bool
+    _posix: bool = True
 
-    def __new__(cls, value: str):
+    def __new__(cls, value: str | None):
         if cls is not command:
             return super().__new__(cls)
 
@@ -757,17 +758,12 @@ def __new__(cls, value: str):
         cls = windows_command if windows else posix_command
         return super().__new__(cls)
 
-    def __init__(self, value: str | tuple[str, tuple[str]] | None):
+    def __init__(self, value: str | None):
         if value is None:
             return
 
-        if isinstance(value, str):
-            self.executable, self.args = self._split(value)
-            return
-
-        executable, self.args = value
-        self.executable = self._path_type(executable)
-        self.args = list(self.args)
+        self._raw = value
+        self.executable, self.args = self._split(self._raw)
 
     def __repr__(self) -> str:
         return f"(executable={self.executable!r}, args={self.args})"
@@ -776,30 +772,29 @@ def __eq__(self, other: object) -> bool:
         if isinstance(other, command):
             return self.executable == other.executable and self.args == other.args
         if isinstance(other, str):
-            return self._join() == other
+            return self._raw == other
         if isinstance(other, (tuple, list)):
             return self.executable == other[0] and self.args == list(other[1:])
 
         return False
 
-    def _split(self, value: str) -> tuple[str, list[str]]:
+    def _split(self, value: str) -> tuple[path, list[str]]:
         executable, *args = shlex.split(value, posix=self._posix)
         executable = executable.strip("'\" ")
 
         return self._path_type(executable), args
 
     def _join(self) -> str:
-        return shlex.join([str(self.executable), *self.args])
+        return self._raw
 
-    def _pack(self) -> tuple[tuple[str, list], str]:
+    def _pack(self) -> tuple[str, int]:
         command_type = TYPE_WINDOWS if isinstance(self, windows_command) else TYPE_POSIX
         if self.executable:
-            _exec, _ = self.executable._pack()
-            return ((_exec, self.args), command_type)
+            return (self._raw, command_type)
         return (None, command_type)
 
     @classmethod
-    def _unpack(cls, data: tuple[tuple[str, tuple] | None, int]) -> command:
+    def _unpack(cls, data: tuple[str | None, int]) -> command:
         _value, _type = data
         if _type == TYPE_WINDOWS:
             return windows_command(_value)
@@ -824,18 +819,8 @@ class windows_command(command):
     _posix = False
     _path_type = windows_path
 
-    def _split(self, value: str) -> tuple[str, list[str]]:
-        executable, args = super()._split(value)
-        if args:
-            args = [" ".join(args)]
+    def _split(self, value: str) -> tuple[path, list[str]]:
+        executable, _args = super()._split(value)
+        args = [" ".join(_args)] if _args else _args
 
         return executable, args
-
-    def _join(self) -> str:
-        arg = f" {self.args[0]}" if self.args else ""
-        executable_str = str(self.executable)
-
-        if " " in executable_str:
-            return f"'{executable_str}'{arg}"
-
-        return f"{executable_str}{arg}"

flow/record/jsonpacker.py

@@ -75,10 +75,7 @@ def pack_obj(self, obj: Any) -> dict | str:
         if isinstance(obj, fieldtypes.path):
             return str(obj)
         if isinstance(obj, fieldtypes.command):
-            return {
-                "executable": obj.executable,
-                "args": obj.args,
-            }
+            return obj._raw
 
         raise TypeError(f"Unpackable type {type(obj)}")
 

tests/_utils.py

@@ -42,3 +42,16 @@ def generate_plain_records(count: int = 100) -> Iterator[Record]:
 
     for i in range(count):
         yield TestRecord(number=i, dt=datetime.datetime.now(datetime.timezone.utc))
+
+
+def generate_command_records(count: int = 100) -> Iterator[Record]:
+    TestRecord = RecordDescriptor(
+        "test/addapter/command",
+        [
+            ("uint32", "number"),
+            ("command", "data"),
+        ],
+    )
+
+    for i in range(count):
+        yield TestRecord(number=i, data=f"/path/to/file {i}")

tests/adapter/test_json.py

@@ -1,24 +1,42 @@
 from __future__ import annotations
 
 import json
-from typing import TYPE_CHECKING
+from enum import Enum, auto
+from typing import TYPE_CHECKING, Callable
 
 import pytest
 
 from flow.record import RecordReader, RecordWriter
-from tests._utils import generate_records
+from tests._utils import generate_command_records, generate_records
 
 if TYPE_CHECKING:
+    from collections.abc import Iterator
     from pathlib import Path
 
+    from flow.record import Record
 
-def test_json_adapter(tmp_path: Path) -> None:
+
+class GeneratorType(Enum):
+    RECORDS = auto()
+    COMMAND = auto()
+
+
+FUNCTIONS = {
+    GeneratorType.RECORDS: generate_records,
+    GeneratorType.COMMAND: generate_command_records,
+}
+
+
+@pytest.mark.parametrize("generator_function", FUNCTIONS.keys())
+def test_json_adapter(generator_function: GeneratorType, tmp_path: Path) -> None:
     json_file = tmp_path.joinpath("records.json")
     record_adapter_path = f"jsonfile://{json_file}"
     writer = RecordWriter(record_adapter_path)
     nr_records = 1337
 
-    for record in generate_records(nr_records):
+    gen_func: Callable[[int], Iterator[Record]] = FUNCTIONS.get(generator_function)
+
+    for record in gen_func(nr_records):
         writer.write(record)
     writer.flush()