Vehicle App Development

• 1: Python Vehicle App Development
• 2: C++ Vehicle App Development
• 3: Vehicle App Integration Testing

1 - Python Vehicle App Development

We recommend that you make yourself familiar with the Vehicle App SDK first, before going through this tutorial.

The following information describes how to develop and test the sample Vehicle App that is included in the Python template repository . You will learn how to use the Vehicle App Python SDK and how to interact with the Vehicle Model.

Once you have completed all steps, you will have a solid understanding of the development workflow and you will be able to reuse the template repository for your own Vehicle App development project.

Develop your first Vehicle App

This section describes how to develop your first Vehicle App. Before you start building a new Vehicle App, make sure you have already read this manual:

• Quickstart

Once you have established your development environment, you will be able to start developing your first Vehicle App.

For this tutorial, you will recreate the Vehicle App that is included with the SDK repository : The Vehicle App allows to change the position of the driver’s seat in the car and also provides its current positions to other applications. A detailed explanation of the use case and the example is available here .

Setting up the basic skeleton of your app

At first, you have to create the main Python script called main.py in /app/src. All the relevant code for your new Vehicle App goes there.

If you’ve created your app development repository from our Python template repository , the Velocitas CLI create command or via digital.auto prototyping a file with this name is already present and can be adjusted to your needs.

Setting up the basic skeleton of an app consists of the following steps:

1. Manage your imports
2. Enable logging
3. Initialize your class
4. Define the entry point of your app

Manage your imports

Before you start development in the main.py you just created, it will be necessary to include the imports required, which you will understand better later through the development:

import asyncio
import json
import logging
import signal

from velocitas_sdk.util.log import (  # type: ignore
    get_opentelemetry_log_factory,
    get_opentelemetry_log_format,
)
from velocitas_sdk.vdb.reply import DataPointReply
from velocitas_sdk.vehicle_app import VehicleApp, subscribe_topic
from vehicle import Vehicle, vehicle  # type: ignore

Enable logging

The following logging configuration applies the default log format provided by the SDK and sets the log level to INFO:

logging.setLogRecordFactory(get_opentelemetry_log_factory())
logging.basicConfig(format=get_opentelemetry_log_format())
logging.getLogger().setLevel("INFO")
logger = logging.getLogger(__name__)

Initialize your class

The main class of your new Vehicle App needs to inherit the VehicleApp provided by the Python SDK .

class MyVehicleApp(VehicleApp):

In class initialization, you have to pass an instance of the Vehicle Model:

def __init__(self, vehicle_client: Vehicle):
    super().__init__()
    self.Vehicle = vehicle_client

We save the vehicle object to use it in our app. Now, you have initialized the app and can continue developing relevant methods.

Entry point of your app

Here’s an example of an entry point to the MyVehicleApp that we just developed:

async def main():
    """Main function"""
    logger.info("Starting my VehicleApp...")
    vehicle_app = MyVehicleApp(vehicle)
    await vehicle_app.run()

LOOP = asyncio.get_event_loop()
LOOP.add_signal_handler(signal.SIGTERM, LOOP.stop)
LOOP.run_until_complete(main())
LOOP.close()

With this your app can now be started. In order to provide some meaningful behaviour of the app, we will enhance it with more features in the next sections.

Vehicle Model Access

In order to facilitate the implementation, the whole vehicle is abstracted into model classes. Please check tutorial about creating models for more details about this topic. In this section, the focus is on using the model.

The first thing you need to do is to get access to the Vehicle Model. If you derived your project repository from our template, we already provide a generated model installed as a Python package named vehicle. Hence, in most cases no additional setup is necessary. How to tailor the model to your needs or how you could get access to vehicle services is described in the tutorial linked above.

If you want to access a single DataPoint e.g. for the vehicle speed, this can be done via

vehicle_speed = (await self.Vehicle.Speed.get()).value

As the get() method of the DataPoint-class there is a coroutine you have to use the await keyword when using it and access its .value.

If you want to get deeper inside the vehicle, to access a single seat for example, you just have to go the model-chain down:

self.DriverSeatPosition = await self.Vehicle.Cabin.Seat.Row1.DriverSide.Position.get()

Subscription to Data Points

