Add commands for register manipulation and command execution; add asy…

…nc subscription; fix defects (#55)

- Add shorter aliases for pub sub
- Introduce FormatterHints
- Add HACK to remove the end-of-document marker from YAML produced by ruamel.yaml
- Update the formatter usage conventions to not emit extra newline at the end
- Use EXIT_CODE_UNSUCCESSFUL
- Suppress errors from PyCyphal during event loop finalization
- Fix `yakut monitor` screen refresh regression.
- Add `yakut register`.
- Add `yakut register-list`.
- Add `yakut register-batch`.
- Add `yakut execute-command`.
- Implement asynchronous subscription in `yakut subscribe` and make it the default option.
- New simpler option `--sync`
- Simplify transfer metadata reporting (related to the future #54 )
* Introduce format selection options --yaml/--json/--tsvh
- Add a new formatter option `AUTO`, which is now the default. It selects between JSON and YAML automatically depending on `isatty(stdout)`. This is needed to enable compatibility with `jq`.
- Close #30 
- Close #52 
- Close #3

.appveyor.yml

@@ -88,7 +88,8 @@ for:
 # https://github.com/appveyor/ci/issues/401#issuecomment-301649481
 # https://www.appveyor.com/docs/packaging-artifacts/#pushing-artifacts-from-scripts
 on_finish:
+- ps: $ErrorActionPreference = 'SilentlyContinue'
 - ps: >
-    $root = Resolve-Path .nox;
+    $root = Resolve-Path .;
     [IO.Directory]::GetFiles($root.Path, '*.log', 'AllDirectories') |
     % { Push-AppveyorArtifact $_ -FileName $_.Substring($root.Path.Length + 1) -DeploymentName to-publish }

setup.cfg

@@ -185,7 +185,9 @@ disable=
     too-many-statements,
     too-many-instance-attributes,
     eval-used,
-    unspecified-encoding
+    unspecified-encoding,
+    not-callable,
+    unbalanced-tuple-unpacking
 
 [pylint.REPORTS]
 output-format=colorized

tests/cmd/call.py

@@ -64,10 +64,10 @@ async def handle_request(
 
     # Invoke the service without discovery and then run the server for a few seconds to let it process the request.
     proc = Subprocess.cli(  # Windows compat: -v blocks stderr pipe on Windows.
-        "--format=json",
+        "-j",
         "call",
         "22",
-        "222:sirius_cyber_corp.PerformLinearLeastSquaresFit",
+        "222:sirius_cyber_corp.performlinearleastsquaresfit",
         "points: [{x: 10, y: 1}, {x: 20, y: 2}]",
         "--priority=SLOW",
         "--with-metadata",
@@ -85,18 +85,18 @@ async def handle_request(
     # Parse the output and validate it.
     parsed = json.loads(stdout)
     print("PARSED RESPONSE:", parsed)
-    assert parsed["222"]["_metadata_"]["priority"] == "slow"
-    assert parsed["222"]["_metadata_"]["source_node_id"] == 22
+    assert parsed["222"]["_meta_"]["priority"] == "slow"
+    assert parsed["222"]["_meta_"]["source_node_id"] == 22
     assert parsed["222"]["slope"] == pytest.approx(0.1)
     assert parsed["222"]["y_intercept"] == pytest.approx(0.0)
 
     # Invoke the service with ID discovery and static type.
     last_metadata = None
     proc = Subprocess.cli(  # Windows compat: -v blocks stderr pipe on Windows.
-        "--format=json",
+        "-j",
         "call",
         "22",
-        "least_squares:sirius_cyber_corp.PerformLinearLeastSquaresFit",
+        "least_squares:sirius_cyber_corp.PERFORMLINEARLEASTSQUARESFIT",
         "points: [{x: 0, y: 0}, {x: 10, y: 3}]",
         "--priority=FAST",
         "--with-metadata",
@@ -115,15 +115,15 @@ async def handle_request(
     # Parse the output and validate it.
     parsed = json.loads(stdout)
     print("PARSED RESPONSE:", parsed)
-    assert parsed["222"]["_metadata_"]["priority"] == "fast"
-    assert parsed["222"]["_metadata_"]["source_node_id"] == 22
+    assert parsed["222"]["_meta_"]["priority"] == "fast"
+    assert parsed["222"]["_meta_"]["source_node_id"] == 22
     assert parsed["222"]["slope"] == pytest.approx(0.3)
     assert parsed["222"]["y_intercept"] == pytest.approx(0.0)
 
     # Invoke the service with full discovery.
     last_metadata = None
     proc = Subprocess.cli(  # Windows compat: -v blocks stderr pipe on Windows.
-        "--format=json",
+        "-j",
         "call",
         "22",
         "least_squares",  # Type not specified -- discovered.
@@ -144,8 +144,8 @@ async def handle_request(
     # Parse the output and validate it.
     parsed = json.loads(stdout)
     print("PARSED RESPONSE:", parsed)
-    assert parsed["222"]["_metadata_"]["priority"] == "nominal"
-    assert parsed["222"]["_metadata_"]["source_node_id"] == 22
+    assert parsed["222"]["_meta_"]["priority"] == "nominal"
+    assert parsed["222"]["_meta_"]["source_node_id"] == 22
     assert parsed["222"]["slope"] == pytest.approx(0.4)
     assert parsed["222"]["y_intercept"] == pytest.approx(0.0)
 
@@ -205,7 +205,7 @@ async def _unittest_call_fixed(transport_factory: TransportFactory, compiled_dsd
 
     # Invoke a fixed port-ID service.
     proc = Subprocess.cli(  # Windows compat: -v blocks stderr pipe on Windows.
-        "--format=json",
+        "-j",
         "call",
         "22",
         "uavcan.node.GetInfo",

tests/cmd/execute_command.py

@@ -0,0 +1,182 @@
+# Copyright (c) 2022 OpenCyphal
+# This software is distributed under the terms of the MIT License.
+# Author: Pavel Kirienko <[email protected]>
+
+from __future__ import annotations
+import asyncio
+from typing import Any, AsyncIterable, Callable, Awaitable
+import json
+import concurrent.futures
+import pytest
+import pycyphal
+from tests.dsdl import OUTPUT_DIR
+from tests.transport import TransportFactory
+from tests.subprocess import execute_cli
+from yakut.util import EXIT_CODE_UNSUCCESSFUL
+
+
+class Remote:
+    def __init__(self, name: str, env: dict[str, str]) -> None:
+        from pycyphal.application import make_registry, make_node, NodeInfo
+        from uavcan.node import ExecuteCommand_1
+
+        self._node = make_node(
+            NodeInfo(name=name),
+            make_registry(environment_variables=env),
+        )
+        self.last_request: ExecuteCommand_1.Request | None = None
+        self.next_response: ExecuteCommand_1.Response | None = None
+
+        async def serve_execute_command(
+            req: ExecuteCommand_1.Request,
+            _meta: pycyphal.presentation.ServiceRequestMetadata,
+        ) -> ExecuteCommand_1.Response | None:
+            # print(self._node, req, _meta, self.next_response, sep="\n\t")
+            self.last_request = req
+            return self.next_response
+
+        self._srv = self._node.get_server(ExecuteCommand_1)
+        self._srv.serve_in_background(serve_execute_command)
+        self._node.start()
+
+    def close(self) -> None:
+        self._srv.close()
+        self._node.close()
+
+
+Runner = Callable[..., Awaitable[Any]]
+
+
+@pytest.fixture
+async def _context(
+    compiled_dsdl: Any,
+    transport_factory: TransportFactory,
+) -> AsyncIterable[tuple[Runner, tuple[Remote, Remote]]]:
+    asyncio.get_running_loop().slow_callback_duration = 10.0
+    _ = compiled_dsdl
+    remote_nodes = (
+        Remote(f"remote_10", env=transport_factory(10).environment),
+        Remote(f"remote_11", env=transport_factory(11).environment),
+    )
+    background_executor = concurrent.futures.ThreadPoolExecutor()
+
+    async def run(*args: str) -> tuple[int, Any]:
+        def call() -> tuple[int, Any]:
+            status, stdout, _stderr = execute_cli(
+                "cmd",
+                *args,
+                environment_variables={
+                    **transport_factory(100).environment,
+                    "YAKUT_PATH": str(OUTPUT_DIR),
+                },
+                timeout=10,
+                ensure_success=False,
+            )
+            return status, json.loads(stdout) if stdout else None
+
+        return await asyncio.get_running_loop().run_in_executor(background_executor, call)
+
+    yield run, remote_nodes
+    for rn in remote_nodes:
+        rn.close()
+    await asyncio.sleep(1.0)
+
+
+@pytest.mark.asyncio
+async def _unittest_basic(_context: tuple[Runner, tuple[Remote, Remote]]) -> None:
+    from uavcan.node import ExecuteCommand_1
+
+    run, (remote_10, remote_11) = _context
+
+    # SUCCESS
+    remote_10.next_response = ExecuteCommand_1.Response(status=0)
+    remote_11.next_response = ExecuteCommand_1.Response(status=0)
+    assert await run("10-12", "restart", "--timeout=3") == (
+        0,
+        {
+            "10": {"status": 0},
+            "11": {"status": 0},
+        },
+    )
+    assert await run("10-12", "111", "COMMAND ARGUMENT", "--timeout=3") == (
+        0,
+        {
+            "10": {"status": 0},
+            "11": {"status": 0},
+        },
+    )
+    assert (
+        remote_10.last_request
+        and remote_10.last_request.command == 111
+        and remote_10.last_request.parameter.tobytes().decode() == "COMMAND ARGUMENT"
+    )
+    assert (
+        remote_11.last_request
+        and remote_11.last_request.command == 111
+        and remote_11.last_request.parameter.tobytes().decode() == "COMMAND ARGUMENT"
+    )
+
+    # REMOTE ERROR; PROPAGATED AND IGNORED
+    remote_10.next_response = ExecuteCommand_1.Response(status=100)
+    remote_11.next_response = ExecuteCommand_1.Response(status=200)
+    assert await run("10-12", "restart", "--timeout=3") == (
+        EXIT_CODE_UNSUCCESSFUL,
+        {
+            "10": {"status": 100},
+            "11": {"status": 200},
+        },
+    )
+    assert await run("10-12", "123", "--expect=100,200", "--timeout=3") == (
+        0,
+        {
+            "10": {"status": 100},
+            "11": {"status": 200},
+        },
+    )
+    assert remote_10.last_request and remote_10.last_request.command == 123
+    assert remote_11.last_request and remote_11.last_request.command == 123
+
+    # ONE TIMED OUT; ERROR PROPAGATED AND IGNORED
+    remote_10.next_response = None
+    remote_11.next_response = ExecuteCommand_1.Response(status=0)
+    assert await run("10-12", "123", "--timeout=3") == (
+        EXIT_CODE_UNSUCCESSFUL,
+        {
+            "10": None,
+            "11": {"status": 0},
+        },
+    )
+    assert await run("10-12", "123", "--expect") == (
+        0,
+        {
+            "10": None,
+            "11": {"status": 0},
+        },
+    )
+
+    # FLAT OUTPUT (NOT GROUPED BY NODE-ID)
+    remote_11.next_response = ExecuteCommand_1.Response(status=210)
+    assert await run("11", "123", "FOO BAR", "--timeout=3") == (
+        EXIT_CODE_UNSUCCESSFUL,
+        {"status": 210},
+    )
+    assert (
+        remote_11.last_request
+        and remote_11.last_request.command == 123
+        and remote_11.last_request.parameter.tobytes().decode() == "FOO BAR"
+    )
+    assert await run("11", "222", "--timeout=3", "--expect=0..256") == (
+        0,
+        {"status": 210},
+    )
+    assert (
+        remote_11.last_request
+        and remote_11.last_request.command == 222
+        and remote_11.last_request.parameter.tobytes().decode() == ""
+    )
+
+    # ERRORS
+    assert (await run("bad"))[0] != 0
+    assert (await run("10", "invalid_command"))[0] != 0
+    assert (await run("10", "99999999999"))[0] != 0  # Bad command code, serialization will fail
+    assert (await run("10", "0", "z" * 1024))[0] != 0  # Bad parameter, serialization will fail

tests/cmd/file_server.py

@@ -170,7 +170,7 @@ def __del__(self) -> None:
         "file-server",
         root,
         f"--plug-and-play={root}/allocation_table.db",
-        f"+U",
+        f"-u",
         environment_variables={
             "UAVCAN__SERIAL__IFACE": serial_broker,
             "UAVCAN__NODE__ID": "42",

tests/cmd/publish/expression.py

@@ -20,7 +20,7 @@ def _unittest_publish_expression_a(compiled_dsdl: typing.Any, serial_broker: str
     }
 
     proc_sub = Subprocess.cli(
-        "--format=json",
+        "-j",
         "sub",
         "7654:uavcan.primitive.array.Real64.1.0",
         environment_variables=env,
@@ -71,7 +71,7 @@ def _unittest_publish_expression_b(compiled_dsdl: typing.Any, serial_broker: str
     }
 
     proc_sub = Subprocess.cli(
-        "--format=json",
+        "-j",
         "sub",
         "7654:uavcan.primitive.String.1.0",
         environment_variables=env,