diff --git a/.agents/ARCHITECTURE.md b/.agents/ARCHITECTURE.md index 7119901..c916144 100644 --- a/.agents/ARCHITECTURE.md +++ b/.agents/ARCHITECTURE.md @@ -1,17 +1,29 @@ # Architecture (Agent Reference) -Read this before making changes to `ControlInterface`, `Ankaios`, the -protocol layer, or exception handling. - -## Control Interface - -The SDK communicates with the Ankaios agent via a Unix socket at -`/run/ankaios/control_interface` (two FIFOs: `input` and `output`). Messages -are length-delimited protobuf (`_control_api` wrapping `_ank_base`). - -`ControlInterface` runs a background reader thread that deserializes -incoming messages and dispatches them to `Ankaios` via callbacks. `Ankaios` -routes responses to the correct caller using a request-ID queue. +Read this before making changes to `ControlInterfaceConnection`, +`CommandInterfaceConnection`, `Ankaios`, the protocol layer, or exception +handling. + +## Connections + +`Ankaios` talks to Ankaios through one of two interchangeable connections, +both implementing the `Connection` abstract base class +(`ankaios_sdk/_components/connection/connection.py`), picked via +`ConnectionType` at construction time: + +- **Control Interface** (`ConnectionType.CONTROL_INTERFACE`, default) — used + from inside an Ankaios-managed workload. Communicates via named pipes at + `/run/ankaios/control_interface` (two FIFOs: `input` and `output`). + Messages are length-delimited protobuf (`_control_api` wrapping + `_ank_base`). Implemented by `ControlInterfaceConnection`. +- **Command Interface** (`ConnectionType.COMMAND_INTERFACE`) — used from + outside a workload, connecting directly to the Ankaios server over gRPC. + Only available if the SDK was installed with the `command` extra. Implemented + by `CommandInterfaceConnection`. + +Both run a background reader thread that deserializes incoming messages and +dispatches them to `Ankaios` via callbacks. `Ankaios` routes responses to the +correct caller using a request-ID queue. `Ankaios` is the primary entry point, typically used as a context manager: diff --git a/AGENTS.md b/AGENTS.md index 16714cf..cf34a3a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -24,5 +24,5 @@ Read these only when they apply to the task at hand: before making any code or test change: API compatibility, coverage philosophy, lint/PEP 8 enforcement, generated proto file handling. - [.agents/ARCHITECTURE.md](.agents/ARCHITECTURE.md) — required before - touching `ControlInterface`, `Ankaios`, the protocol layer, or exception - handling. + touching `ControlInterfaceConnection`, `CommandInterfaceConnection`, + `Ankaios`, the protocol layer, or exception handling. diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index e2e510b..217f44e 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -47,7 +47,7 @@ The [tools/](tools/) folder contains helper scripts for specific tasks — see [ - Tests mirror the `ankaios_sdk/_components/` structure - Use `unittest.mock.patch` / `MagicMock` for all external dependencies - Each test module exports a `generate_test_()` helper for fixtures -- Accessing private members in tests (e.g. `ankaios._control_interface`) is normal +- Accessing private members in tests (e.g. `ankaios._connection`) is normal **Typical test setup pattern:** @@ -56,12 +56,13 @@ from unittest.mock import patch, PropertyMock from ankaios_sdk import Ankaios, ControlInterfaceState def generate_test_ankaios() -> Ankaios: - with patch("ankaios_sdk.ControlInterface.connect"), patch( - "ankaios_sdk.ControlInterface.connected", new_callable=PropertyMock + with patch("ankaios_sdk.ControlInterfaceConnection.connect"), patch( + "ankaios_sdk.ControlInterfaceConnection.connected", + new_callable=PropertyMock, ) as mock_connected: mock_connected.return_value = True ankaios = Ankaios() - ankaios._control_interface._state = ControlInterfaceState.CONNECTED + ankaios._connection._state = ControlInterfaceState.CONNECTED return ankaios ``` diff --git a/README.md b/README.md index e9f31ea..891644a 100644 --- a/README.md +++ b/README.md @@ -69,6 +69,8 @@ are using. For information regarding versioning, please refer to this table: After installation, you can use the Ankaios SDK to configure and run workloads and request the state of the Ankaios system and the connected agents. +### Connecting over The Control Interface + The following example assumes that the code is running in a managed by Ankaios workload with configured control interface access: @@ -134,6 +136,33 @@ with Ankaios() as ankaios: [workload_name][workload_id].state)) ``` +### Connecting over The Command Interface + +To connect to an Ankaios server directly from outside a workload (e.g. from +a CI job or a management tool), use the Command Interface instead, which +uses a direct gRPC connection: + +```python +from ankaios_sdk import Ankaios, ConnectionType + +with Ankaios( + connection_type=ConnectionType.COMMAND_INTERFACE, + server_url="http://127.0.0.1:25551", +) as ankaios: + ... +``` + +This requires the `command` extra: + +```sh +pip install ankaios-sdk[command] +``` + +For mTLS-secured connections, also pass `ca_pem`, `crt_pem` and `key_pem` +(the PEM-encoded CA certificate, client certificate and client key content). + +### Resources + For more details, please visit: * [Ankaios documentation](https://eclipse-ankaios.github.io/ankaios/latest/) diff --git a/ankaios_sdk/_components/__init__.py b/ankaios_sdk/_components/__init__.py index dfc71b7..619f305 100644 --- a/ankaios_sdk/_components/__init__.py +++ b/ankaios_sdk/_components/__init__.py @@ -44,7 +44,7 @@ from .manifest import * from .log_campaign import * from .event_campaign import * -from .control_interface import * +from .connection import * from .file import * __all__ = [name for name in globals() if not name.startswith("_")] diff --git a/ankaios_sdk/_components/connection/__init__.py b/ankaios_sdk/_components/connection/__init__.py new file mode 100644 index 0000000..f4778d5 --- /dev/null +++ b/ankaios_sdk/_components/connection/__init__.py @@ -0,0 +1,50 @@ +# Copyright (c) 2026 Elektrobit Automotive GmbH +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://www.apache.org/licenses/LICENSE-2.0. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# SPDX-License-Identifier: Apache-2.0 + +""" +This module initializes the connection package by importing the +Connection abstraction and its interface implementations. + +Imports +------- + +- Connection component: + the abstract base class for a connection to Ankaios. +- ControlInterfaceConnection component: + the control interface (named pipes) implementation of Connection. +- CommandInterfaceConnection component: + the command interface (gRPC server interface) implementation of + Connection. Only available if the 'grpc' extra is installed. +""" + +import types + +from .connection import * +from .control_interface import * + +try: + from .command_interface import * +except ImportError: + # The 'grpc' extra is not installed; CommandInterfaceConnection + # stays unavailable, but the rest of the SDK must still work. + pass + +# A submodule sharing its name with this package (connection/connection.py) +# gets bound as an attribute of the package itself by Python's import +# system. This is not desired, so we remove it from the package's namespace. +__all__ = [ + name + for name, value in globals().items() + if not name.startswith("_") and not isinstance(value, types.ModuleType) +] diff --git a/ankaios_sdk/_components/connection/command_interface.py b/ankaios_sdk/_components/connection/command_interface.py new file mode 100644 index 0000000..1a29914 --- /dev/null +++ b/ankaios_sdk/_components/connection/command_interface.py @@ -0,0 +1,468 @@ +# Copyright (c) 2026 Elektrobit Automotive GmbH +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://www.apache.org/licenses/LICENSE-2.0. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# SPDX-License-Identifier: Apache-2.0 + +""" +This script defines the CommandInterfaceConnection class, implementing the +Connection abstraction over a direct gRPC connection to the Ankaios +server, used to connect to Ankaios from outside a workload. + +Classes +------- + +- :class:`CommandInterfaceConnection`: + Handles the interaction with Ankaios over gRPC. + +Enums +----- + +- :class:`CommandInterfaceState`: + Represents the state of the gRPC connection. + +Usage +----- + +- Create a CommandInterfaceConnection instance, connect and disconnect. + .. code-block:: python + + conn = CommandInterfaceConnection( + "http://127.0.0.1:25551", + ) + conn.connect() + ... + conn.disconnect() +""" + + +__all__ = ["CommandInterfaceConnection", "CommandInterfaceState"] + + +import queue +import threading +import time +from enum import Enum +from typing import Callable, Optional + +import grpc + +from ..._protos import grpc_api_pb2 as _grpc_api +from ..._protos import grpc_api_pb2_grpc as _grpc_api_grpc +from ...exceptions import ConnectionException +from ...utils import ANKAIOS_VERSION +from ..request import Request +from ..response import Response +from .connection import Connection + + +class CommandInterfaceState(Enum): + """The state of the gRPC connection.""" + + INITIALIZED = 1 + "(int): The connection was initialized but not yet accepted." + CONNECTED = 2 + "(int): The connection is established." + TERMINATED = 3 + "(int): The connection is terminated." + RECONNECTING = 4 + "(int): A lost connection is being retried." + + def __str__(self) -> str: + """ + Returns the string representation of the state. + + :returns: The state as a string. + :rtype: str + """ + return self.name + + +# pylint: disable=too-many-instance-attributes +class CommandInterfaceConnection(Connection): + """ + This class handles the interaction with an Ankaios server over the + command interface: a direct gRPC connection to the server, + playing the commander role via the CommandConnection service. + + The initial :func:`connect` attempt is never retried. Once a + connection has been established, losing it is treated as + transient: it is retried every ``RECONNECT_INTERVAL`` seconds + until it succeeds or :func:`disconnect` is called. + """ + + RECONNECT_INTERVAL = 0.5 + "(float): Seconds to wait between reconnect attempts." + CHANNEL_READY_TIMEOUT = 5.0 + "(float): Seconds to wait for a single (re)connect attempt." + SERVER_TLS_NAME = "ank-server" + "(str): The domain name Ankaios server certificates are issued for." + _URL_SCHEME_PREFIXES = ("http://", "https://") + "(tuple): The URL scheme prefixes supported by the connection." + + # pylint: disable=too-many-arguments, too-many-positional-arguments + def __init__( + self, + server_url: str, + add_response_callback: Callable, + add_log_callback: Callable, + add_event_callback: Callable, + *, + ca_pem: Optional[str] = None, + crt_pem: Optional[str] = None, + key_pem: Optional[str] = None, + ) -> None: + """ + Initialize the CommandInterfaceConnection object. This is used to + interact with an Ankaios server directly over the command interface. + + If none of ca_pem, crt_pem and key_pem are provided, the + connection is a plaintext (insecure) one. If all three are + provided, the connection is mTLS-secured. Providing only + some of them is invalid. + + :param server_url: The URL of the Ankaios server, e.g. + "http://127.0.0.1:25551" (insecure) or + "https://127.0.0.1:25551" (mTLS-secured). + :type server_url: str + :param add_response_callback: The callback function to add + a response to the Ankaios class. + :type add_response_callback: Callable + :param add_log_callback: The callback function to add + a log to the Ankaios class. + :type add_log_callback: Callable + :param add_event_callback: The callback function to add + an event to the Ankaios class. + :type add_event_callback: Callable + :param ca_pem: The PEM-encoded CA certificate content. + :type ca_pem: Optional[str] + :param crt_pem: The PEM-encoded client certificate content. + :type crt_pem: Optional[str] + :param key_pem: The PEM-encoded client private key content. + :type key_pem: Optional[str] + + :raises ValueError: If only some of ca_pem, crt_pem and + key_pem are provided. + """ + provided = ( + ca_pem is not None, + crt_pem is not None, + key_pem is not None, + ) + if any(provided) and not all(provided): + raise ValueError( + "ca_pem, crt_pem and key_pem must all be provided " + "together for a secured connection, or all omitted " + "for an insecure connection." + ) + super().__init__( + add_response_callback, add_log_callback, add_event_callback + ) + self._server_url = server_url + self._ca_pem = ca_pem + self._crt_pem = crt_pem + self._key_pem = key_pem + + # The state of the command interface must not be changed directly. + # Use the change_state method instead. + self._state = CommandInterfaceState.TERMINATED + self._state_lock = threading.Lock() + # Serializes connect() / disconnect() so the lifecycle transitions + # and the resources they own cannot interleave. + self._lifecycle_lock = threading.Lock() + # Guards _channel, _call and _write_queue, which are replaced by + # the reconnect loop (reader thread) and read/torn down by + # disconnect() and write_request() (caller thread). + self._stream_lock = threading.Lock() + self._channel: Optional[grpc.Channel] = None + self._call = None + self._write_queue: "queue.Queue" = queue.Queue() + self._reader_thread: Optional[threading.Thread] = None + + @property + def connected(self) -> bool: + """ + Check if the gRPC connection is established. + + :returns: True if connected, False otherwise. + :rtype: bool + """ + return self._state == CommandInterfaceState.CONNECTED + + def connect(self) -> None: + """ + Establish the gRPC connection to the Ankaios server. + + This attempt is never retried by this method; once + established, a later lost connection is retried + transparently every ``RECONNECT_INTERVAL`` seconds. + + :raises ConnectionException: If already connected, or if + the connection could not be established. + """ + with self._lifecycle_lock: + if self._state in ( + CommandInterfaceState.INITIALIZED, + CommandInterfaceState.CONNECTED, + CommandInterfaceState.RECONNECTING, + ): + raise ConnectionException("Already connected.") + + # Only change the state once past the point where connecting + # can still fail, so a failed attempt leaves the connection + # exactly as it was and free to retry via a plain connect(). + call = self._open_stream() + self.change_state(CommandInterfaceState.INITIALIZED) + + self._reader_thread = threading.Thread( + target=self._read_from_grpc, args=(call,), daemon=True + ) + self._reader_thread.start() + self.change_state(CommandInterfaceState.CONNECTED) + self._logger.debug("Connected to the Ankaios server over gRPC.") + + def disconnect(self) -> None: + """ + Disconnect from the gRPC connection. + """ + with self._lifecycle_lock: + if self._state == CommandInterfaceState.TERMINATED: + self._logger.debug("Already disconnected.") + return + + self._logger.debug("Disconnecting..") + self.change_state(CommandInterfaceState.TERMINATED) + with self._stream_lock: + call = self._call + if call is not None: + call.cancel() + if self._reader_thread is not None: + self._reader_thread.join(timeout=2) + if self._reader_thread.is_alive(): + self._logger.error("Reader thread did not stop.") + self._reader_thread = None + with self._stream_lock: + if self._channel is not None: + self._channel.close() + self._channel = None + self._call = None + + def write_request(self, request: Request) -> None: + """ + Sends the request through the gRPC connection. + + :param request: The request object to be written. + :type request: Request + + :raises ConnectionException: If not connected. + """ + with self._state_lock: + if self._state != CommandInterfaceState.CONNECTED: + self._logger.error( + "Could not write to the gRPC connection, not connected." + ) + raise ConnectionException( + "Could not write to the gRPC connection, not connected." + ) + # Held so the queue cannot be replaced by a concurrent reconnect + # between reading the reference and putting onto it. + with self._stream_lock: + self._write_queue.put( + _grpc_api.ToServer(request=request._to_proto()) + ) + + def _read_from_grpc(self, call) -> None: + """ + Reads continuously from the gRPC bidi stream. This is meant + to be run in a separate thread. If the connection is lost + after having been established, it is retried every + ``RECONNECT_INTERVAL`` seconds until it succeeds or + :func:`disconnect` is called. + + :param call: The bidi call to read FromServer messages from. + """ + while True: + try: + for from_server in call: + self._handle_from_server(from_server) + except grpc.RpcError as e: + if self._state == CommandInterfaceState.TERMINATED: + # disconnect() already cancelled the call itself; + # this is the expected, self-inflicted result. + self._logger.debug( + "gRPC connection cancelled by disconnect(): '%s'", e + ) + return + self._logger.warning( + "Error while reading from the gRPC connection: '%s'", + e, + ) + else: + # The stream ended without an error; still need to know + # whether that was disconnect()'s doing before retrying. + if self._state == CommandInterfaceState.TERMINATED: + return + + self.change_state(CommandInterfaceState.RECONNECTING) + self._logger.warning( + "Lost connection to the Ankaios server, attempting to " + "reconnect every %ss..", + self.RECONNECT_INTERVAL, + ) + call = self._reconnect() + if call is None: + return + + def _handle_from_server(self, from_server) -> None: + """ + Handles a decoded FromServer message and dispatches to the + appropriate callback. + + :param from_server: The decoded FromServer message. + """ + response_type = from_server.WhichOneof("FromServerEnum") + if response_type == "response": + response = Response._from_ank_base_response(from_server.response) + self._dispatch_response(response) + elif response_type == "serverHello": + self._logger.debug("Received server hello.") + else: + self._logger.warning( + "Received unexpected message from the Ankaios server: '%s'", + response_type, + ) + + def _reconnect(self): + """ + Retries opening the gRPC stream every ``RECONNECT_INTERVAL`` + seconds until it succeeds or :func:`disconnect` is called. + + :returns: The newly opened call, or None if disconnect() was + called while reconnecting. + :rtype: Optional[grpc.Call] + """ + while True: + time.sleep(self.RECONNECT_INTERVAL) + if self._state == CommandInterfaceState.TERMINATED: + return None + try: + call = self._open_stream() + except ConnectionException as e: + self._logger.debug("Reconnect attempt failed: '%s'", e) + continue + self.change_state(CommandInterfaceState.CONNECTED) + self._logger.info("Reconnected to the Ankaios server.") + return call + + def change_state(self, state: CommandInterfaceState) -> None: + """ + Change the state of the connection. + + Unlike the control interface, there is no separate + unrecoverable state here: TERMINATED is both the initial + state and the one disconnect() ends in, and connect() must + be able to leave it again. + + :param state: The new state. + :type state: CommandInterfaceState + """ + with self._state_lock: + if state == self._state: + self._logger.debug("State is already %s.", state) + return + self._state = state + self._logger.debug("State changed to %s.", state) + + def _open_stream(self): + """ + Builds the gRPC channel, sends the initial CommanderHello and + opens the ConnectCommand bidi stream. Used both for the + initial connect and for every reconnect attempt. + + :returns: The opened bidi call, iterable for FromServer + messages. + :rtype: grpc.Call + + :raises ConnectionException: If the channel does not become + ready in time, or the stream could not be opened. + """ + try: + channel = self._build_channel() + grpc.channel_ready_future(channel).result( + timeout=self.CHANNEL_READY_TIMEOUT + ) + except (grpc.FutureTimeoutError, grpc.RpcError) as e: + raise ConnectionException( + f"Could not connect to the Ankaios server: '{e}'" + ) from e + + stub = _grpc_api_grpc.CommandConnectionStub(channel) + write_queue = queue.Queue() + write_queue.put( + _grpc_api.ToServer( + commanderHello=_grpc_api.CommanderHello( + protocolVersion=str(ANKAIOS_VERSION) + ) + ) + ) + + # Held across the replacement so disconnect() and write_request() + # never observe a half-replaced write_queue/channel/call. + with self._stream_lock: + self._write_queue = write_queue + call = stub.ConnectCommand(self._request_iterator()) + self._channel = channel + self._call = call + return call + + def _request_iterator(self): + """ + Generator yielding ToServer messages queued via + :func:`write_request` (and the initial hello), until + :func:`disconnect` cancels the call. + """ + write_queue = self._write_queue + while True: + yield write_queue.get() + + def _build_channel(self) -> grpc.Channel: + """ + Builds the (optionally mTLS-secured) gRPC channel used to + reach the Ankaios server. + + :returns: The gRPC channel. + :rtype: grpc.Channel + """ + target = self._grpc_target() + if self._ca_pem is None: + return grpc.insecure_channel(target) + + credentials = grpc.ssl_channel_credentials( + root_certificates=self._ca_pem.encode(), + private_key=self._key_pem.encode(), + certificate_chain=self._crt_pem.encode(), + ) + # Ankaios server certificates are always issued for this + # domain name, regardless of the actual connection address. + options = (("grpc.ssl_target_name_override", self.SERVER_TLS_NAME),) + return grpc.secure_channel(target, credentials, options=options) + + def _grpc_target(self) -> str: + """ + Returns server_url as a bare host:port gRPC channel target. + + :returns: The channel target. + :rtype: str + """ + for prefix in self._URL_SCHEME_PREFIXES: + if self._server_url.startswith(prefix): + return self._server_url[len(prefix):] + return self._server_url diff --git a/ankaios_sdk/_components/connection/connection.py b/ankaios_sdk/_components/connection/connection.py new file mode 100644 index 0000000..0d61adf --- /dev/null +++ b/ankaios_sdk/_components/connection/connection.py @@ -0,0 +1,160 @@ +# Copyright (c) 2026 Elektrobit Automotive GmbH +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://www.apache.org/licenses/LICENSE-2.0. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# SPDX-License-Identifier: Apache-2.0 + +""" +This script defines the Connection abstract base class, which specifies +the common interface implemented by every way this SDK can talk to +Ankaios (the control interface, used from inside a workload, or a +direct gRPC connection to the Ankaios server, used from outside a +workload). + +Classes +------- + +- :class:`Connection`: + Abstract base class for a connection to Ankaios. + +Enums +----- + +- :class:`ConnectionType`: + Represents which connection the Ankaios class should use. +""" + + +__all__ = ["Connection", "ConnectionType"] + + +from abc import ABC, abstractmethod +from enum import Enum +from typing import Callable + +from ...utils import get_logger +from ..request import Request +from ..response import Response, ResponseType + + +class ConnectionType(Enum): + """Represents which connection the Ankaios class should use.""" + + CONTROL_INTERFACE = 1 + "(int): Connect via the control interface (named pipes)." + COMMAND_INTERFACE = 2 + "(int): Connect via the command interface (direct gRPC)." + + def __str__(self) -> str: + """ + Returns the string representation of the connection type. + + :returns: The connection type as a string. + :rtype: str + """ + return self.name + + +class Connection(ABC): + """ + Abstract base class that defines the common interface for every + connection to Ankaios, so that the :class:`Ankaios` class can work + identically regardless of which connection implementation is used. + """ + + def __init__( + self, + add_response_callback: Callable, + add_log_callback: Callable, + add_event_callback: Callable, + ) -> None: + """ + Stores the callbacks shared by every connection implementation + for forwarding responses, logs and events to the Ankaios class. + + :param add_response_callback: The callback function to add + a response to the Ankaios class. + :type add_response_callback: Callable + :param add_log_callback: The callback function to add + a log to the Ankaios class. + :type add_log_callback: Callable + :param add_event_callback: The callback function to add + an event to the Ankaios class. + :type add_event_callback: Callable + """ + self._add_response_callback = add_response_callback + self._add_log_callback = add_log_callback + self._add_event_callback = add_event_callback + self._logger = get_logger() + + def _dispatch_response(self, response: Response) -> bool: + """ + Routes a decoded response to the callback matching its content + type: logs and events go to their own callbacks, every other + type goes to the generic response callback. Shared by every + connection implementation, since this routing has nothing to + do with how the response was received. + + :param response: The decoded response to dispatch. + :type response: Response + + :returns: False if the response went to the generic response + callback (some content types still need extra handling by + the caller, e.g. control interface handshake state); + True if it was already fully handled here. + :rtype: bool + """ + if response.content_type in ( + ResponseType.LOGS_ENTRY, + ResponseType.LOGS_STOP_RESPONSE, + ): + self._add_log_callback( + response.get_request_id(), response.content + ) + return True + if response.content_type == ResponseType.EVENT_RESPONSE: + self._add_event_callback( + response.get_request_id(), response.content + ) + return True + self._add_response_callback(response) + return False + + @property + @abstractmethod + def connected(self) -> bool: + """ + Check if the connection is established. + + :returns: True if connected, False otherwise. + :rtype: bool + """ + + @abstractmethod + def connect(self) -> None: + """ + Establish the connection. + """ + + @abstractmethod + def disconnect(self) -> None: + """ + Tear down the connection. + """ + + @abstractmethod + def write_request(self, request: Request) -> None: + """ + Send a request through the connection. + + :param request: The request object to be sent. + :type request: Request + """ diff --git a/ankaios_sdk/_components/control_interface.py b/ankaios_sdk/_components/connection/control_interface.py similarity index 84% rename from ankaios_sdk/_components/control_interface.py rename to ankaios_sdk/_components/connection/control_interface.py index 7608eee..ba9e6fd 100644 --- a/ankaios_sdk/_components/control_interface.py +++ b/ankaios_sdk/_components/connection/control_interface.py @@ -13,40 +13,41 @@ # SPDX-License-Identifier: Apache-2.0 """ -This script defines the ControlInterface class that handles the writing -and reading of data to and from the Ankaios control interface. +This script defines the ControlInterfaceConnection class that handles +the writing and reading of data to and from the Ankaios control +interface. Classes ------- -- :class:`ControlInterface`: +- :class:`ControlInterfaceConnection`: Handles the interaction with the Ankaios control interface. Enums ----- - :class:`ControlInterfaceState`: - Represents the state of the control interface. + Represents the state of the control interface connection. Usage ----- -- Create a Control Interface instance, connect and disconnect. +- Create a ControlInterfaceConnection instance, connect and disconnect. .. code-block:: python - ci = ControlInterface() + ci = ControlInterfaceConnection() ci.connect() ... ci.disconnect() -- Change the state of the control interface. +- Change the state of the control interface connection. .. code-block:: python ci.change_state(ControlInterfaceState.TERMINATED) """ -__all__ = ["ControlInterface", "ControlInterfaceState"] +__all__ = ["ControlInterfaceConnection", "ControlInterfaceState"] import os @@ -57,15 +58,16 @@ from google.protobuf.internal.encoder import _VarintBytes from google.protobuf.internal.decoder import _DecodeVarint -from .._protos import _control_api -from .request import Request -from .response import Response, ResponseException, ResponseType -from ..exceptions import ControlInterfaceException, ConnectionClosedException -from ..utils import DEFAULT_CONTROL_INTERFACE_PATH, get_logger, ANKAIOS_VERSION +from ..._protos import _control_api +from ..request import Request +from ..response import Response, ResponseException, ResponseType +from ...exceptions import ConnectionException, ConnectionClosedException +from ...utils import DEFAULT_CONTROL_INTERFACE_PATH, ANKAIOS_VERSION +from .connection import Connection class ControlInterfaceState(Enum): - """The state of the control interface.""" + """The state of the control interface connection.""" INITIALIZED = 1 "(int): Connection initialized state." @@ -89,7 +91,7 @@ def __str__(self) -> str: # pylint: disable=too-many-instance-attributes -class ControlInterface: +class ControlInterfaceConnection(Connection): """ This class handles the interaction with the Ankaios control interface. It provides methods to send and receive data to and from the control @@ -109,7 +111,7 @@ def __init__( add_event_callback: Callable, ) -> None: """ - Initialize the ControlInterface object. This is used + Initialize the ControlInterfaceConnection object. This is used to interact with the control interface. :param add_response_callback: The callback function to add @@ -122,6 +124,9 @@ def __init__( an event to the Ankaios class. :type add_event_callback: Callable """ + super().__init__( + add_response_callback, add_log_callback, add_event_callback + ) self._input_file = None self._output_file = None # The state of the control interface must not be changed directly. @@ -140,12 +145,6 @@ def __init__( self._read_thread = None self._disconnect_event = threading.Event() - self._add_response_callback = add_response_callback - self._add_log_callback = add_log_callback - self._add_event_callback = add_event_callback - - self._logger = get_logger() - @property def connected(self) -> bool: """ @@ -161,26 +160,26 @@ def connect(self) -> None: Connect to the control interface by starting to read from the input fifo and opening the output fifo. - :raises ControlInterfaceException: If an error occurred. + :raises ConnectionException: If an error occurred. """ with self._lifecycle_lock: if self._state in [ ControlInterfaceState.INITIALIZED, ControlInterfaceState.CONNECTED, ]: - raise ControlInterfaceException("Already connected.") + raise ConnectionException("Already connected.") if not os.path.exists( f"{self.ANKAIOS_CONTROL_INTERFACE_BASE_PATH}/input" ): - raise ControlInterfaceException( + raise ConnectionException( "Control interface input fifo does not exist." ) if not os.path.exists( f"{self.ANKAIOS_CONTROL_INTERFACE_BASE_PATH}/output" ): - raise ControlInterfaceException( + raise ConnectionException( "Control interface output fifo does not exist." ) @@ -194,7 +193,7 @@ def connect(self) -> None: self._logger.error( "Error while opening output fifo: %s", e ) - raise ControlInterfaceException( + raise ConnectionException( "Error while opening output fifo." ) from e @@ -284,7 +283,7 @@ def _read_from_control_interface(self) -> None: This is meant to be run in a separate thread. The responses are then sent to the Ankaios class to be handled. - :raises ControlInterfaceException: If an error occurs + :raises ConnectionException: If an error occurs while reading the fifo. """ # The pragma: no cover is used on small checks that are not expected @@ -306,7 +305,7 @@ def _read_from_control_interface(self) -> None: # the current thread). self._disconnect_event.set() self._cleanup() - raise ControlInterfaceException( + raise ConnectionException( "Error while opening input fifo." ) from e os.set_blocking(self._input_file.fileno(), False) @@ -352,7 +351,7 @@ def _read_from_control_interface(self) -> None: msg_buf += next_byte try: - response = Response(bytes(msg_buf)) + response = self._decode_response(bytes(msg_buf)) except ResponseException as e: # pragma: no cover self._logger.error("Error while reading: %s", e) continue @@ -366,6 +365,42 @@ def _read_from_control_interface(self) -> None: # race a concurrent disconnect() touching the same handle. self._cleanup() + @staticmethod + def _decode_response(message_buffer: bytes) -> Response: + """ + Decodes a message read from the control interface's own + envelope (a length-delimited `_control_api.FromAnkaios` + message) into a Response, owning the control-interface- + specific envelope unwrapping so that Response itself only + needs to know about the shared ank_base.Response payload. + + :param message_buffer: The received message buffer. + :type message_buffer: bytes + + :returns: The decoded Response object. + :rtype: Response + + :raises ResponseException: If there is an error parsing the + message buffer, or if it contains none of the expected + variants. + """ + from_ankaios = _control_api.FromAnkaios() + try: + from_ankaios.ParseFromString(message_buffer) + except Exception as e: + raise ResponseException(f"Parsing error: '{e}'") from e + if from_ankaios.HasField("response"): + return Response._from_ank_base_response(from_ankaios.response) + if from_ankaios.HasField("controlInterfaceAccepted"): + return Response._control_interface_accepted() + if from_ankaios.HasField("connectionClosed"): + return Response._connection_closed( + from_ankaios.connectionClosed.reason + ) + raise ResponseException( # pragma: no cover + "Invalid response type." + ) + def _handle_response(self, response: Response) -> None: """ Handle the response received from the control interface. @@ -373,7 +408,7 @@ def _handle_response(self, response: Response) -> None: :param response: The response object to handle. :type response: Response - :raises ControlInterfaceException: + :raises ConnectionException: If the response is not in a valid state. :raises ConnectionClosedException: If the connection is closed. """ @@ -403,28 +438,9 @@ def _handle_response(self, response: Response) -> None: # Handle the connected state elif self._state == ControlInterfaceState.CONNECTED: - # Filter out the logs responses - if response.content_type in [ - ResponseType.LOGS_ENTRY, - ResponseType.LOGS_STOP_RESPONSE, - ]: - self._add_log_callback( - response.get_request_id(), response.content - ) + if self._dispatch_response(response): return - # Filter out the events - if response.content_type in [ - ResponseType.EVENT_RESPONSE, - ]: - self._add_event_callback( - response.get_request_id(), response.content - ) - return - - # Send out the response to the Ankaios class - self._add_response_callback(response) - # Check if the response is connection closed in order to # terminate the thread. if response.content_type == ResponseType.CONNECTION_CLOSED: @@ -478,7 +494,7 @@ def _write_to_pipe(self, to_ankaios: _control_api.ToAnkaios) -> None: :param to_ankaios: The ToAnkaios proto message. :type to_ankaios: _control_api.ToAnkaios - :raises ControlInterfaceException: If the output pipe is None. + :raises ConnectionException: If the output pipe is None. """ # Held across the whole write so the length prefix and the payload # cannot be split by another writer, and so the handle cannot be @@ -488,7 +504,7 @@ def _write_to_pipe(self, to_ankaios: _control_api.ToAnkaios) -> None: self._logger.error( "Could not write to pipe, output file handler is None." ) - raise ControlInterfaceException( + raise ConnectionException( "Could not write to pipe, output file handler is None." ) @@ -505,7 +521,7 @@ def write_request(self, request: Request) -> None: :param request: The request object to be written. :type request: Request - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises ConnectionClosedException: If the connection is closed. """ with self._state_lock: @@ -514,7 +530,7 @@ def write_request(self, request: Request) -> None: "Could not write to pipe, connection closed." ) if self._state != ControlInterfaceState.CONNECTED: - raise ControlInterfaceException( + raise ConnectionException( "Could not write to pipe, not connected." ) @@ -528,7 +544,7 @@ def _send_initial_hello(self) -> None: Send an initial hello message with the version to the control interface. - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. """ initial_hello = _control_api.ToAnkaios( hello=_control_api.Hello(protocolVersion=str(ANKAIOS_VERSION)) diff --git a/ankaios_sdk/_components/response.py b/ankaios_sdk/_components/response.py index d59538c..66460ac 100644 --- a/ankaios_sdk/_components/response.py +++ b/ankaios_sdk/_components/response.py @@ -91,6 +91,8 @@ logger = get_logger() +_LOG_RESPONSE_RECEIVED = "Got response of type '%s' with request id '%s'" + class Response: """ @@ -134,8 +136,7 @@ def _parse_response(self) -> None: logger.error("Error parsing the received message: %s", e) raise ResponseException(f"Parsing error: '{e}'") from e if from_ankaios.HasField("response"): - self._response = from_ankaios.response - self._from_proto() + self._from_ank_base(from_ankaios.response) elif from_ankaios.HasField("controlInterfaceAccepted"): self.content_type = ResponseType.CONTROL_INTERFACE_ACCEPTED elif from_ankaios.HasField("connectionClosed"): @@ -146,11 +147,103 @@ def _parse_response(self) -> None: "Invalid response type." ) logger.debug( - "Got response of type '%s' with request id '%s'", + _LOG_RESPONSE_RECEIVED, self.content_type, self.get_request_id(), ) + def _from_ank_base(self, ank_base_response: _ank_base.Response) -> None: + """ + Converts an already-decoded ank_base.Response into this + Response's content. Shared by every connection (control + interface, gRPC), since ank_base.Response carries no + envelope-specific concepts of its own. + + :param ank_base_response: The decoded ank_base Response message. + :type ank_base_response: _ank_base.Response + """ + self._response = ank_base_response + self._from_proto() + + @classmethod + def _from_ank_base_response( + cls, ank_base_response: _ank_base.Response + ) -> "Response": + """ + Creates a Response directly from an already-decoded + ank_base.Response message. Used by every connection + implementation once it has unwrapped its own envelope (e.g. + gRPC's FromServer, or the control interface's FromAnkaios) and + found the actual response payload, so the ank_base-specific + parsing in :func:`_from_proto` is never duplicated. + + :param ank_base_response: The decoded ank_base Response message. + :type ank_base_response: _ank_base.Response + + :returns: The constructed Response object. + :rtype: Response + """ + response = cls.__new__(cls) + response.buffer = None + response.content_type = None + response.content = None + response._from_ank_base(ank_base_response) + logger.debug( + _LOG_RESPONSE_RECEIVED, + response.content_type, + response.get_request_id(), + ) + return response + + @classmethod + def _control_interface_accepted(cls) -> "Response": + """ + Creates a Response representing the control interface's + handshake-accepted message. Carries no ank_base.Response + payload, since this is a concept specific to the control + interface's own envelope. + + :returns: The constructed Response object. + :rtype: Response + """ + response = cls.__new__(cls) + response.buffer = None + response._response = None + response.content_type = ResponseType.CONTROL_INTERFACE_ACCEPTED + response.content = None + logger.debug( + _LOG_RESPONSE_RECEIVED, + response.content_type, + response.get_request_id(), + ) + return response + + @classmethod + def _connection_closed(cls, reason: str) -> "Response": + """ + Creates a Response representing the control interface's + connection-closed message. Carries no ank_base.Response + payload, since this is a concept specific to the control + interface's own envelope. + + :param reason: The reason the connection was closed. + :type reason: str + + :returns: The constructed Response object. + :rtype: Response + """ + response = cls.__new__(cls) + response.buffer = None + response._response = None + response.content_type = ResponseType.CONNECTION_CLOSED + response.content = reason + logger.debug( + _LOG_RESPONSE_RECEIVED, + response.content_type, + response.get_request_id(), + ) + return response + # pylint: disable=too-many-branches def _from_proto(self) -> None: """ diff --git a/ankaios_sdk/ankaios.py b/ankaios_sdk/ankaios.py index b0e9b4f..9fe2d56 100644 --- a/ankaios_sdk/ankaios.py +++ b/ankaios_sdk/ankaios.py @@ -13,14 +13,15 @@ # SPDX-License-Identifier: Apache-2.0 """ -This script defines the Ankaios class for interacting with the -Ankaios control interface. +This script defines the Ankaios class for interacting with Ankaios, +either via the control interface (default, used from inside a +workload) or directly over gRPC (used from outside a workload). Classes ------- - :class:`Ankaios`: - Handles the interaction with the Ankaios control interface. + Handles the interaction with the Ankaios cluster. Usage ----- @@ -28,10 +29,41 @@ - Create an Ankaios object, connect and disconnect from the control interface: .. code-block:: python + from ankaios_sdk import Ankaios + ankaios = Ankaios() ... del ankaios +- Create an Ankaios object, connect and disconnect from the gRPC server + interface: + + .. code-block:: python + + from ankaios_sdk import Ankaios, ConnectionType + + ankaios = Ankaios( + connection_type=ConnectionType.COMMAND_INTERFACE, + server_url="http://127.0.0.1:25551", + ) + ... + del ankaios + +- Create an Ankaios object using a mTLS-secured gRPC connection: + .. code-block:: python + + from ankaios_sdk import Ankaios, ConnectionType + + ankaios = Ankaios( + connection_type=ConnectionType.COMMAND_INTERFACE, + server_url="https://127.0.0.1:25551", + ca_pem=ca_pem, + crt_pem=crt_pem, + key_pem=key_pem, + ) + ... + del ankaios + - Connect and disconnect using a context manager: .. code-block:: python @@ -127,7 +159,7 @@ __all__ = ["Ankaios"] import time -from typing import Union, Callable +from typing import Union, Callable, Optional from datetime import datetime from queue import Queue, Empty @@ -151,7 +183,9 @@ WorkloadInstanceName, WorkloadStateEnum, WorkloadExecutionState, - ControlInterface, + Connection, + ConnectionType, + ControlInterfaceConnection, LogCampaignResponse, LogQueue, LogResponse, @@ -185,7 +219,7 @@ class Ankaios: """ This class is used to interact with the Ankaios using an intuitive API. The class automatically handles the session creation and the requests - and responses sent and received over the Ankaios Control Interface. + and responses sent and received over the underlying connection. :var logging.Logger logger: The logger for the Ankaios class. @@ -194,16 +228,51 @@ class Ankaios: DEFAULT_TIMEOUT = 5.0 "(float): The default timeout, if not manually provided." + # pylint: disable=too-many-arguments, too-many-positional-arguments def __init__( - self, log_level: AnkaiosLogLevel = AnkaiosLogLevel.INFO + self, + connection_type: ConnectionType = ConnectionType.CONTROL_INTERFACE, + log_level: AnkaiosLogLevel = AnkaiosLogLevel.INFO, + *, + server_url: Optional[str] = None, + ca_pem: Optional[str] = None, + crt_pem: Optional[str] = None, + key_pem: Optional[str] = None, ) -> None: """ Initialize the Ankaios object. The logger will be created and - the connection to the control interface will be established. + the connection will be established, either via the control + interface (default, used from inside a workload) or directly + over gRPC to the Ankaios server (used from outside a + workload). + :param connection_type: Which connection to use. + :type connection_type: ConnectionType :param log_level: The log level to be set. :type log_level: AnkaiosLogLevel - + :param server_url: The URL of the Ankaios server, e.g. + "http://127.0.0.1:25551". Required when connection_type + is ConnectionType.COMMAND_INTERFACE, ignored otherwise. + :type server_url: Optional[str] + :param ca_pem: The PEM-encoded CA certificate content, for a + mTLS-secured gRPC connection. Ignored unless + connection_type is ConnectionType.COMMAND_INTERFACE. + :type ca_pem: Optional[str] + :param crt_pem: The PEM-encoded client certificate content, + for a mTLS-secured gRPC connection. Ignored unless + connection_type is ConnectionType.COMMAND_INTERFACE. + :type crt_pem: Optional[str] + :param key_pem: The PEM-encoded client private key content, + for a mTLS-secured gRPC connection. Ignored unless + connection_type is ConnectionType.COMMAND_INTERFACE. + :type key_pem: Optional[str] + + :raises ValueError: If connection_type is + ConnectionType.COMMAND_INTERFACE and server_url is not + provided. + :raises ImportError: If connection_type is + ConnectionType.COMMAND_INTERFACE and the SDK was + installed without the 'grpc' extra. :raises ConnectionClosedException: If the connection is closed at startup. """ @@ -215,27 +284,76 @@ def __init__( self.logger = get_logger() self.set_logger_level(log_level) - # Connect to the control interface - self._control_interface = ControlInterface( - add_response_callback=self._add_response, - add_log_callback=self._add_logs, - add_event_callback=self._add_events, + self._connection = self._create_connection( + connection_type, server_url, ca_pem, crt_pem, key_pem ) - self._control_interface.connect() + self._connection.connect() # Wait for the connection to be established start_time = time.time() - while not self._control_interface.connected: + while not self._connection.connected: if time.time() - start_time > self.DEFAULT_TIMEOUT: - self.logger.error( - "Connection to the control interface timed out." - ) - self._control_interface.disconnect() + self.logger.error("Connection to Ankaios timed out.") + self._connection.disconnect() raise ConnectionClosedException( - "Connection to the control interface timed out." + "Connection to Ankaios timed out." ) time.sleep(0.1) + # pylint: disable=too-many-arguments, too-many-positional-arguments + def _create_connection( + self, + connection_type: ConnectionType, + server_url: Optional[str], + ca_pem: Optional[str], + crt_pem: Optional[str], + key_pem: Optional[str], + ) -> Connection: + """ + Builds the connection matching connection_type, wiring in + this object's response/log/event callbacks. + + :returns: The constructed connection. + :rtype: Connection + + :raises ValueError: If connection_type is + ConnectionType.COMMAND_INTERFACE and server_url is not + provided. + :raises ImportError: If connection_type is + ConnectionType.COMMAND_INTERFACE and the SDK was + installed without the 'grpc' extra. + """ + if connection_type == ConnectionType.COMMAND_INTERFACE: + if server_url is None: + raise ValueError( + "server_url is required when connection_type is " + "ConnectionType.COMMAND_INTERFACE." + ) + try: + # pylint: disable=import-outside-toplevel + from ._components.connection.command_interface import ( + CommandInterfaceConnection, + ) + except ImportError as e: + raise ImportError( + "Command Interface support requires the 'command' extra: " + "pip install ankaios-sdk[command]" + ) from e + return CommandInterfaceConnection( + server_url, + add_response_callback=self._add_response, + add_log_callback=self._add_logs, + add_event_callback=self._add_events, + ca_pem=ca_pem, + crt_pem=crt_pem, + key_pem=key_pem, + ) + return ControlInterfaceConnection( + add_response_callback=self._add_response, + add_log_callback=self._add_logs, + add_event_callback=self._add_events, + ) + def __enter__(self) -> "Ankaios": """ Used for context management. @@ -247,7 +365,7 @@ def __enter__(self) -> "Ankaios": def __exit__(self, exc_type, exc_value, traceback) -> None: """ - Used for context management. Disconnects from the control interface. + Used for context management. Disconnects from Ankaios. :param exc_type: The exception type. :type exc_type: type @@ -263,11 +381,11 @@ def __exit__(self, exc_type, exc_value, traceback) -> None: exc_value, traceback, ) - self._control_interface.disconnect() + self._connection.disconnect() def _add_response(self, response: Response) -> None: """ - Method will be called automatically from the Control Interface + Method will be called automatically from the connection when a response is received. :param response: The received response. @@ -279,7 +397,7 @@ def _add_response(self, response: Response) -> None: def _add_logs(self, request_id: str, logs: list[LogResponse]) -> None: """ - Method will be called automatically from the Control Interface + Method will be called automatically from the connection when a log is received. :param request_id: The request id of the logs campaign. @@ -298,7 +416,7 @@ def _add_logs(self, request_id: str, logs: list[LogResponse]) -> None: def _add_events(self, request_id: str, event: EventEntry) -> None: """ - Method will be called automatically from the Control Interface + Method will be called automatically from the connection when an event is received. :param request_id: The request id of the event campaign. @@ -370,10 +488,10 @@ def _send_request( :rtype: Response :raises TimeoutError: If the request timed out. - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises ConnectionClosedException: If the connection is closed. """ - self._control_interface.write_request(request) + self._connection.write_request(request) response = self._get_response_by_id(request.get_id(), timeout) return response @@ -400,7 +518,7 @@ def apply_manifest( :returns: The update state success object. :rtype: UpdateStateSuccess - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises TimeoutError: If the request timed out. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If the response has unexpected @@ -449,7 +567,7 @@ def delete_manifest( :returns: The update state success object. :rtype: UpdateStateSuccess - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises TimeoutError: If the request timed out. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If the response has unexpected @@ -498,7 +616,7 @@ def apply_workload( :returns: The update state success object. :rtype: UpdateStateSuccess - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises TimeoutError: If the request timed out. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If the response has unexpected @@ -554,7 +672,7 @@ def get_workload( :rtype: list[Workload] :raises TimeoutError: If the request timed out. - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If an error occurred while getting the state. @@ -578,7 +696,7 @@ def delete_workload( :returns: The update state success object. :rtype: UpdateStateSuccess - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises TimeoutError: If the request timed out. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If the response has unexpected @@ -621,7 +739,7 @@ def update_configs(self, configs: dict, timeout: float = DEFAULT_TIMEOUT): :param timeout: The maximum time to wait for the response. :type timeout: float - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises TimeoutError: If the request timed out. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If the response has unexpected @@ -668,7 +786,7 @@ def add_config( :param timeout: The maximum time to wait for the response. :type timeout: float - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises TimeoutError: If the request timed out. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If the response has unexpected @@ -709,7 +827,7 @@ def get_configs(self, timeout: float = DEFAULT_TIMEOUT) -> dict: :rtype: dict :raises TimeoutError: If the request timed out. - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If an error occurred while getting the state. @@ -732,7 +850,7 @@ def get_config(self, name: str, timeout: float = DEFAULT_TIMEOUT) -> dict: :rtype: dict :raises TimeoutError: If the request timed out. - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If an error occurred while getting the state. @@ -749,7 +867,7 @@ def delete_all_configs(self, timeout: float = DEFAULT_TIMEOUT): :param timeout: The maximum time to wait for the response. :type timeout: float - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises TimeoutError: If the request timed out. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If the response has unexpected @@ -786,7 +904,7 @@ def delete_config(self, name: str, timeout: float = DEFAULT_TIMEOUT): :param timeout: The maximum time to wait for the response. :type timeout: float - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises TimeoutError: If the request timed out. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If the response has unexpected @@ -833,7 +951,7 @@ def get_state( :rtype: CompleteState :raises TimeoutError: If the request timed out. - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If an error occurred while getting the state. @@ -873,7 +991,7 @@ def set_agent_tags( :param timeout: The maximum time to wait for the response, in seconds. :type timeout: float - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises TimeoutError: If the request timed out. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If the response has unexpected @@ -916,7 +1034,7 @@ def get_agents(self, timeout: float = DEFAULT_TIMEOUT) -> dict: :rtype: dict :raises TimeoutError: If the request timed out. - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If an error occurred while getting the state. @@ -939,7 +1057,7 @@ def get_agent( :rtype: AgentAttributes :raises TimeoutError: If the request timed out. - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If an error occurred while getting the state or the agent is not found. @@ -967,7 +1085,7 @@ def get_workload_states( :rtype: WorkloadStateCollection :raises TimeoutError: If the request timed out. - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If an error occurred while getting the state. @@ -995,7 +1113,7 @@ def get_execution_state_for_instance_name( :rtype: WorkloadExecutionState :raises TimeoutError: If the request timed out. - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If an error occurred while getting the state. @@ -1032,7 +1150,7 @@ def get_workload_states_on_agent( :rtype: WorkloadStateCollection :raises TimeoutError: If the request timed out. - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If an error occurred while getting the state. @@ -1057,7 +1175,7 @@ def get_workload_states_for_name( :rtype: WorkloadStateCollection :raises TimeoutError: If the request timed out. - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If an error occurred while getting the state. @@ -1092,7 +1210,7 @@ def wait_for_workload_to_reach_state( :raises TimeoutError: If the request timed out or if the workload did not reach the state in time. - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If an error occurred while getting the state. @@ -1143,7 +1261,7 @@ def request_logs( :returns: The log campaign response object. :rtype: LogCampaignResponse - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises ConnectionClosedException: If the connection is closed. """ @@ -1192,7 +1310,7 @@ def stop_receiving_logs( :param timeout: The maximum time to wait for the response, in seconds. :type timeout: float - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises ConnectionClosedException: If the connection is closed. """ request = LogsCancelRequest(request_id=log_campaign.queue._request_id) @@ -1230,7 +1348,7 @@ def register_event( :returns: The event queue. :rtype: EventQueue - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises TimeoutError: If the request timed out. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If the response has unexpected @@ -1274,7 +1392,7 @@ def unregister_event( :param timeout: The maximum time to wait for the response, in seconds. :type timeout: float - :raises ControlInterfaceException: If not connected. + :raises ConnectionException: If not connected. :raises TimeoutError: If the request timed out. :raises AnkaiosResponseError: If the response is an error. :raises AnkaiosProtocolException: If the response has unexpected diff --git a/ankaios_sdk/exceptions.py b/ankaios_sdk/exceptions.py index 87dd6d3..4f304c6 100644 --- a/ankaios_sdk/exceptions.py +++ b/ankaios_sdk/exceptions.py @@ -32,7 +32,7 @@ Raised when the connection is closed. - :class:`ResponseException`: Raised when the response is invalid. -- :class:`ControlInterfaceException`: +- :class:`ConnectionException`: Raised when an operation fails. - :class:`AnkaiosProtocolException`: Raised when something unexpected is received. @@ -49,7 +49,7 @@ "InvalidManifestException", "ConnectionClosedException", "ResponseException", - "ControlInterfaceException", + "ConnectionException", "AnkaiosProtocolException", "AnkaiosResponseError", ] @@ -84,8 +84,8 @@ class ResponseException(AnkaiosException): """Raised when the response is invalid.""" -class ControlInterfaceException(AnkaiosException): - """Raised when an operation on the Control Interface fails""" +class ConnectionException(AnkaiosException): + """Raised when an operation on a connection to Ankaios fails.""" class AnkaiosProtocolException(AnkaiosException): diff --git a/ankaios_sdk/utils.py b/ankaios_sdk/utils.py index 5632dc4..6f89ef5 100644 --- a/ankaios_sdk/utils.py +++ b/ankaios_sdk/utils.py @@ -40,7 +40,7 @@ SUPPORTED_API_VERSION = "v1" "(str): The supported API version of the Ankaios SDK." -ANKAIOS_VERSION = "1.0.0" +ANKAIOS_VERSION = "1.1.0-pre" "(str): The version of the compatible Ankaios." WORKLOADS_PREFIX = "desiredState.workloads" diff --git a/command_example/README.md b/command_example/README.md new file mode 100644 index 0000000..7604797 --- /dev/null +++ b/command_example/README.md @@ -0,0 +1,45 @@ +# Command Interface Examples + +Standalone scripts that connect to an Ankaios server directly over +the Command Interface (`ConnectionType.COMMAND_INTERFACE`) which uses +the gRPC protocol, , from *outside* a workload — unlike the +apps in [`../examples`](../examples), which run *inside* a workload +via the control interface. + +## Prerequisites + +An Ankaios server (and at least one agent) must already be running +and reachable, e.g. started locally with: + +```shell +ank-server --insecure --address 0.0.0.0:25551 & +ank-agent --insecure --name agent_A --server-url http://127.0.0.1:25551 & +``` + +Install the SDK with the `command` extra from the repository root: + +```shell +pip install -e ".[command]" +``` + +## Running + +Each script takes the server URL and agent name as optional +positional arguments (defaulting to `http://127.0.0.1:25551` and +`agent_A`): + +```shell +cd command_example +python3 basic_test.py [server_url] [agent_name] +python3 logs_events_test.py [server_url] [agent_name] +``` + +## Scripts + +- **`basic_test.py`** — applies a workload, waits for it to reach the + `RUNNING` state, updates it, then deletes it, printing the + workload states after every change. +- **`logs_events_test.py`** — registers for events on a workload, + applies it, then streams its logs (printing them) until they stop, + printing any events received along the way. +- **`common.py`** — shared helpers used by the scripts above. diff --git a/command_example/basic_test.py b/command_example/basic_test.py new file mode 100644 index 0000000..458e885 --- /dev/null +++ b/command_example/basic_test.py @@ -0,0 +1,107 @@ +# Copyright (c) 2026 Elektrobit Automotive GmbH +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://www.apache.org/licenses/LICENSE-2.0. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# SPDX-License-Identifier: Apache-2.0 + +""" +Connects to an Ankaios server directly over the Command Interface, +applies a workload, updates it, then deletes it, printing +the workload states after every change. + +Usage: + python3 basic_test.py [server_url] [agent_name] +""" + +import sys +import time + +from common import ( + DEFAULT_AGENT_NAME, + DEFAULT_SERVER_URL, + print_workload_states, +) +from ankaios_sdk import ( + Ankaios, + AnkaiosException, + ConnectionType, + Workload, + WorkloadStateEnum, +) + + +def main() -> None: + """Applies, updates and deletes a workload over the Command Interface.""" + server_url = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_SERVER_URL + agent_name = sys.argv[2] if len(sys.argv) > 2 else DEFAULT_AGENT_NAME + + with Ankaios( + connection_type=ConnectionType.COMMAND_INTERFACE, server_url=server_url + ) as ankaios: + workload = ( + Workload.builder() + .workload_name("dynamic_nginx") + .agent_name(agent_name) + .runtime("podman") + .restart_policy("NEVER") + .runtime_config( + "image: docker.io/library/nginx\n" + 'commandOptions: ["-p", "8080:80"]' + ) + .build() + ) + + try: + # Run the workload + update_response = ankaios.apply_workload(workload) + workload_instance_name = update_response.added_workloads[0] + + try: + ankaios.wait_for_workload_to_reach_state( + workload_instance_name, WorkloadStateEnum.RUNNING + ) + print("Workload reached the RUNNING state.") + except TimeoutError: + print("Workload didn't reach the required state in time.") + print_workload_states(ankaios) + + time.sleep(2) + + # Update the workload + workloads = ankaios.get_workload( + workload_instance_name.workload_name + ) + workload = workloads[0] + workload.update_restart_policy("ALWAYS") + ankaios.apply_workload(workload) + + try: + ankaios.wait_for_workload_to_reach_state( + workload_instance_name, WorkloadStateEnum.RUNNING + ) + print("Workload reached the RUNNING state after update.") + except TimeoutError: + print("Workload didn't reach the required state in time.") + print_workload_states(ankaios) + + time.sleep(2) + + # Delete the workload + ankaios.delete_workload(workload_instance_name.workload_name) + time.sleep(5) + print_workload_states(ankaios) + + except AnkaiosException as e: + print("Ankaios Exception occurred: ", e) + + +if __name__ == "__main__": + main() diff --git a/command_example/common.py b/command_example/common.py new file mode 100644 index 0000000..748b9f3 --- /dev/null +++ b/command_example/common.py @@ -0,0 +1,39 @@ +# Copyright (c) 2026 Elektrobit Automotive GmbH +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://www.apache.org/licenses/LICENSE-2.0. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# SPDX-License-Identifier: Apache-2.0 + +""" +Shared helpers for the Command Interface example scripts. +""" + +from ankaios_sdk import Ankaios + +DEFAULT_SERVER_URL = "http://127.0.0.1:25551" +DEFAULT_AGENT_NAME = "agent_A" + + +def print_workload_states(ankaios: Ankaios) -> None: + """ + Prints the state of every workload in the cluster. + + :param ankaios: The Ankaios object to query. + :type ankaios: Ankaios + """ + complete_state = ankaios.get_state(field_masks=["workloadStates"]) + for workload_state in complete_state.get_workload_states().get_as_list(): + instance_name = workload_state.workload_instance_name + print( + f"Workload {instance_name.workload_name} on agent " + f"{instance_name.agent_name} has the state " + f"{workload_state.execution_state.state}" + ) diff --git a/command_example/logs_events_test.py b/command_example/logs_events_test.py new file mode 100644 index 0000000..2667d32 --- /dev/null +++ b/command_example/logs_events_test.py @@ -0,0 +1,135 @@ +# Copyright (c) 2026 Elektrobit Automotive GmbH +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://www.apache.org/licenses/LICENSE-2.0. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# SPDX-License-Identifier: Apache-2.0 + +""" +Connects to an Ankaios server directly over the Command Interface, +registers for events on a workload, applies it, streams its +logs until they stop while printing any events received along the +way, then unregisters/deletes everything. + +Usage: + python3 logs_events_test.py [server_url] [agent_name] +""" + +import sys +import threading +from queue import Empty, Queue + +from common import DEFAULT_AGENT_NAME, DEFAULT_SERVER_URL +from ankaios_sdk import ( + Ankaios, + AnkaiosException, + ConnectionType, + EventEntry, + LogEntry, + LogsStopResponse, + Workload, +) + + +def _print_events(event_queue: Queue, stop_event: threading.Event) -> None: + """Prints events from the queue until stop_event is set.""" + while not stop_event.is_set(): + try: + event: EventEntry = event_queue.get(timeout=0.2) + except Empty: + continue + print( + f"[event] added={event.added_fields} " + f"updated={event.updated_fields} " + f"removed={event.removed_fields}" + ) + + +def main() -> None: + """Streams logs and events for a short-lived workload over the Command Interface.""" + server_url = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_SERVER_URL + agent_name = sys.argv[2] if len(sys.argv) > 2 else DEFAULT_AGENT_NAME + + with Ankaios( + connection_type=ConnectionType.COMMAND_INTERFACE, server_url=server_url + ) as ankaios: + # Workload that prints a handful of lines then exits. + workload = ( + Workload.builder() + .workload_name("count_to_five") + .agent_name(agent_name) + .runtime("podman") + .restart_policy("NEVER") + .runtime_config( + "image: ghcr.io/eclipse-ankaios/tests/alpine:latest\n" + 'commandOptions: [ "--entrypoint", "/bin/sh" ]\n' + "commandArgs: [ \"-c\", \"echo -e '1\\n2\\n3\\n4\\n5';\" ]" + ) + .build() + ) + + try: + # Subscribe to changes on this workload's desired state + # before applying it. + event_queue = ankaios.register_event( + field_masks=["desiredState.workloads.count_to_five"], + ) + + update_response = ankaios.apply_workload(workload) + workload_instance_name = update_response.added_workloads[0] + + log_campaign = ankaios.request_logs( + workload_names=[workload_instance_name], + ) + if ( + workload_instance_name + not in log_campaign.accepted_workload_names + ): + print( + f"Workload '{workload_instance_name}' not accepted " + "for log retrieval" + ) + + # Print events in the background while logs are streamed + # on the main thread below. + stop_event = threading.Event() + events_thread = threading.Thread( + target=_print_events, + args=(event_queue, stop_event), + daemon=True, + ) + events_thread.start() + + # Stream logs until they stop. + while True: + log = log_campaign.queue.get() + match log: + case LogEntry(): + print(f"[log] {log.message}") + case LogsStopResponse(): + print( + "No more logs available for " + f"'{workload_instance_name}'." + ) + break + + stop_event.set() + events_thread.join() + + ankaios.stop_receiving_logs(log_campaign) + ankaios.unregister_event(event_queue) + ankaios.delete_workload(workload_instance_name.workload_name) + + except AnkaiosException as e: + print("Ankaios Exception occurred: ", e) + + +if __name__ == "__main__": + main() diff --git a/docs/source/command_interface.rst b/docs/source/command_interface.rst new file mode 100644 index 0000000..de7bbc4 --- /dev/null +++ b/docs/source/command_interface.rst @@ -0,0 +1,22 @@ +CommandInterface +================ + +.. automodule:: ankaios_sdk._components.connection.command_interface + +CommandInterfaceConnection Class +-------------------------------- + +.. autoclass:: ankaios_sdk._components.connection.command_interface.CommandInterfaceConnection + :special-members: __init__ + :members: + :undoc-members: + :show-inheritance: + +CommandInterfaceState Enum +-------------------------- + +.. autoclass:: ankaios_sdk._components.connection.command_interface.CommandInterfaceState + :special-members: __str__ + :members: + :undoc-members: + :show-inheritance: \ No newline at end of file diff --git a/docs/source/conf.py b/docs/source/conf.py index 41ad8d9..f1b6178 100644 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -44,6 +44,11 @@ exclude_patterns = [] autodoc_member_order = "bysource" +# CommandInterfaceConnection depends on the optional 'grpc' extra, which +# the docs build environment doesn't install; mock it so autodoc can +# still document the class without needing it importable for real. +autodoc_mock_imports = ["grpc"] + # -- Options for HTML output ------------------------------------------------- # https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output diff --git a/docs/source/connection.rst b/docs/source/connection.rst new file mode 100644 index 0000000..dc79eeb --- /dev/null +++ b/docs/source/connection.rst @@ -0,0 +1,22 @@ +Connection +========== + +.. automodule:: ankaios_sdk._components.connection.connection + +Connection Class +---------------- + +.. autoclass:: ankaios_sdk._components.connection.connection.Connection + :special-members: __init__ + :members: + :undoc-members: + :show-inheritance: + +ConnectionType Enum +------------------- + +.. autoclass:: ankaios_sdk._components.connection.connection.ConnectionType + :special-members: __str__ + :members: + :undoc-members: + :show-inheritance: diff --git a/docs/source/control_interface.rst b/docs/source/control_interface.rst index 21ef117..d19dbaf 100644 --- a/docs/source/control_interface.rst +++ b/docs/source/control_interface.rst @@ -1,12 +1,12 @@ ControlInterface ================ -.. automodule:: ankaios_sdk._components.control_interface +.. automodule:: ankaios_sdk._components.connection.control_interface -ControlInterface Class ----------------------- +ControlInterfaceConnection Class +-------------------------------- -.. autoclass:: ankaios_sdk._components.control_interface.ControlInterface +.. autoclass:: ankaios_sdk._components.connection.control_interface.ControlInterfaceConnection :special-members: __init__ :members: :undoc-members: @@ -15,7 +15,7 @@ ControlInterface Class ControlInterfaceState Enum -------------------------- -.. autoclass:: ankaios_sdk._components.control_interface.ControlInterfaceState +.. autoclass:: ankaios_sdk._components.connection.control_interface.ControlInterfaceState :special-members: __str__ :members: :undoc-members: diff --git a/docs/source/index.rst b/docs/source/index.rst index 12f5177..f3c3282 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -23,7 +23,9 @@ file log_campaign event_campaign + connection control_interface + command_interface utils exceptions diff --git a/examples/apps/follow_state.py b/examples/apps/follow_state.py index 3f771ff..32ec191 100644 --- a/examples/apps/follow_state.py +++ b/examples/apps/follow_state.py @@ -12,7 +12,7 @@ # # SPDX-License-Identifier: Apache-2.0 -from ankaios_sdk import Ankaios, ControlInterfaceException, Workload +from ankaios_sdk import Ankaios, ConnectionException, Workload from time import sleep import sys, signal @@ -36,7 +36,7 @@ def signal_handler(sig, frame): complete_state = ankaios.get_state( timeout=5, field_masks=["workloadStates"] ) - except ControlInterfaceException as e: + except ConnectionException as e: print(f"Error while getting the state: {e}") else: # Get the workload states present in the complete_state diff --git a/setup.cfg b/setup.cfg index 20753b4..f3dbe13 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,7 +1,7 @@ [metadata] name = ankaios-sdk -version = 1.0.1 -ankaios_version = 1.0.0 +version = 1.1.0-pre +ankaios_version = 1.1.0-pre author = Elektrobit Automotive GmbH and Ankaios contributors license = Apache-2.0 diff --git a/setup.py b/setup.py index 48e97a6..2120bfb 100644 --- a/setup.py +++ b/setup.py @@ -23,6 +23,13 @@ PROTO_FILES = [ "ankaios_api/proto/ank_base.proto", "ankaios_api/proto/control_api.proto", + "grpc/proto/grpc_api.proto", +] + +# gRPC client library and codegen tooling, needed both to actually run +# the gRPC connection and to develop/test it (hence reused in "dev"). +COMMAND_REQUIRES = [ + "grpcio-tools==1.76.0", ] config = configparser.ConfigParser() @@ -105,9 +112,9 @@ def generate_protos(): if protoc.main(command) != 0: raise RuntimeError(f"Error: {proto_file} compilation failed") - # Fix the import path in the generated control_api_pb2 + # Fix the import path in the generated protobuf files to relative imports # https://github.com/protocolbuffers/protobuf/issues/1491#issuecomment-261914766 - if "control_api" in proto_file: + if "control_api" in proto_file or "grpc_api" in proto_file: with open(output_file, "r") as file: filedata = file.read() newdata = filedata.replace( @@ -116,6 +123,16 @@ def generate_protos(): ) with open(output_file, "w") as file: file.write(newdata) + if "grpc_api" in proto_file: + grpc_output_file = proto_path.replace(".proto", "_pb2_grpc.py") + with open(grpc_output_file, "r") as file: + filedata = file.read() + newdata = filedata.replace( + "import grpc_api_pb2 as grpc__api__pb2", + "from . import grpc_api_pb2 as grpc__api__pb2", + ) + with open(grpc_output_file, "w") as file: + file.write(newdata) # Copy the generated files to the proto directory for file in os.listdir(protos_dir): @@ -163,7 +180,8 @@ def generate_protos(): "pytest-cov", # Coverage plugin "pylint", # Linter "pycodestyle", # Style guide checker - ], + ] + + COMMAND_REQUIRES, # Documentation dependencies "docs": [ "sphinx", # Documentation generator @@ -173,6 +191,8 @@ def generate_protos(): "sphinx-versioned-docs", # Versioned docs support "google-api-python-client", # Required for the Google API docstring extension ], + # Command Interface dependencies + "command": COMMAND_REQUIRES, }, ) diff --git a/tests/connection/__init__.py b/tests/connection/__init__.py new file mode 100644 index 0000000..a629f7c --- /dev/null +++ b/tests/connection/__init__.py @@ -0,0 +1,13 @@ +# Copyright (c) 2026 Elektrobit Automotive GmbH +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://www.apache.org/licenses/LICENSE-2.0. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# SPDX-License-Identifier: Apache-2.0 diff --git a/tests/connection/test_command_interface.py b/tests/connection/test_command_interface.py new file mode 100644 index 0000000..5cc73af --- /dev/null +++ b/tests/connection/test_command_interface.py @@ -0,0 +1,624 @@ +# Copyright (c) 2026 Elektrobit Automotive GmbH +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://www.apache.org/licenses/LICENSE-2.0. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# SPDX-License-Identifier: Apache-2.0 + +""" +This module contains unit tests for the CommandInterfaceConnection class in the +ankaios_sdk. +""" + +import threading +import time +from unittest.mock import MagicMock, patch + +import grpc +import pytest + +from ankaios_sdk import ConnectionException +from ankaios_sdk._components.connection.command_interface import ( + CommandInterfaceConnection, + CommandInterfaceState, +) +from ankaios_sdk._protos import grpc_api_pb2 as _grpc_api +from ankaios_sdk._protos import _ank_base +from ankaios_sdk.utils import ANKAIOS_VERSION +from tests.request.test_request import generate_test_request +from tests.response.test_log_response import ( + generate_test_log_entry, + generate_test_logs_stop_reponse, +) + + +SERVER_URL = "http://127.0.0.1:25551" +SERVER_TARGET = "127.0.0.1:25551" + + +class _FakeCall: + """ + A fake bidi call standing in for the object returned by + stub.ConnectCommand(...): an iterable of FromServer messages, + optionally followed by an error, with a working cancel(). + """ + + def __init__(self, messages=None, error=None): + self._messages = list(messages) if messages else [] + self._error = error + self.cancelled = threading.Event() + + def __iter__(self): + for message in self._messages: + if self.cancelled.is_set(): + return + yield message + if self._error is not None: + raise self._error + while not self.cancelled.is_set(): + time.sleep(0.01) + + def cancel(self): + """Cancels the fake call, unblocking __iter__.""" + self.cancelled.set() + + +def _mock_stub(call_factory): + """ + Builds a MagicMock standing in for CommandConnectionStub, whose + ConnectCommand(request_iterator) drains the request_iterator in a + background thread (collecting the sent ToServer messages) and + returns a call built via call_factory(). + + :returns: A tuple of (stub_instance, sent_messages list). + """ + sent_messages = [] + + def _connect_command(request_iterator): + def _drain(): + for message in request_iterator: + sent_messages.append(message) + + threading.Thread(target=_drain, daemon=True).start() + return call_factory() + + stub_instance = MagicMock() + stub_instance.ConnectCommand.side_effect = _connect_command + return stub_instance, sent_messages + + +def _generate_test_connection(**kwargs) -> CommandInterfaceConnection: + return CommandInterfaceConnection( + SERVER_URL, + add_response_callback=MagicMock(), + add_log_callback=MagicMock(), + add_event_callback=MagicMock(), + **kwargs, + ) + + +def test_grpc_connection_state_str(): + """ + Test the string representation of the CommandInterfaceState enum. + """ + assert str(CommandInterfaceState.CONNECTED) == "CONNECTED" + assert str(CommandInterfaceState.TERMINATED) == "TERMINATED" + + +def test_init_validation(): + """ + Test the __init__ validation of the TLS certificate arguments. + """ + # No certs: insecure connection, no error + conn = _generate_test_connection() + assert conn._ca_pem is None + assert conn._crt_pem is None + assert conn._key_pem is None + + # All three certs: secured connection, no error + conn = _generate_test_connection( + ca_pem="ca-secret", crt_pem="crt-secret", key_pem="key-secret" + ) + assert conn._ca_pem == "ca-secret" + assert conn._crt_pem == "crt-secret" + assert conn._key_pem == "key-secret" + + # Partial certs: error + with pytest.raises(ValueError, match="must all be provided together"): + _generate_test_connection(crt_pem="crt-secret") + + +def test_connected_property_default(): + """ + Test that a freshly created connection is not connected. + """ + conn = _generate_test_connection() + assert conn.connected is False + + +def test_change_state_no_op_when_already_in_state(): + """ + Test that change_state() is a no-op (but still logs) when asked + to change to the state the connection is already in. + """ + conn = _generate_test_connection() + conn._logger = MagicMock() + + conn.change_state(CommandInterfaceState.TERMINATED) + conn._logger.debug.assert_called_with( + "State is already %s.", CommandInterfaceState.TERMINATED + ) + + conn.change_state(CommandInterfaceState.CONNECTED) + conn._logger.debug.assert_called_with( + "State changed to %s.", CommandInterfaceState.CONNECTED + ) + assert conn._state == CommandInterfaceState.CONNECTED + + +def test_connect_already_connected_raises(): + """ + Test that connect() raises if already connected. + """ + conn = _generate_test_connection() + conn._state = CommandInterfaceState.CONNECTED + with pytest.raises(ConnectionException, match="Already connected."): + conn.connect() + + +def test_connect_fails_on_unreachable_server(): + """ + Test that connect() fails (and leaves the state untouched) when + the server is unreachable. Port 0 is never a valid connection + target, so this fails fast without needing a real server. + """ + conn = _generate_test_connection() + conn._server_url = "127.0.0.1:0" + + with patch.object( + CommandInterfaceConnection, "CHANNEL_READY_TIMEOUT", 1.0 + ), pytest.raises( + ConnectionException, match="Could not connect to the Ankaios server" + ): + conn.connect() + assert conn._state == CommandInterfaceState.TERMINATED + assert conn.connected is False + + +def test_disconnect_without_connect(): + """ + Test that disconnect() without a prior connect() is a no-op. + """ + conn = _generate_test_connection() + conn._logger = MagicMock() + conn.disconnect() + conn._logger.debug.assert_called_with("Already disconnected.") + + +def test_write_request_not_connected_raises(): + """ + Test that write_request() raises if not connected. + """ + conn = _generate_test_connection() + request = generate_test_request() + with pytest.raises( + ConnectionException, match="Could not write to the gRPC connection" + ): + conn.write_request(request) + + +def test_connect_write_and_disconnect_success(): + """ + Test the full connect -> write_request -> disconnect flow using a + mocked gRPC channel/stub boundary. + """ + fake_call = _FakeCall() + stub_instance, sent_messages = _mock_stub(lambda: fake_call) + + with patch( + "ankaios_sdk._components.connection.command_interface." + "_grpc_api_grpc.CommandConnectionStub" + ) as mock_stub_cls, patch("grpc.insecure_channel") as mock_channel, patch( + "grpc.channel_ready_future" + ) as mock_ready: + mock_stub_cls.return_value = stub_instance + mock_channel.return_value = MagicMock() + mock_ready.return_value = MagicMock( + result=MagicMock(return_value=None) + ) + + conn = _generate_test_connection() + conn.connect() + assert conn.connected is True + assert conn._reader_thread.is_alive() + + time.sleep(0.05) + assert len(sent_messages) == 1 + assert sent_messages[0].WhichOneof("ToServerEnum") == "commanderHello" + assert ( + sent_messages[0].commanderHello.protocolVersion + == ANKAIOS_VERSION + ) + + request = generate_test_request() + conn.write_request(request) + time.sleep(0.05) + assert len(sent_messages) == 2 + assert sent_messages[1].WhichOneof("ToServerEnum") == "request" + assert sent_messages[1].request == request._to_proto() + + conn.disconnect() + assert conn.connected is False + assert fake_call.cancelled.is_set() + assert conn._reader_thread is None + + # Disconnecting again is a no-op. + conn.disconnect() + + +def test_open_stream_channel_not_ready_raises(): + """ + Test that _open_stream() raises ConnectionException if the + channel does not become ready in time. + """ + with patch("grpc.insecure_channel") as mock_channel, patch( + "grpc.channel_ready_future" + ) as mock_ready: + mock_channel.return_value = MagicMock() + mock_ready.return_value = MagicMock( + result=MagicMock(side_effect=grpc.FutureTimeoutError()) + ) + conn = _generate_test_connection() + with pytest.raises( + ConnectionException, match="Could not connect" + ): + conn._open_stream() + + +def test_build_channel_insecure(): + """ + Test that _build_channel() builds a plaintext channel when no + certificates are provided, using the bare host:port target (the + http:// scheme is stripped, since grpc-python's channel target + doesn't understand URL schemes). + """ + with patch("grpc.insecure_channel") as mock_channel: + conn = _generate_test_connection() + conn._build_channel() + mock_channel.assert_called_once_with(SERVER_TARGET) + + +def test_build_channel_secure(): + """ + Test that _build_channel() builds a mTLS-secured channel when + certificates are provided, using the bare host:port target. + """ + with patch("grpc.secure_channel") as mock_secure_channel, patch( + "grpc.ssl_channel_credentials" + ) as mock_credentials: + mock_credentials.return_value = "credentials" + conn = _generate_test_connection( + ca_pem="ca-secret", crt_pem="crt-secret", key_pem="key-secret" + ) + conn._build_channel() + + mock_credentials.assert_called_once_with( + root_certificates=b"ca-secret", + private_key=b"key-secret", + certificate_chain=b"crt-secret", + ) + mock_secure_channel.assert_called_once_with( + SERVER_TARGET, + "credentials", + options=(("grpc.ssl_target_name_override", "ank-server"),), + ) + + +def test_grpc_target_strips_url_scheme(): + """ + Test that _grpc_target() strips a http(s):// scheme prefix from + server_url, since grpc's channel target must be a bare host:port. + A server_url given without a scheme is passed through unchanged. + """ + conn = _generate_test_connection() + + conn._server_url = "http://127.0.0.1:25551" + assert conn._grpc_target() == "127.0.0.1:25551" + + conn._server_url = "https://127.0.0.1:25551" + assert conn._grpc_target() == "127.0.0.1:25551" + + conn._server_url = "127.0.0.1:25551" + assert conn._grpc_target() == "127.0.0.1:25551" + + +def test_request_iterator(): + """ + Test that _request_iterator() yields exactly the queued messages, + in order. + """ + conn = _generate_test_connection() + conn._write_queue.put("message_1") + conn._write_queue.put("message_2") + + iterator = conn._request_iterator() + assert next(iterator) == "message_1" + assert next(iterator) == "message_2" + + +def test_handle_from_server_generic_response(): + """ + Test that a generic response is forwarded via add_response_callback. + """ + conn = _generate_test_connection() + ank_base_response = _ank_base.Response( + requestId="1122", + error=_ank_base.Error(message="Test error message"), + ) + conn._handle_from_server( + _grpc_api.FromServer(response=ank_base_response) + ) + conn._add_response_callback.assert_called_once() + response = conn._add_response_callback.call_args[0][0] + assert response.get_request_id() == "1122" + conn._add_log_callback.assert_not_called() + conn._add_event_callback.assert_not_called() + + +def test_handle_from_server_log_entries(): + """ + Test that a log entries response is forwarded via + add_log_callback. + """ + conn = _generate_test_connection() + ank_base_response = _ank_base.Response( + requestId="1122", + logEntriesResponse=_ank_base.LogEntriesResponse( + logEntries=[generate_test_log_entry()] + ), + ) + conn._handle_from_server( + _grpc_api.FromServer(response=ank_base_response) + ) + conn._add_log_callback.assert_called_once() + request_id, content = conn._add_log_callback.call_args[0] + assert request_id == "1122" + assert len(content) == 1 + conn._add_response_callback.assert_not_called() + + +def test_handle_from_server_logs_stop_response(): + """ + Test that a logs stop response is forwarded via add_log_callback. + """ + conn = _generate_test_connection() + ank_base_response = _ank_base.Response( + requestId="1122", + logsStopResponse=generate_test_logs_stop_reponse(), + ) + conn._handle_from_server( + _grpc_api.FromServer(response=ank_base_response) + ) + conn._add_log_callback.assert_called_once() + conn._add_response_callback.assert_not_called() + + +def test_handle_from_server_event_response(): + """ + Test that an event response is forwarded via add_event_callback. + """ + conn = _generate_test_connection() + ank_base_response = _ank_base.Response( + requestId="1122", + completeStateResponse=_ank_base.CompleteStateResponse( + completeState=_ank_base.CompleteState( + desiredState=_ank_base.State(apiVersion="v1"), + ), + alteredFields=_ank_base.AlteredFields( + addedFields=["desiredState.workloads.test"], + ), + ), + ) + conn._handle_from_server( + _grpc_api.FromServer(response=ank_base_response) + ) + conn._add_event_callback.assert_called_once() + request_id, content = conn._add_event_callback.call_args[0] + assert request_id == "1122" + assert content.added_fields == ["desiredState.workloads.test"] + conn._add_response_callback.assert_not_called() + + +def test_handle_from_server_ignores_non_response_messages(): + """ + Test that non-response messages (serverHello, or anything else + targeted at agents rather than commanders) are ignored, without + raising or forwarding anything. + """ + conn = _generate_test_connection() + + conn._handle_from_server( + _grpc_api.FromServer(serverHello=_grpc_api.ServerHello()) + ) + conn._handle_from_server( + _grpc_api.FromServer(updateWorkload=_grpc_api.UpdateWorkload()) + ) + conn._handle_from_server(_grpc_api.FromServer()) + + conn._add_response_callback.assert_not_called() + conn._add_log_callback.assert_not_called() + conn._add_event_callback.assert_not_called() + + +def test_read_from_grpc_stops_on_terminated_state(): + """ + Test that the reader loop returns without attempting a reconnect + if the connection was terminated (disconnect() was called), and + logs the resulting RpcError at debug level, since it's the + expected result of disconnect()'s own call.cancel(). + """ + conn = _generate_test_connection() + conn._state = CommandInterfaceState.TERMINATED + conn._logger = MagicMock() + fake_call = _FakeCall(error=grpc.RpcError("boom")) + + with patch.object(conn, "_reconnect") as mock_reconnect: + conn._read_from_grpc(fake_call) + mock_reconnect.assert_not_called() + + conn._logger.debug.assert_called_once() + conn._logger.warning.assert_not_called() + + +def test_read_from_grpc_reconnects_on_lost_connection(): + """ + Test that the reader loop attempts a reconnect (via the real + _reconnect logic) when the stream errors while the connection is + still expected to be up, logging the RpcError as a warning (since + this is an unexpected drop, not one caused by disconnect()), and + resumes reading from the newly reconnected call. + """ + conn = _generate_test_connection() + conn._state = CommandInterfaceState.CONNECTED + conn._logger = MagicMock() + first_call = _FakeCall(error=grpc.RpcError("boom")) + second_call = _FakeCall() + + with patch.object( + CommandInterfaceConnection, "RECONNECT_INTERVAL", 0.01 + ), patch.object(conn, "_open_stream", return_value=second_call): + reader_thread = threading.Thread( + target=conn._read_from_grpc, args=(first_call,), daemon=True + ) + reader_thread.start() + reader_thread.join(timeout=1) + + assert conn._state == CommandInterfaceState.CONNECTED + conn._logger.warning.assert_any_call( + "Error while reading from the gRPC connection: '%s'", + first_call._error, + ) + conn._logger.debug.assert_any_call( + "State changed to %s.", CommandInterfaceState.CONNECTED + ) + + second_call.cancel() + reader_thread.join(timeout=1) + + +def test_reconnect_gives_up_when_terminated(): + """ + Test that _reconnect() gives up (returns None) as soon as it + notices the connection was terminated, without attempting to + reopen the stream. + """ + conn = _generate_test_connection() + conn._state = CommandInterfaceState.TERMINATED + + with patch.object( + CommandInterfaceConnection, "RECONNECT_INTERVAL", 0.01 + ), patch.object(conn, "_open_stream") as mock_open_stream: + result = conn._reconnect() + assert result is None + mock_open_stream.assert_not_called() + + +def test_reconnect_retries_until_success(): + """ + Test that _reconnect() retries opening the stream until it + succeeds, and updates the state back to CONNECTED. + """ + conn = _generate_test_connection() + conn._state = CommandInterfaceState.RECONNECTING + fake_call = _FakeCall() + + attempts = [ConnectionException("still down"), fake_call] + + def _fake_open_stream(): + attempt = attempts.pop(0) + if isinstance(attempt, Exception): + raise attempt + return attempt + + with patch.object( + CommandInterfaceConnection, "RECONNECT_INTERVAL", 0.01 + ), patch.object(conn, "_open_stream", side_effect=_fake_open_stream): + result = conn._reconnect() + + assert result is fake_call + assert conn._state == CommandInterfaceState.CONNECTED + + +def test_read_from_grpc_dispatches_messages(): + """ + Test that messages read from the bidi call are dispatched via + _handle_from_server while the reader loop is running. + """ + conn = _generate_test_connection() + conn._state = CommandInterfaceState.CONNECTED + fake_call = _FakeCall( + messages=[_grpc_api.FromServer(serverHello=_grpc_api.ServerHello())] + ) + reader_thread = threading.Thread( + target=conn._read_from_grpc, args=(fake_call,), daemon=True + ) + reader_thread.start() + time.sleep(0.05) + + conn._state = CommandInterfaceState.TERMINATED + fake_call.cancel() + reader_thread.join(timeout=1) + assert not reader_thread.is_alive() + + +def test_read_from_grpc_stops_when_reconnect_gives_up(): + """ + Test that the reader loop returns once _reconnect() gives up + (returns None), without looping forever. + """ + conn = _generate_test_connection() + conn._state = CommandInterfaceState.CONNECTED + fake_call = _FakeCall(error=grpc.RpcError("boom")) + + with patch.object(conn, "_reconnect", return_value=None): + conn._read_from_grpc(fake_call) + + +def test_disconnect_logs_error_when_reader_thread_does_not_stop(): + """ + Test that disconnect() logs an error if the reader thread does + not stop within the join timeout. + """ + fake_call = _FakeCall() + stub_instance, _ = _mock_stub(lambda: fake_call) + + with patch( + "ankaios_sdk._components.connection.command_interface." + "_grpc_api_grpc.CommandConnectionStub" + ) as mock_stub_cls, patch("grpc.insecure_channel") as mock_channel, patch( + "grpc.channel_ready_future" + ) as mock_ready, patch("threading.Thread") as mock_thread: + mock_stub_cls.return_value = stub_instance + mock_channel.return_value = MagicMock() + mock_ready.return_value = MagicMock( + result=MagicMock(return_value=None) + ) + mock_thread_instance = MagicMock() + mock_thread.return_value = mock_thread_instance + + conn = _generate_test_connection() + conn._logger = MagicMock() + conn.connect() + conn.disconnect() + + mock_thread_instance.join.assert_called_once_with(timeout=2) + conn._logger.error.assert_called_once_with( + "Reader thread did not stop." + ) diff --git a/tests/connection/test_connection.py b/tests/connection/test_connection.py new file mode 100644 index 0000000..f3c2e11 --- /dev/null +++ b/tests/connection/test_connection.py @@ -0,0 +1,86 @@ +# Copyright (c) 2026 Elektrobit Automotive GmbH +# +# This program and the accompanying materials are made available under the +# terms of the Apache License, Version 2.0 which is available at +# https://www.apache.org/licenses/LICENSE-2.0. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# SPDX-License-Identifier: Apache-2.0 + +""" +This module contains unit tests for the Connection abstract base +class in the ankaios_sdk. +""" + +import pytest +from ankaios_sdk import Connection, ConnectionType +from tests.request.test_request import generate_test_request + + +class _DummyConnection(Connection): + """Minimal concrete Connection used to verify the abstract + contract.""" + + def __init__(self) -> None: + super().__init__( + lambda response: None, + lambda request_id, log: None, + lambda request_id, event: None, + ) + self.written_requests = [] + self._connected = False + + @property + def connected(self) -> bool: + return self._connected + + def connect(self) -> None: + self._connected = True + + def disconnect(self) -> None: + self._connected = False + + def write_request(self, request) -> None: + self.written_requests.append(request) + + +def test_connection_type_str(): + """ + Test the string representation of the ConnectionType enum. + """ + assert str(ConnectionType.CONTROL_INTERFACE) == "CONTROL_INTERFACE" + assert str(ConnectionType.COMMAND_INTERFACE) == "COMMAND_INTERFACE" + + +def test_connection_cannot_be_instantiated_directly(): + """ + Test that Connection cannot be instantiated directly, since it + is an abstract base class. + """ + with pytest.raises(TypeError): + # pylint: disable=abstract-class-instantiated,no-value-for-parameter + Connection() + + +def test_connection_concrete_subclass_implements_contract(): + """ + Test that a concrete Connection subclass can be instantiated and + fulfills the connect/disconnect/write_request/connected contract. + """ + conn = _DummyConnection() + assert conn.connected is False + + conn.connect() + assert conn.connected is True + + request = generate_test_request() + conn.write_request(request) + assert conn.written_requests == [request] + + conn.disconnect() + assert conn.connected is False diff --git a/tests/test_control_interface.py b/tests/connection/test_control_interface.py similarity index 83% rename from tests/test_control_interface.py rename to tests/connection/test_control_interface.py index 27623e6..9f094f3 100644 --- a/tests/test_control_interface.py +++ b/tests/connection/test_control_interface.py @@ -13,7 +13,7 @@ # SPDX-License-Identifier: Apache-2.0 """ -This module contains unit tests for the ControlInterface class +This module contains unit tests for the ControlInterfaceConnection class in the ankaios_sdk. """ @@ -22,10 +22,12 @@ from unittest.mock import patch, mock_open, MagicMock import pytest from ankaios_sdk import ( - ControlInterface, + ControlInterfaceConnection, Response, + ResponseException, + ResponseType, ControlInterfaceState, - ControlInterfaceException, + ConnectionException, ConnectionClosedException, ) from ankaios_sdk.utils import ANKAIOS_VERSION @@ -46,7 +48,7 @@ def test_state(): """ Test the state enum and the changing of the state. """ - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=lambda _: None, add_log_callback=lambda _: None, add_event_callback=lambda _: None, @@ -71,7 +73,7 @@ def test_connection(): """ Test the connect / disconnect functionality. """ - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=lambda _: None, add_log_callback=lambda _: None, add_event_callback=lambda _: None, @@ -80,14 +82,14 @@ def test_connection(): assert ci.connected # Already connected - with pytest.raises(ControlInterfaceException, match="Already connected."): + with pytest.raises(ConnectionException, match="Already connected."): ci.connect() # Test input pipe does not exist ci._state = ControlInterfaceState.TERMINATED assert not ci.connected with patch("os.path.exists") as mock_exists, pytest.raises( - ControlInterfaceException, match="Control interface input fifo" + ConnectionException, match="Control interface input fifo" ): mock_exists.side_effect = ( lambda path: path != "/run/ankaios/control_interface/input" @@ -96,7 +98,7 @@ def test_connection(): # Test output pipe does not exist with patch("os.path.exists") as mock_exists, pytest.raises( - ControlInterfaceException, match="Control interface output fifo" + ConnectionException, match="Control interface output fifo" ): mock_exists.side_effect = ( lambda path: path != "/run/ankaios/control_interface/output" @@ -107,7 +109,7 @@ def test_connection(): with patch("os.path.exists") as mock_exists, patch( "builtins.open" ) as mock_open_file, pytest.raises( - ControlInterfaceException, match="Error while opening output fifo" + ConnectionException, match="Error while opening output fifo" ): mock_exists.return_value = True mock_open_file.side_effect = OSError @@ -117,7 +119,7 @@ def test_connection(): with patch("os.path.exists") as mock_exists, patch( "threading.Thread" ) as mock_thread, patch("builtins.open") as mock_open_file, patch( - "ankaios_sdk.ControlInterface._send_initial_hello" + "ankaios_sdk.ControlInterfaceConnection._send_initial_hello" ) as mock_initial_hello: mock_exists.return_value = True mock_thread_instance = MagicMock() @@ -163,7 +165,7 @@ def test_connect_clears_disconnect_event(): connect() must clear a disconnect_event left set by a prior disconnect, otherwise the freshly started reader thread would stop immediately. """ - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=lambda _: None, add_log_callback=lambda _: None, add_event_callback=lambda _: None, @@ -172,7 +174,7 @@ def test_connect_clears_disconnect_event(): with patch("os.path.exists", return_value=True), patch( "threading.Thread" ) as mock_thread, patch("builtins.open"), patch( - "ankaios_sdk.ControlInterface._send_initial_hello" + "ankaios_sdk.ControlInterfaceConnection._send_initial_hello" ): mock_thread.return_value = MagicMock() ci.connect() @@ -186,7 +188,7 @@ def test_cleanup_is_idempotent(): _cleanup() may be reached by both the reader thread and a caller; a repeat call must not raise or double-close the handles. """ - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=lambda _: None, add_log_callback=lambda _: None, add_event_callback=lambda _: None, @@ -207,10 +209,38 @@ def test_cleanup_is_idempotent(): assert ci._state == ControlInterfaceState.TERMINATED +def test_decode_response(): + """ + Test the _decode_response static method of the ControlInterfaceConnection + class, which owns the control interface's own envelope + (FromAnkaios) unwrapping, for all 3 possible variants plus the + parsing-error case. + """ + response = ControlInterfaceConnection._decode_response( + MESSAGE_BUFFER_UPDATE_SUCCESS + ) + assert response.content_type == ResponseType.UPDATE_STATE_SUCCESS + + response = ControlInterfaceConnection._decode_response( + MESSAGE_BUFFER_CONTROL_INTERFACE_ACCEPTED + ) + assert response.content_type == ResponseType.CONTROL_INTERFACE_ACCEPTED + + response = ControlInterfaceConnection._decode_response( + MESSAGE_BUFFER_CONNECTION_CLOSED + ) + assert response.content_type == ResponseType.CONNECTION_CLOSED + assert response.content == "Connection closed reason" + + with pytest.raises(ResponseException, match="Parsing error"): + ControlInterfaceConnection._decode_response(b"invalid_buffer{") + + def test_read_thread_general(): """ - Test the _read_from_control_interface method of the ControlInterface class. - Test success and error with the input file. + Test the _read_from_control_interface method of the + ControlInterfaceConnection class. Test success and error with + the input file. """ update_success_content = ( MESSAGE_BUFFER_UPDATE_SUCCESS_LENGTH + MESSAGE_BUFFER_UPDATE_SUCCESS @@ -218,15 +248,15 @@ def test_read_thread_general(): # Test error while opening input pipe with patch("builtins.open", side_effect=OSError), patch( - "ankaios_sdk.ControlInterface._cleanup" + "ankaios_sdk.ControlInterfaceConnection._cleanup" ) as mock_cleanup: - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=lambda _: None, add_log_callback=lambda _: None, add_event_callback=lambda _: None, ) with pytest.raises( - ControlInterfaceException, match="Error while opening input fifo" + ConnectionException, match="Error while opening input fifo" ): ci._read_from_control_interface() # The reader thread tears down directly instead of calling @@ -236,7 +266,7 @@ def test_read_thread_general(): # Test success with patch("builtins.open", mock_open()) as mock_file, patch( - "ankaios_sdk.ControlInterface._handle_response" + "ankaios_sdk.ControlInterfaceConnection._handle_response" ) as mock_handle_response, patch("os.set_blocking") as _, patch( "select.select" ) as mock_select: @@ -246,7 +276,7 @@ def test_read_thread_general(): bytes([b]) for b in update_success_content ] - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=lambda _: None, add_log_callback=lambda _: None, add_event_callback=lambda _: None, @@ -279,7 +309,7 @@ def test_read_thread_agent_disconnected(): with patch("builtins.open", mock_open()) as mock_file, patch( "os.set_blocking" ) as _, patch("select.select") as mock_select, patch( - "ankaios_sdk.ControlInterface._agent_gone_routine" + "ankaios_sdk.ControlInterfaceConnection._agent_gone_routine" ) as mock_agent_gone: # Data is available, but read returns empty @@ -287,7 +317,7 @@ def test_read_thread_agent_disconnected(): mock_file_handle = mock_file.return_value.__enter__.return_value mock_file_handle.read.return_value = b"" - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=lambda _: None, add_log_callback=lambda _: None, add_event_callback=lambda _: None, @@ -325,7 +355,7 @@ def test_read_thread_connection_closed(): ) with patch("builtins.open", mock_open()) as mock_file, patch( - "ankaios_sdk.ControlInterface._handle_response", + "ankaios_sdk.ControlInterfaceConnection._handle_response", side_effect=ConnectionClosedException, ), patch("os.set_blocking") as _, patch("select.select") as mock_select: mock_select.return_value = ([True], [], []) @@ -334,7 +364,7 @@ def test_read_thread_connection_closed(): bytes([b]) for b in connection_closed_content ] - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=lambda _: None, add_log_callback=lambda _: None, add_event_callback=lambda _: None, @@ -364,7 +394,7 @@ def test_handle_response(): """ response = Response(MESSAGE_BUFFER_UPDATE_SUCCESS) response_callback = MagicMock() - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=response_callback, add_log_callback=lambda _: None, add_event_callback=lambda _: None, @@ -410,7 +440,7 @@ def test_handle_response_control_interface_accepted(): # Got control interface accepted response as initial response response_callback = MagicMock() - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=response_callback, add_log_callback=lambda _: None, add_event_callback=lambda _: None, @@ -423,7 +453,7 @@ def test_handle_response_control_interface_accepted(): # Got control interface accepted response while already connected response_callback = MagicMock() - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=response_callback, add_log_callback=lambda _: None, add_event_callback=lambda _: None, @@ -447,7 +477,7 @@ def test_handle_response_connection_closed(): # Got connection closed response as initial response response_callback = MagicMock() - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=response_callback, add_log_callback=lambda _: None, add_event_callback=lambda _: None, @@ -460,7 +490,7 @@ def test_handle_response_connection_closed(): # Got connection closed response while already connected response_callback = MagicMock() - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=response_callback, add_log_callback=lambda _: None, add_event_callback=lambda _: None, @@ -481,7 +511,7 @@ def test_handle_response_logs(): response_callback = MagicMock() logs_callback = MagicMock() - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=response_callback, add_log_callback=logs_callback, add_event_callback=lambda _: None, @@ -503,7 +533,7 @@ def test_handle_response_events(): response_callback = MagicMock() events_callback = MagicMock() - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=response_callback, add_log_callback=lambda _: None, add_event_callback=events_callback, @@ -518,25 +548,26 @@ def test_handle_response_events(): def test_agent_gone_routine(): """ - Test the _agent_gone_routine method of the ControlInterface class. + Test the _agent_gone_routine method of the + ControlInterfaceConnection class. """ - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=lambda _: None, add_log_callback=lambda _: None, add_event_callback=lambda _: None, ) ci._state = ControlInterfaceState.CONNECTED with patch( - "ankaios_sdk.ControlInterface._send_initial_hello" + "ankaios_sdk.ControlInterfaceConnection._send_initial_hello" ) as mock_initial_hello: ci._agent_gone_routine() mock_initial_hello.assert_not_called() ci._state = ControlInterfaceState.AGENT_DISCONNECTED with patch.object( - ControlInterface, "AGENT_RECONNECT_INTERVAL_SEC", 0.01 + ControlInterfaceConnection, "AGENT_RECONNECT_INTERVAL_SEC", 0.01 ), patch( - "ankaios_sdk.ControlInterface._send_initial_hello" + "ankaios_sdk.ControlInterfaceConnection._send_initial_hello" ) as mock_initial_hello: mock_initial_hello.side_effect = BrokenPipeError @@ -560,14 +591,14 @@ def test_agent_gone_routine_stops_on_disconnect(): _agent_gone_routine must exit promptly when a disconnect is requested, even while the agent stays gone. """ - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=lambda _: None, add_log_callback=lambda _: None, add_event_callback=lambda _: None, ) ci._state = ControlInterfaceState.AGENT_DISCONNECTED with patch( - "ankaios_sdk.ControlInterface._send_initial_hello", + "ankaios_sdk.ControlInterfaceConnection._send_initial_hello", side_effect=BrokenPipeError, ): agent_gone_thread = threading.Thread( @@ -583,9 +614,9 @@ def test_agent_gone_routine_stops_on_disconnect(): def test_write_to_pipe(): """ - Test the _write_to_pipe method of the ControlInterface class. + Test the _write_to_pipe method of the ControlInterfaceConnection class. """ - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=lambda _: None, add_log_callback=lambda _: None, add_event_callback=lambda _: None, @@ -593,7 +624,7 @@ def test_write_to_pipe(): ci._output_file = None with pytest.raises( - ControlInterfaceException, match="Could not write to pipe" + ConnectionException, match="Could not write to pipe" ): ci._write_to_pipe(_control_api.FromAnkaios()) @@ -608,9 +639,9 @@ def test_write_to_pipe(): def test_write_request(): """ - Test the write_request method of the ControlInterface class. + Test the write_request method of the ControlInterfaceConnection class. """ - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=lambda _: None, add_log_callback=lambda _: None, add_event_callback=lambda _: None, @@ -619,12 +650,14 @@ def test_write_request(): ci._state = ControlInterfaceState.TERMINATED with pytest.raises( - ControlInterfaceException, match="Could not write to pipe" + ConnectionException, match="Could not write to pipe" ): ci.write_request(test_request) ci._state = ControlInterfaceState.CONNECTED - with patch("ankaios_sdk.ControlInterface._write_to_pipe") as mock_write: + with patch( + "ankaios_sdk.ControlInterfaceConnection._write_to_pipe" + ) as mock_write: ci.write_request(test_request) mock_write.assert_called_once() @@ -640,12 +673,14 @@ def test_send_initial_hello(): """ Test the _send_initial_hello method of the Ankaios class. """ - ci = ControlInterface( + ci = ControlInterfaceConnection( add_response_callback=lambda _: None, add_log_callback=lambda _: None, add_event_callback=lambda _: None, ) - with patch("ankaios_sdk.ControlInterface._write_to_pipe") as mock_write: + with patch( + "ankaios_sdk.ControlInterfaceConnection._write_to_pipe" + ) as mock_write: initial_hello = _control_api.ToAnkaios( hello=_control_api.Hello(protocolVersion=str(ANKAIOS_VERSION)) ) diff --git a/tests/response/test_response.py b/tests/response/test_response.py index a644709..b94ab56 100644 --- a/tests/response/test_response.py +++ b/tests/response/test_response.py @@ -107,13 +107,6 @@ MESSAGE_LOGS_STOP_RESPONSE.SerializeToString() ) - -MESSAGE_BUFFER_LOGS_CANCEL_REQUEST_ACCEPTED = _control_api.FromAnkaios( - response=_ank_base.Response( - requestId="4455", logsCancelAccepted=_ank_base.LogsCancelAccepted() - ) -).SerializeToString() - MESSAGE_BUFFER_LOGS_CANCEL_ACCEPTED_RESPONSE = _control_api.FromAnkaios( response=_ank_base.Response( requestId="4455", @@ -252,6 +245,50 @@ def test_initialisation(): _ = Response(MESSAGE_BUFFER_INVALID_RESPONSE) +def test_from_ank_base_response(): + """ + Test that Response._from_ank_base_response builds the same + content as the byte-parsing path, given an already decoded + ank_base.Response (as handed over by any connection once it has + unwrapped its own envelope). + """ + ank_base_response = _ank_base.Response( + requestId="5566", + error=_ank_base.Error(message="unwrapped error message"), + ) + response = Response._from_ank_base_response(ank_base_response) + assert response.buffer is None + assert response.content_type == ResponseType.ERROR + assert response.content == "unwrapped error message" + assert response.get_request_id() == "5566" + + +def test_control_interface_accepted(): + """ + Test that Response._control_interface_accepted builds a Response + with no ank_base payload, matching the byte-parsing path's + handling of the same envelope variant. + """ + response = Response._control_interface_accepted() + assert response.buffer is None + assert response.content_type == ResponseType.CONTROL_INTERFACE_ACCEPTED + assert response.content is None + assert response.get_request_id() is None + + +def test_connection_closed(): + """ + Test that Response._connection_closed builds a Response with no + ank_base payload but the given reason, matching the byte-parsing + path's handling of the same envelope variant. + """ + response = Response._connection_closed("Connection closed reason") + assert response.buffer is None + assert response.content_type == ResponseType.CONNECTION_CLOSED + assert response.content == "Connection closed reason" + assert response.get_request_id() is None + + def test_getters(): """ Test the getter methods of the Response class. diff --git a/tests/test_ankaios.py b/tests/test_ankaios.py index e7502b3..bd993f3 100644 --- a/tests/test_ankaios.py +++ b/tests/test_ankaios.py @@ -20,11 +20,13 @@ from io import StringIO import logging +import sys from unittest.mock import patch, MagicMock, PropertyMock import pytest from ankaios_sdk import ( Ankaios, AnkaiosLogLevel, + ConnectionType, LogEntry, Response, UpdateStateSuccess, @@ -33,6 +35,7 @@ WorkloadInstanceName, WorkloadStateCollection, WorkloadStateEnum, + ControlInterfaceConnection, ControlInterfaceState, AnkaiosProtocolException, AnkaiosResponseError, @@ -52,7 +55,7 @@ MESSAGE_BUFFER_UPDATE_SUCCESS, MESSAGE_BUFFER_CONNECTION_CLOSED, MESSAGE_BUFFER_LOGS_REQUEST_ACCEPTED, - MESSAGE_BUFFER_LOGS_CANCEL_REQUEST_ACCEPTED, + MESSAGE_BUFFER_LOGS_CANCEL_ACCEPTED_RESPONSE, MESSAGE_BUFFER_EVENTS_CANCEL_ACCEPTED_RESPONSE, ) from tests.test_manifest import MANIFEST_DICT @@ -70,13 +73,16 @@ def generate_test_ankaios() -> Ankaios: Returns: Ankaios: The Ankaios instance. """ - with patch("ankaios_sdk.ControlInterface.connect") as mock_connect, patch( - "ankaios_sdk.ControlInterface.connected", new_callable=PropertyMock + with patch( + "ankaios_sdk.ControlInterfaceConnection.connect" + ) as mock_connect, patch( + "ankaios_sdk.ControlInterfaceConnection.connected", + new_callable=PropertyMock, ) as mock_connected: mock_connected.return_value = True ankaios = Ankaios() mock_connect.assert_called_once() - ankaios._control_interface._state = ControlInterfaceState.CONNECTED + ankaios._connection._state = ControlInterfaceState.CONNECTED return ankaios @@ -104,11 +110,12 @@ def test_connect_disconnect(): Test the connect and disconnect of the Ankaios class. """ with patch( - "ankaios_sdk.ControlInterface.connect" + "ankaios_sdk.ControlInterfaceConnection.connect" ) as mock_ci_connect, patch( - "ankaios_sdk.ControlInterface.connected", new_callable=PropertyMock + "ankaios_sdk.ControlInterfaceConnection.connected", + new_callable=PropertyMock, ) as mock_ci_connected, patch( - "ankaios_sdk.ControlInterface.disconnect" + "ankaios_sdk.ControlInterfaceConnection.disconnect" ) as mock_ci_disconnect: mock_ci_connected.return_value = True with Ankaios() as ankaios: @@ -123,11 +130,12 @@ def test_connection_timeout(): Test the connection timeout case. """ with patch("time.time") as mock_time, patch("time.sleep"), patch( - "ankaios_sdk.ControlInterface.connect" + "ankaios_sdk.ControlInterfaceConnection.connect" ) as mock_ci_connect, patch( - "ankaios_sdk.ControlInterface.disconnect" + "ankaios_sdk.ControlInterfaceConnection.disconnect" ) as _, patch( - "ankaios_sdk.ControlInterface.connected", new_callable=PropertyMock + "ankaios_sdk.ControlInterfaceConnection.connected", + new_callable=PropertyMock, ) as mock_ci_connected: # The first 2 values are needed to call the sleep # The last 2 values are needed to exceed the timeout properly @@ -138,10 +146,80 @@ def test_connection_timeout(): mock_ci_connect.assert_called_once() +def test_create_connection_default_is_control_interface(): + """ + Test that the default connection_type builds a ControlInterfaceConnection. + """ + ankaios = generate_test_ankaios() + assert isinstance(ankaios._connection, ControlInterfaceConnection) + + +def test_create_connection_grpc_missing_server_url_raises(): + """ + Test that using ConnectionType.COMMAND_INTERFACE without a + server_url raises ValueError before any connection is attempted. + """ + with pytest.raises(ValueError, match="server_url is required"): + Ankaios(connection_type=ConnectionType.COMMAND_INTERFACE) + + +def test_create_connection_grpc_missing_dependency_raises_import_error(): + """ + Test that using ConnectionType.COMMAND_INTERFACE without the 'command' + extra installed raises a clear ImportError. + """ + with patch.dict( + sys.modules, + {"ankaios_sdk._components.connection.command_interface": None}, + ): + with pytest.raises( + ImportError, match="pip install ankaios-sdk\\[command\\]" + ): + Ankaios( + connection_type=ConnectionType.COMMAND_INTERFACE, + server_url="http://127.0.0.1:25551", + ) + + +def test_create_connection_grpc_success(): + """ + Test that Ankaios builds and connects a CommandInterfaceConnection + when using ConnectionType.COMMAND_INTERFACE, wiring in its own + callbacks and the given gRPC-specific arguments. + """ + with patch( + "ankaios_sdk._components.connection.command_interface." + "CommandInterfaceConnection" + ) as mock_grpc_connection_cls: + mock_instance = MagicMock() + mock_instance.connected = True + mock_grpc_connection_cls.return_value = mock_instance + + ankaios = Ankaios( + connection_type=ConnectionType.COMMAND_INTERFACE, + server_url="http://127.0.0.1:25551", + ca_pem="ca-secret", + crt_pem="crt-secret", + key_pem="key-secret", + ) + + mock_grpc_connection_cls.assert_called_once_with( + "http://127.0.0.1:25551", + add_response_callback=ankaios._add_response, + add_log_callback=ankaios._add_logs, + add_event_callback=ankaios._add_events, + ca_pem="ca-secret", + crt_pem="crt-secret", + key_pem="key-secret", + ) + mock_instance.connect.assert_called_once() + assert ankaios._connection is mock_instance + + def test_add_response(): """ Test the _add_response method of the Ankaios class. - This method is called from the ControlInterface when a response + This method is called from the ControlInterfaceConnection when a response is received. """ response = Response(MESSAGE_BUFFER_UPDATE_SUCCESS) @@ -157,7 +235,7 @@ def test_add_response(): def test_add_logs(): """ Test the _add_logs method of the Ankaios class. - This method is called from the ControlInterface when a response + This method is called from the ControlInterfaceConnection when a response of type Logs Entries is received. """ log_entries = [ @@ -182,7 +260,7 @@ def test_add_logs(): def test_add_events(): """ Test the _add_events method of the Ankaios class. - This method is called from the ControlInterface when a response + This method is called from the ControlInterfaceConnection when a response of type EventEntry is received. """ event_entry = generate_test_event_entry() @@ -232,7 +310,7 @@ def test_send_request(): request = generate_test_request() with patch( - "ankaios_sdk.ControlInterface.write_request" + "ankaios_sdk.ControlInterfaceConnection.write_request" ) as mock_write, patch( "ankaios_sdk.Ankaios._get_response_by_id" ) as mock_get_response: @@ -243,7 +321,7 @@ def test_send_request(): ) with patch( - "ankaios_sdk.ControlInterface.write_request" + "ankaios_sdk.ControlInterfaceConnection.write_request" ) as mock_write, patch( "ankaios_sdk.Ankaios._get_response_by_id" ) as mock_get_response: @@ -1027,7 +1105,7 @@ def test_stop_receiving_logs(): # Test success with patch("ankaios_sdk.Ankaios._send_request") as mock_send_request: mock_send_request.return_value = Response( - MESSAGE_BUFFER_LOGS_CANCEL_REQUEST_ACCEPTED + MESSAGE_BUFFER_LOGS_CANCEL_ACCEPTED_RESPONSE ) cancel_request = LogsCancelRequest(log_campaign.queue._request_id) ankaios.stop_receiving_logs(log_campaign)