If you want to get notified about changes of a specific DataPoint, you can subscribe to this event, e.g. as part of the on_start() method in your app.

    async def on_start(self):
        """Run when the vehicle app starts"""
        await self.Vehicle.Cabin.Seat.Row1.DriverSide.Position.subscribe(
            self.on_seat_position_changed
        )

Every DataPoint provides a .subscribe() method that allows for providing a callback function which will be invoked on every data point update. Subscribed data is available in the respective DataPointReply object and need to be accessed via the reference to the subscribed data point. The returned object is of type TypedDataPointResult which holds the value of the data point and the timestamp at which the value was captured by the Databroker. Therefore the on_seat_position_changed callback function needs to be implemented like this:

    async def on_seat_position_changed(self, data: DataPointReply):
        # handle the event here
        response_topic = "seatadjuster/currentPosition"
        position = data.get(self.Vehicle.Cabin.Seat.Row1.DriverSide.Position).value
        # ...

Subscription using Annotations

The Python SDK also supports annotations for subscribing to data point changes with @subscribe_data_points defined by the whole path to the DataPoint of interest. This would replace the implementation of the Subscription to Data Points

@subscribe_data_points("Vehicle.Cabin.Seat.Row1.DriverSide.Position")
async def on_seat_position_changed(self, data: DataPointReply):
    response_topic = "seatadjuster/currentPosition"
    response_data = {"position": data.get(self.Vehicle.Cabin.Seat.Row1.DriverSide.Position).value}

    await self.publish_event(response_topic, json.dumps(response_data))

Similarly, subscribed data is available in the respective DataPointReply object and needs to be accessed via the reference to the subscribed data point.

Services

Services are used to communicate with other parts of the vehicle via remote function calls (RPC). Please read the basics about them here .

location = SeatLocation(row=1, index=1)
await self.vehicle_client.Cabin.SeatService.MoveComponent(
    location, BASE, data["position"]
    )

In order to define which seat you like to move, you have to pass a SeatLocation object as the first parameter. The second argument specifies the component of the seat to be moved. The possible components are defined in the proto files. The last parameter to be passed into the method is the desired position of the component.

Make sure to use the await keyword when calling service methods, since these methods are asynchronously working coroutines.

MQTT

Interaction with other Vehicle Apps or with the cloud is enabled by using the Mosquitto MQTT Broker. The MQTT broker runs inside a docker container, which is started as part of one of our predefined runtimes .

In the quickstart section about the Vehicle App, you already tested sending MQTT messages to the app. In the previous sections, you generally saw how to use Vehicle Models, DataPoints and Services. In this section, you will learn how to combine them with MQTT.

In order to receive and process MQTT messages inside your app, simply use the @subscribe_topic annotations from the SDK for an additional method on_set_position_request_received() you have to implement:

    @subscribe_topic("seatadjuster/setPosition/request")
    async def on_set_position_request_received(self, data_str: str) -> None:
        logger.info(f"Got message: {data_str!r}")
        data = json.loads(data_str)
        response_topic = "seatadjuster/setPosition/response"
        response_data = {"requestId": data["requestId"], "result": {}}

        # ...

The on_set_position_request_received method will now be invoked every time a message is published to the subscribed topic "seatadjuster/setPosition/response". The message data (string) is provided as parameter. In the example above the data is parsed from json (data = json.loads(data_str)).

In order to publish data to topics, the SDK provides the appropriate convenience method: self.publish_event() which will be added to the on_seat_position_changed callback function from before.

    async def on_seat_position_changed(self, data: DataPointReply):
        response_topic = "seatadjuster/currentPosition"
        position = data.get(self.Vehicle.Cabin.Seat.Row1.DriverSide.Position).value
        await self.publish_event(
            response_topic,
            json.dumps({"position": position}),
        )

The above example illustrates how one can easily publish messages. In this case, every time the seat position changes, the new position is published to seatadjuster/currentPosition

Your main.py should now have a full implementation for class MyVehicleApp(VehicleApp): containing:

• __init__()
• on_start()
• on_seat_position_changed()
• on_set_position_request_received()

and last but not least a main() method to run the app.

Check the seat-adjuster example to see a more detailed implementation including error handling.

UnitTests

Unit testing is an important part of the development, so let’s have a look at how to do that. You can find some example tests in /app/tests/unit. First, you have to import the relevant packages for unit testing and everything you need for your implementation:

from unittest import mock

