Feature: Add support for app manifest v3 (#30)

Author: doosuu

.github/workflows/ci.yml

@@ -105,7 +105,15 @@ jobs:
 
       - name: Prepare .velocitas.json for integration test
         run: |
-          NEW_CONFIG="$(jq --arg GITHUB_SHA "$GITHUB_SHA" '.packages[0].version |= $GITHUB_SHA' test/${{ matrix.language }}/.velocitas.json)"
+          COMMIT_REF=$GITHUB_SHA
+
+          if [ "$GITHUB_EVENT_NAME" = "pull_request" ];
+          then
+            echo "Running in context of a PR!"
+            COMMIT_REF=$GITHUB_HEAD_REF
+          fi
+
+          NEW_CONFIG="$(jq --arg COMMIT_REF "$COMMIT_REF" '.packages[0].version |= $COMMIT_REF' test/${{ matrix.language }}/.velocitas.json)"
           echo "${NEW_CONFIG}" > test/${{ matrix.language }}/repo/.velocitas.json
 
       - name: Init velocitas project

setup/src/python/common/devcontainer.json

@@ -79,7 +79,6 @@
 				"ms-azuretools.vscode-docker",
 				"ms-python.python",
 				"cschleiden.vscode-github-actions",
-				"pspester.pester-test",
 				"rpdswtk.vsmqtt",
 				"dotjoshjohnson.xml",
 				"ms-kubernetes-tools.vscode-kubernetes-tools",

vehicle-model-lifecycle/src/download_vspec.py

@@ -17,9 +17,12 @@
 import json
 import os
 import re
+from typing import Any, Dict, List
 
 import requests
 
+FUNCTIONAL_INTERFACE_TYPE_KEY = "vehicle-signal-interface"
+
 
 def require_env(name: str) -> str:
     """Require and return an environment variable.
@@ -84,21 +87,104 @@ def download_file(uri: str, local_file_path: str):
                 outfile.write(chunk)
 
 
-def get_vehicle_model_src():
-    manifest_data_str = require_env("VELOCITAS_APP_MANIFEST")
-    manifest_data = json.loads(manifest_data_str)
+def is_legacy_app_manifest(app_manifest_dict: Dict[str, Any]) -> bool:
+    """Check if the used app manifest is a legacy file.
+
+    Args:
+        app_manifest_dict (Dict[str, Any]): The app manifest.
+
+    Returns:
+        bool: True if the app manifest is a legacy file, False if not.
+    """
+    return "manifestVersion" not in app_manifest_dict
+
+
+def get_legacy_model_src(app_manifest_dict: Dict[str, Any]) -> str:
+    """Get the source from the legacy vehicle model (app manifest < v3)
+
+    Args:
+        app_manifest_dict (Dict[str, Any]): The app manifest dict.
 
+    Returns:
+        str: The source URI of the vehicle model
+    """
     possible_keys = ["vehicleModel", "VehicleModel"]
 
     for key in possible_keys:
-        if key in manifest_data:
-            return manifest_data[key]["src"]
+        if key in app_manifest_dict:
+            return app_manifest_dict[key]["src"]
 
     raise KeyError("App manifest does not contain a valid vehicle model!")
 
 
-def main():
-    vspec_src = get_vehicle_model_src()
+def is_proper_interface_type(interface: Dict[str, Any]) -> bool:
+    """Return if the interface is of the correct type.
+
+    Args:
+        interface (Dict[str, Any]): The interface to check.
+
+    Returns:
+        bool: True if the type matches, False otherwise.
+    """
+    return "type" in interface and interface["type"] == FUNCTIONAL_INTERFACE_TYPE_KEY
+
+
+def get_vehicle_signal_interfaces(
+    app_manifest_dict: Dict[str, Any]
+) -> List[Dict[str, Any]]:
+    """Return all vehicle signal interfaces in the app manifest.
+
+    Args:
+        app_manifest_dict (Dict[str, Any]): The app manifest.
+
+    Returns:
+        List[Dict[str, Any]]: List containing all functional interfaces
+            of type {FUNCTIONAL_INTERFACE_TYPE_KEY}
+    """
+    interfaces = list()
+
+    for interface in app_manifest_dict["interfaces"]:
+        if is_proper_interface_type(interface):
+            interfaces.append(interface)
+
+    return interfaces
+
+
+def get_vehicle_signal_interface_src(interface: Dict[str, Any]) -> str:
+    """Return the URI of the source for the Vehicle Signal Interface.
+
+    Args:
+        interface (Dict[str, Any]): The interface.
+
+    Returns:
+        str: The URI of the source for the Vehicle Signal Interface.
+    """
+    return interface["config"]["src"]
+
+
+def main(app_manifest_dict: Dict[str, Any]):
+    """Entry point for downloading the vspec file for a
+    vehicle-signal-interface.
+
+    Args:
+        app_manifest_dict (Dict[str, Any]): The app manifest.
+
+    Raises:
+        KeyError: If there are multiple vehicle signal interfaces defined
+            in the app manifest.
+    """
+    if is_legacy_app_manifest(app_manifest_dict):
+        vspec_src = get_legacy_model_src(app_manifest_dict)
+    else:
+        interfaces = get_vehicle_signal_interfaces(app_manifest_dict)
+        if len(interfaces) > 1:
+            raise KeyError(
+                f"Only up to one {FUNCTIONAL_INTERFACE_TYPE_KEY!r} supported!"
+            )
+        elif len(interfaces) == 0:
+            return
+        vspec_src = get_vehicle_signal_interface_src(interfaces[0])
+
     local_vspec_path = os.path.join(
         get_velocitas_workspace_dir(), os.path.normpath(vspec_src)
     )
@@ -113,4 +199,7 @@ def main():
 
 
 if __name__ == "__main__":
-    main()
+    manifest_data_str = require_env("VELOCITAS_APP_MANIFEST")
+    app_manifest_dict = json.loads(manifest_data_str)
+
+    main(app_manifest_dict)

vehicle-model-lifecycle/src/generate_model.py

@@ -107,12 +107,22 @@ def install_model_if_required(language: str, model_path: str) -> None:
         subprocess.check_call([sys.executable, "-m", "pip", "install", model_path])
 
 
-if __name__ == "__main__":
-    model_src_file = json.loads(require_env("VELOCITAS_CACHE_DATA"))["vspec_file_path"]
+def main():
+    """Main entry point for generation of vehicle models."""
+    cache_data = json.loads(require_env("VELOCITAS_CACHE_DATA"))
+
+    if "vspec_file_path" not in cache_data:
+        return
+
+    model_src_file = cache_data["vspec_file_path"]
     model_language = require_env("language")
     model_output_dir = get_model_output_dir()
     os.makedirs(model_output_dir, exist_ok=True)
 
     remove_old_model(model_output_dir)
     invoke_generator(model_src_file, model_language, model_output_dir)
     install_model_if_required(model_language, model_output_dir)
+
+
+if __name__ == "__main__":
+    main()

vehicle-model-lifecycle/test/test_download_vspec.py

@@ -19,10 +19,24 @@
 from test_lib import capture_stdout, mock_env
 
 sys.path.append(os.path.join(os.path.dirname(__file__), "..", "src"))
-from download_vspec import download_file, get_vehicle_model_src, is_uri, main  # noqa
+from download_vspec import (  # noqa
+    download_file,
+    get_legacy_model_src,
+    is_proper_interface_type,
+    is_uri,
+    main,
+    require_env,
+)
 
 vspec_300_uri = "https://github.com/COVESA/vehicle_signal_specification/releases/download/v3.0/vss_rel_3.0.json"  # noqa
-app_manifest = {"VehicleModel": {"src": ""}}
+
+
+def get_vspec_file_path_value(capturedOutput: str) -> str:
+    return (
+        capturedOutput.split("vspec_file_path=")[1]
+        .split(" >> VELOCITAS_CACHE")[0]
+        .replace("'", "")
+    )
 
 
 def test_is_uri():
@@ -38,48 +52,100 @@ def test_download_file():
     assert os.path.isfile(local_file_path)
 
 
-def test_camel_case_vehicle_model_key():
+def test_get_legacy_model_src__camel_case_vehicle_model_key():
     app_manifest_dict = {"vehicleModel": {"src": "foo"}}
-    with mock_env(app_manifest_dict):
-        assert get_vehicle_model_src() == "foo"
+    assert get_legacy_model_src(app_manifest_dict) == "foo"
 
 
-def test_pascal_case_vehicle_model_key():
+def test_get_legacy_model_src__pascal_case_vehicle_model_key():
     app_manifest_dict = {"VehicleModel": {"src": "bar"}}
-    with mock_env(app_manifest_dict):
-        assert get_vehicle_model_src() == "bar"
+    assert get_legacy_model_src(app_manifest_dict) == "bar"
 
 
-def test_invalid_vehicle_model_key():
+def test_get_legacy_model_src__invalid_vehicle_model_key():
     app_manifest_dict = {"Vehicle.Model": {"src": "baz"}}
-    with mock_env(app_manifest_dict):
-        with pytest.raises(KeyError):
-            get_vehicle_model_src()
+    with pytest.raises(KeyError):
+        get_legacy_model_src(app_manifest_dict)
+
+
+def test_proper_interface_type__wrong_type():
+    assert not is_proper_interface_type({"type": "foo"})
+
 
+def test_proper_interface_type__no_type():
+    assert not is_proper_interface_type({"notype": "foo"})
 
-def test_int_relative_src_converted_to_absolute():
-    app_manifest["VehicleModel"]["src"] = "./app/vspec.json"
-    with capture_stdout() as capture, mock_env(app_manifest):
-        main()
+
+def test_proper_interface_type__correct_type():
+    assert is_proper_interface_type({"type": "vehicle-signal-interface"})
+
+
+def test_require_env__var_not_present__raises_error():
+    with pytest.raises(ValueError):
+        require_env("foo")
+
+
+@pytest.mark.parametrize(
+    "app_manifest",
+    [
+        {
+            "manifestVersion": "v3",
+            "interfaces": [
+                {
+                    "type": "vehicle-signal-interface",
+                    "config": {"src": "./app/vspec.json"},
+                }
+            ],
+        },
+        {"VehicleModel": {"src": "./app/vspec.json"}},
+    ],
+)
+def test_main__relative_src__converted_to_absolute(app_manifest):
+    with capture_stdout() as capture, mock_env():
+        main(app_manifest)
 
         vspec_file_path = get_vspec_file_path_value(capture.getvalue())
         assert os.path.isabs(vspec_file_path)
         assert vspec_file_path == "/workspaces/my_vehicle_app/app/vspec.json"
 
 
-def test_int_uri_src_downloaded_and_stored_in_cache():
-    app_manifest["VehicleModel"]["src"] = vspec_300_uri
-    with capture_stdout() as capture, mock_env(app_manifest):
-        main()
+@pytest.mark.parametrize(
+    "app_manifest",
+    [
+        {
+            "manifestVersion": "v3",
+            "interfaces": [
+                {"type": "vehicle-signal-interface", "config": {"src": vspec_300_uri}}
+            ],
+        },
+        {"VehicleModel": {"src": vspec_300_uri}},
+    ],
+)
+def test_main__valid_app_manifest__uri_src_downloaded_and_stored_in_cache(app_manifest):
+    with capture_stdout() as capture, mock_env():
+        main(app_manifest)
 
         vspec_file_path = get_vspec_file_path_value(capture.getvalue())
         assert os.path.isabs(vspec_file_path)
         assert vspec_file_path == "/tmp/velocitas/vspec.json"
 
 
-def get_vspec_file_path_value(capturedOutput: str) -> str:
-    return (
-        capturedOutput.split("vspec_file_path=")[1]
-        .split(" >> VELOCITAS_CACHE")[0]
-        .replace("'", "")
-    )
+def test_main__duplicate_vehicle_signal_interface__raises_error():
+    app_manifest = {
+        "manifestVersion": "v3",
+        "interfaces": [
+            {"type": "vehicle-signal-interface", "config": {"src": vspec_300_uri}},
+            {"type": "vehicle-signal-interface", "config": {"src": vspec_300_uri}},
+        ],
+    }
+    with pytest.raises(KeyError):
+        main(app_manifest)
+
+
+def test_main__no_vehicle_signal_interface__silently_exists():
+    app_manifest = {
+        "manifestVersion": "v3",
+        "interfaces": [{"type": "pubsub", "config": {}}],
+    }
+    with capture_stdout(), mock_env():
+        main(app_manifest)

vehicle-model-lifecycle/test/test_lib.py

@@ -55,5 +55,5 @@ def __exit__(self, _type, _value, _traceback):
         del os.environ["VELOCITAS_WORKSPACE_DIR"]
 
 
-def mock_env(app_manifest) -> MockEnv:
+def mock_env(app_manifest=None) -> MockEnv:
     return MockEnv(app_manifest)