import pytest
from sdv.vehicle_app import VehicleApp
from sdv_model.Cabin.SeatService import SeatService  # type: ignore
from sdv_model.proto.seats_pb2 import BASE, SeatLocation  # type: ignore

@pytest.mark.asyncio
async def test_for_publish_to_topic():
    # Disable no-value-for-parameter, seems to be false positive with mock lib

    with mock.patch.object(
        VehicleApp, "publish_event", new_callable=mock.AsyncMock, return_value=-1
    ):
        response = await VehicleApp.publish_event(
            str("sampleTopic"), get_sample_request_data()  # type: ignore
        )
        assert response == -1


def get_sample_request_data():
    return {"position": 330, "requestId": 123}

Looking at a test you notice the annotation @pytest.mark.asyncio. This is required if the test is defined as a coroutine. The next step is to create a mock from all the external dependencies. The method takes 4 arguments: first is the object to be mocked, second the method for which you want to modify the return value, third a callable and the last argument is the return value. After creating the mock, you can test the method and check the response. Use asserts to make your test fail if the response does not match. Check the seat-adjuster unit tests to see a more detailed implementation.

See the results

Once the implementation is done, it is time to run and debug the app.

Run your App

In order to run the app:

• Make sure the devenv-runtimes & devenv-devcontainer-setup packages are part of your .velocitas.json (which should be the default).
• Have a correctly configured app/AppManifest.json . See more
• Trigger our automated vehicle model lifecycle . (e. g. velocitas init)
• A runtime needs to be up and running. Read more about it in the run runtime services section.

Now chose one of the options to start the VehicleApp under development:

1. Press F5

or:

1. Press F1
2. Select command Tasks: Run Task
3. Select Local Runtime - Run VehicleApp

Debug your Vehicle App

In the introduction about debugging , you saw how to start a debugging session. In this section, you will learn what is happening in the background.

The debug session launch settings are already prepared for the VehicleApp in /.vscode/launch.json.

{
    "configurations": [
        {
            "type": "python",
            "justMyCode": false,
            "request": "launch",
            "name": "VehicleApp",
            "program": "${workspaceFolder}/app/src/main.py",
            "console": "integratedTerminal",
            "env": {
                "SDV_MIDDLEWARE_TYPE": "native",
                "SDV_VEHICLEDATABROKER_ADDRESS": "grpc://127.0.0.1:55555",
                "SDV_MQTT_ADDRESS": "mqtt://127.0.0.1:1883"
            }
        }
    ]
}

We specify which python-script to run using the program key.

You can adapt the configuration in /.vscode/launch.json and in /.vscode/tasks.json to your needs (e.g., change the ports, add new tasks) or even add a completely new configuration for another Vehicle App.

Once you are done, you have to switch to the debugging tab (sidebar on the left) and select your configuration using the dropdown on the top. You can now start the debug session by clicking the play button or F5. Debugging is now as simple as in every other IDE, just place your breakpoints and follow the flow of your Vehicle App.

Next steps

• Concept: SDK Overview
• Tutorial: Deploy runtime services in Kanto
• Tutorial: Start runtime services locally
• Tutorial: Creating a Python Vehicle Model
• Tutorial: Develop and run integration tests for a Vehicle App
• Concept: Deployment Model

2 - C++ Vehicle App Development

We recommend that you make yourself familiar with the Vehicle App SDK first, before going through this tutorial.

The following information describes how to develop and test the sample Vehicle App that is included in the C++ template repository . You will learn how to use the Vehicle App C++ SDK and how to interact with the Vehicle Model.

Once you have completed all steps, you will have a solid understanding of the development workflow and you will be able to reuse the template repository for your own Vehicle App development project.

Develop your first Vehicle App

This section describes how to develop your first Vehicle App. Before you start building a new Vehicle App, make sure you have already read this manual:

• Quickstart

For this tutorial, you will recreate the Vehicle App that is included in the template repository : The Vehicle App allows you to change the position of the driver’s seat in the car and also provides its current positions to other applications. A detailed explanation of the use case and the example is available here .

Setting up the basic skeleton of your app

At first, you have to create the main C++ file which we will call App.cpp in /app/src. All the relevant code for your new Vehicle App goes there. Afterwards, there are several steps you need to consider when developing the app:

1. Manage your includes
2. Initialize your class
3. Define the entry point of your app

Manage your includes

Before you start development in the App.cpp you just created, it will be necessary to include all required header files, which you will understand better later through the development:

#include "sdk/VehicleApp.h"
#include "sdk/IPubSubClient.h"
#include "sdk/IVehicleDataBrokerClient.h"
#include "sdk/Logger.h"

#include "vehicle/Vehicle.hpp"

#include <memory>

using namespace velocitas;

Initialize your class

The main class of your new Vehicle App needs to inherit the VehicleApp provided by the C++ SDK .

class MyVehicleApp : public VehicleApp {
public:
    // <remaining code in this tutorial goes here>
private:
    vehicle::Vehicle Vehicle; // this member exists to provide simple access to the vehicle model
}

In your constructor, you have to choose which implementations to use for the VehicleDataBrokerClient and the PubSubClient. By default we suggest you use the factory methods to generate the default implementations: IVehicleDataBrokerClient::createInstance and IPubSubClient::createInstance. These will create a VehicleDataBrokerClient which connects to the VAL via gRPC and an MQTT-based pub-sub client.

MyVehicleApp()
    : VehicleApp(IVehicleDataBrokerClient::createInstance("vehicledatabroker"), // this is the app-id of the KUKSA Databroker in the VAL.
                 IPubSubClient::createInstance("MyVehicleApp")) // the clientId identifies the client at the pub/sub broker
    {}
{}

Now, you have initialized the app and can continue developing relevant methods.

Entry point of your app

Here’s an example of an entry point to the MyVehicleApp that we just developed:

int main(int argc, char** argv) {
    MyVehicleApp app;
    app.run();
    return 0;
}

With this your app can now be started. In order to provide some meaningful behaviour of the app, we will enhance it with more features in the next sections.

Vehicle Model Access

In order to facilitate the implementation, the whole set of vehicle signals is abstracted into model classes. Please check the tutorial about creating models for more details. In this section, the focus is on using the model.

The first thing you need to do is to get access to the Vehicle Model. If you derived your project repository from our template, we already provide a generated model as a Conan source package. The library is already referenced as “include folder” in the CMake files. Hence, in most cases no additional setup is necessary. How to tailor the model to your needs or how you could get access to vehicle services is described in the tutorial linked above. In your source code the model is included via #include "vehicle/Vehicle.hpp" (as shown above).

If you want to read a single signal/data point e.g. for the vehicle speed, the simplest way is to do it via a blocking call and directly accessing the value of the speed:

auto vehicleSpeed = Vehicle.Speed.get()->await().value();

Lets have a look, what this line contains:

• The term Vehicle.Speed addresses the signal we like to query, i.e. the current speed of the vehicle.
• The term .get() tells that we want to get/read the current state of that signal from the Data Broker. Behind the scenes this triggers a request-response flow via IPC with the Data Broker.
• The term ->await() blocks the execution until the response was received.
• Finally, the term .value() tries to access the returned speed value.

The get() returns a shared_ptr to an AsyncResult which, as the name implies, is the result of an asynchronous operation. We have two options to access the value of the asynchronous result. First we can use await() and block the calling thread until a result is available or use onResult(...) which allows us to inject a function pointer or a lambda which is called once the result is available:

Vehicle.Speed.get()
    ->onResult([](auto vehicleSpeed){
        logger().info("Got speed!");
    })
    ->onError(auto status){
        logger().info("Something went wrong communicating to the data broker!");
    });

If you want to get deeper inside to the vehicle, to access a single seat for example, you just have to go the model-chain down:

auto driverSeatPosition = Vehicle.Cabin.Seat.Row1.Pos1.Position.get()->await();

Class TypedDataPointValue

If you have a detailed look at the AsyncResult class, you will observe that the object returned by the await() or passed to the onResult callback is not directly the current value of the signal, but instead an object of type TypedDataPointValue. This object does not only contain the current value of the signal but also some additional metadata accessible via these functions:

• getPath() provides the signal name, i.e. the complete path,
• getType() provides the data type of the signal,
• getTimeStamp() provides the timestamp when the current state was received by the data broker,
• isValid() returns true if the current state represents a valid value of the signal or false otherwise,
• getFailure() returns the reason, why the current state does not represent a valid value (it returns NONE if the value is valid),
• getValue() returns the a valid current value. It will throw an InvalidValueException if the current value is invalid for whatever reason.

The latter three points lead us to the next chapter.

Failure Handling

As indicated above, there might be reasons/situations why the get operation is not able to deliver a valid value for the requested signal. Those shall be handled properly by any application (that wants “to be more” than a prototype).

There are two ways to handle the failure situations:

• Either you can catch the exception thrown by the .value() function:

try {
    auto vehicleSpeed = Vehicle.Speed.get()->await().value();
    // use the speed value
} catch (AsyncException& e) {
    // thrown by the await(): Something went wrong on communication level with the data broker
} catch (InvalidValueException& e) {
    // thrown by .value(): The vehicle speed signal does not contain a valid value, currently
}

• Throwing the InvalidValueException can be avoided if you first check that .isValid() returns true before calling .value():

auto vehicleSpeed = Vehicle.Speed.get()->await();
if (vehicleSpeed.isValid())
    // Accessing .value() now wont throw an exception
    auto speed = vehicleSpeed.value()
    ...
} else {
    // Do your failure handling here
    switch (vehicleSpeed.getFailure()) {
    case Failure::INVALID_VALUE:
        ...
        break;
    case ...
    default:
         ...
    }
}

(isValid() is a convenience function for checking .getFailure() == Failure::NONE.)

Failure Reasons

There are two levels where errors accessing signal/data points might occure.

Communication with the Data Broker (IPC Level)

The data broker might be (temporarly) unavailable because

• it’s not yet started up,
• temporary “stopped” due to a crash or a “live update”,
• some temporary network issues (if running on a different hardware node),
• …

Errors on the IPC level between the application and the data broker will be reported either via

• an AsyncException thrown by the await() function of the AsyncResult class or
• calling the function passed to the onError function of the AsyncResult/AsyncSubscription class.

Errors on this level always make the overall call fail: If getting/setting multiple data points in a single call, the overall operation will fail. In case of setting multiple signals/data points, this means that none of the signals/data points are set. In case of an error on a subscription, this means that the overall subscription could not be established or is terminated now.

Signal / Data Point Level

Failures on signal/data point level are always reported individually per signal/data point. If accessing multiple signals/data points in a single call, getting or setting some certain signal might be successfully done but another one will report an error or failure.

The reasons why a valid value of signal/data point can be missing are explained here:

It is the application developer’s decision if it makes sense to distinguish between the different failure reasons or if some or all of them can be handled as “just one”.

Subscription to DataPoints

If you want to get notified about changes of a specific DataPoint, you can subscribe to this event, e.g. as part of the onStart() method in your app:

void onStart() override {
    subscribeDataPoints(QueryBuilder::select(Vehicle.Cabin.Seat.Row1.Pos1.Position).build())
        ->onItem([this](auto&& item) { onSeatPositionChanged(std::forward<decltype(item)>(item)); })
        ->onError([this](auto&& status) { onError(std::forward<decltype(status)>(status)); });
}

void onSeatPositionChanged(const DataPointsResult& result) {
    const auto dataPoint = result.get(Vehicle.Cabin.Seat.Row1.Pos1.Position);
    logger().info(dataPoint->value());
    // do something with the data point value
}

The VehicleApp class provides the subscribeDataPoints() method which allows to listen for changes on one or multiple data points. Once a change in any of the data points is registered, the callback registered via AsyncSubscription::onItem() is called. Conversely, the callback registered via AsyncSubscription::onError() is called once there is an error during communication with the KUKSA Databroker.

The result passed to the callback registered via onItem() is an object of type DataPointsResult which holds the current state of all data points that were part of the respective subscription. The state of individual data points can be accessed by their reference: result.get(Vehicle.Cabin.Seat.Row1.Pos1.Position))

Services

Services are used to communicate with other parts of the vehicle via remote procedure calls (RPC). Please read the basics about them here .

The following code snippet shows how to use the moveComponent() method of the SeatService from the vehicle model:

vehicle::cabin::SeatService::SeatLocation location{1, 1};
Vehicle.Cabin.SeatService.moveComponent(
    location, vehicle::cabin::SeatService::Component::Base, 300
    )->await();

In order to define which seat you like to move, you have to pass a SeatLocation object as the first parameter. The second argument specifies the component of the seat to be moved. The possible components are defined in the proto-files. The last parameter to be passed into the method is the final position of the component.

Make sure to call the await() method when calling service methods or register a callback via onResult() otherwise you don’t know when your asynchronous call will finish.

MQTT

Interaction with other Vehicle Apps or with the cloud is enabled by using the Mosquitto MQTT Broker. When using the provided template repository you can start a MQTT Broker as part the local runtime. More information can be found here .

In the quickstart section about the Vehicle App, you already tested sending MQTT messages to the app. In the previous sections, you generally saw how to use Vehicle Models, DataPoints and GRPC Services. In this section, you will learn how to combine them with MQTT.

In order to receive and process MQTT messages inside your app, simply use the VehicleApp::subscribeTopic(<topic>) method provided by the SDK:

void onStart() override {
    subscribeTopic("seatadjuster/setPosition/request")
        ->onItem([this](auto&& item){ onSetPositionRequestReceived(std::forward<decltype(item)>(item);)});
}

void onSetPositionRequestReceived(const std::string& data) {
    const auto jsonData = nlohmann::json::parse(data);
    const auto responseTopic = "seatadjuster/setPosition/response";
    nlohmann::json respData({{"requestId", jsonData["requestId"]}, {"result", {}}});
}

The onSetPositionRequestReceived method will now be invoked every time a message is created on the subscribed topic seatadjuster/setPosition/response. The message data is provided as a string parameter. In the example above the data is parsed to json (data = json.loads(data_str)).

In order to publish data to other subscribers, the SDK provides the appropriate convenience method: VehicleApp::publishToTopic(...)

void MyVehicleApp::onSeatPositionChanged(const DataPointsResult& result):
    const auto responseTopic = "seatadjuster/currentPosition";
    nlohmann::json respData({"position": result.get(Vehicle.Cabin.Seat.Row1.Pos1.Position)->value()});

    publishToTopic(
        responseTopic,
        respData.dump(),
    );

The above example illustrates how one can easily publish messages. In this case, every time the seat position changes, the new position is published to seatadjuster/currentPosition

See the results

Once the implementation is done, it is time to run and debug the app.

Build your App

Before you can run the Vehicle App you need to build it first. To do so, simply run the provided build.sh script found in the root of the SDK. It does accept some arguments, but that is out of scope for this tutorial.

Run your App

In order to run the app make sure the devenv-runtimes package is part of your .velocitas.json (which should be the default) and the runtime is up and running. Read more about it in the run runtime services section.

Now chose one of the options to start the VehicleApp under development:

1. Press F5

or:

1. Press F1
2. Select command Tasks: Run Task
3. Select Local Runtime - Run VehicleApp

Debug your Vehicle App

In the introduction about debugging , you saw how to start a debugging session. In this section, you will learn what is happening in the background.

The debug session launch settings are already prepared for the VehicleApp.

{
    "configurations": [
        {
            "name": "VehicleApp - Debug (Native)",
            "type": "cppdbg",
            "request": "launch",
            "program": "${workspaceFolder}/build/bin/app",
            "args": [ ],
            "stopAtEntry": false,
            "cwd": "${workspaceFolder}",
            "environment": [
                {
                    "name": "SDV_MIDDLEWARE_TYPE",
                    "value": "native"
                },
                {
                    "name": "SDV_VEHICLEDATABROKER_ADDRESS",
                    "value": "127.0.0.1:55555"
                },
                {
                    "name": "SDV_MQTT_ADDRESS",
                    "value": "127.0.0.1:1883"
                }
            ],
            "externalConsole": false,
            "MIMode": "gdb",
            "setupCommands": [ ],
        }
    ]
}

We specify which binary to run using the program key. In the environment you can specify all needed environment variables. You can adapt the JSON to your needs (e.g., change the ports, add new tasks) or even add a completely new configuration for another Vehicle App.

Once you are done, you have to switch to the debugging tab (sidebar on the left) and select your configuration using the dropdown on the top. You can now start the debug session by clicking the play button or F5. Debugging is now as simple as in every other IDE, just place your breakpoints and follow the flow of your Vehicle App.

Next steps

• Concept: SDK Overview
• Concept: Deployment Model
• Tutorial: Deploy runtime services in Kanto
• Tutorial: Start runtime services locally
• Tutorial: Creating a Vehicle Model
• Tutorial: Develop and run integration tests for a Vehicle App

3 - Vehicle App Integration Testing

To be sure that a newly created Vehicle App will run together with the KUKSA Databroker and potentially other dependant Vehicle Services or Vehicle Apps, it’s essential to write integration tests along with developing the app.

To execute an integration test, the dependant components need to be running and be accessible from the test runner. This guide will describe how integration tests can be written and integrated in the CI pipeline so that they are executed automatically when building the application.

Writing Test Cases

To write an integration test, you should check the sample that comes with the template ( /app/tests/integration/integration_test.py ). To support interacting with the MQTT broker and the KUKSA Databroker (to get and set values for data points), there are two classes present in Python SDK that will help:

• MqttClient: this class provides methods for interacting with the MQTT broker. Currently, the following methods are available:

  • publish_and_wait_for_response: publishes the specified payload to the given request topic and waits (till timeout) for a message to the response topic. The payload of the first message that arrives in the response topic will be returned. If the timeout expires before, an empty string ("") is returned.

  • publish_and_wait_for_property: publishes the specified payload to the given request topic and waits (till timeout) until the given property value is found in an incoming message to the response topic. The path describes the property location within the response message, the value the property value to look for.

    Example:

    {
        "status": "success",
        "result": {
            "responsecode": 10
        }
    }
    

    If the responsecode property should be checked for the value 10, the path would be ["result", "responsecode"], property value would be 10. When the requested value has been found in a response message, the payload of that message will be returned. If the timeout expires before receiving a matching message, an empty string ("") is returned.

  This class can be initialized with a given port. If no port is specified, the environment variable MQTT_PORT will be checked. If this is not possible either, the default value of 1883 will be used. It’s recommended to specify no port when initializing that class as it will locally use the default port 1883 and in CI the port is set by the environment variable MQTT_PORT. This will prevent a check-in in the wrong port during local development.

• IntTestHelper: this class provides functionality to interact with the KUKSA Databroker.

  • register_datapoint: registers a new data point with given name and type ( here you can find more information about the available types)
  • set_..._datapoint: set the given value for the data point with the given name (with given type). If the data point does not exist, it will be registered.

  This class can be initialized with a given port. If no port is specified, the environment variable VDB_PORT will be checked. If this is not possible either, the default value of 55555 will be used. It’s recommended to specify no port when initializing that class as it will locally use the default port 55555 and in CI the port is set by the environment variable VDB_PORT. This will prevent a check-in in the wrong port during local development.

Runtime components

To be able to test the Vehicle App in an integrated way, the following components should be running:

• Mosquitto
• Databroker
• Vehicle Mock Provider

We distinguish between two environments for executing the Vehicle App and the runtime components:

• Local execution: components are running locally in the development environment
• Kanto execution: components (and application) are deployed and running in a Kanto control plane

Local execution

First, make sure that the runtime services are configured and running like described here .

The application itself can be executed by using a Visual Studio Launch Config (by pressing F5) or by executing the provided task Local Runtime - Run VehicleApp.

When the runtime services and the application are running, integration tests can be executed locally via

  pytest ./app/tests/integration

or using the testing tab in the sidebar to the left.

Kanto runtime

First, make sure that the runtime and the services are up and running, like described here .

The application itself can be deployed by executing the provided task Kanto Runtime - Deploy VehicleApp or Kanto Runtime - Deploy VehicleApp (without rebuild). Depending on whether your app is already available as a container or not.

When the runtime services and the application are running, integration tests can be executed locally via

  pytest ./app/tests/integration

or using the testing tab in the sidebar to the left.

Integration Tests in CI pipeline

The tests will be discovered and executed automatically in the provided CI pipeline . The job Run Integration Tests contains all steps to set up and execute all integration tests in the Kanto runtime. Basically it is doing the same steps as you saw above:

1. start the Kanto runtime
2. deploy the Vehicle App container
3. set the correct MQTT and Databroker ports
4. execute the integration tests

Finally the test results are collected and published as artifacts of the workflow.

Troubleshooting

Troubleshoot IntTestHelper

• Make sure that the KUKSA Databroker is up and running by checking the task log.
• Make sure that you are using the right ports.
• Make sure that you installed the correct version of the SDK (SDV-package).

Troubleshoot Mosquitto (MQTT Broker)

• Make sure that Mosquitto is up and running by checking the task log.
• Make sure that you are using the right ports.
• Use VsMqtt extension to connect to MQTT broker locally (localhost:1883) to monitor topics in MQTT broker by subscribing to all topics using #.

Next steps

• Concept: Deployment Model
• Concept: Build and release process