tests, license, ...

Author: mbakalarski

LICENSE

@@ -186,7 +186,7 @@
       same "printed page" as the copyright notice for easier
       identification within third-party archives.
 
-   Copyright [yyyy] [name of copyright owner]
+   Copyright 2025-present Michal Bakalarski and Netclab Contributors
 
    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.

NOTICE

@@ -0,0 +1,9 @@
+function-eapi
+
+Copyright 2025–present Michal Bakalarski and Netclab Contributors
+
+This product includes software derived from
+Crossplane function-template-python
+https://github.com/crossplane/function-template-python
+
+Licensed under the Apache License, Version 2.0.

README.md

@@ -1,46 +1,37 @@
-# function-template-python
+# eAPI Function
 
-[![CI](https://github.com/crossplane/function-template-python/actions/workflows/ci.yml/badge.svg)](https://github.com/crossplane/function-template-go/actions/workflows/ci.yml)
+This Crossplane composition function generates resources for interacting with Arista EOS devices via JSON-RPC.<br>
+It is intended for use in Crossplane Configuration packages to manage device state automatically.
 
-A template for writing a [composition function][functions] in [Python][python].
-
-To learn how to use this template:
-
-* [Follow the guide to writing a composition function in Python][function guide]
-* [Learn about how composition functions work][functions]
-* [Read the function-sdk-python package documentation][package docs]
-
-If you just want to jump in and get started:
-
-1. Replace `function-template-python` with your function's name in
-   `package/crossplane.yaml`.
-1. Add your logic to `RunFunction` in `function/fn.py`
-1. Add tests for your logic in `test/test_fn.py`
-1. Update this file, `README.md`, to be about your function!
-
-This template uses [Python][python], [Docker][docker], and the [Crossplane
-CLI][cli] to build functions.
 
 ```shell
-# Run the code in development mode, for crossplane beta render
+# Run the code in development mode, for crossplane render
 hatch run development
+```
 
+```shell
 # Lint and format the code - see pyproject.toml
 hatch fmt
+```
 
+```shell
 # Run unit tests - see tests/test_fn.py
 hatch test
+```
 
+```shell
 # Build the function's runtime image - see Dockerfile
-$ docker build . --tag=runtime
+docker build . --tag=runtime
+```
 
+```shell
 # Build a function package - see package/crossplane.yaml
-$ crossplane xpkg build -f package --embed-runtime-image=runtime
+crossplane xpkg build -f package --embed-runtime-image=runtime
 ```
 
-[functions]: https://docs.crossplane.io/latest/concepts/composition-functions
-[function guide]: https://docs.crossplane.io/knowledge-base/guides/write-a-composition-function-in-python
-[package docs]: https://crossplane.github.io/function-sdk-python
-[python]: https://python.org
-[docker]: https://www.docker.com
-[cli]: https://docs.crossplane.io/latest/cli
+## License
+
+This project is licensed under the Apache License 2.0.
+
+This function was originally created using the
+Crossplane function-template-python project.

example/README.md

@@ -1,6 +1,6 @@
 # Example manifests
 
-You can run your function locally and test it using `crossplane beta render`
+You can run your function locally and test it using `crossplane render`
 with these example manifests.
 
 ```shell

function/fn.py

@@ -1,13 +1,16 @@
 """A Crossplane composition function."""
 
 import hashlib
+import json
 
 import grpc
 from crossplane.function import logging, request, resource, response
 from crossplane.function.proto.v1 import run_function_pb2 as fnv1
 from crossplane.function.proto.v1 import run_function_pb2_grpc as grpcv1
 from jsonrpcclient import request_json
 
+JSONRPC_BASE = {"version": 1, "format": "json"}
+
 
 class FunctionRunner(grpcv1.FunctionRunnerService):
     """A FunctionRunner handles gRPC RunFunctionRequests."""
@@ -47,29 +50,29 @@ async def RunFunction(
         }
 
         jsonrpc_observe_params = {
-            "version": 1,
-            "format": "json",
+            **JSONRPC_BASE,
             "cmds": ["enable", "show running-config"],
         }
         jsonrpc_observe = request_json("runCmds", params=jsonrpc_observe_params)
 
-        command_paths = walk_cmds(cmds)
+        command_paths = sorted(walk_cmds(cmds))
         log.info("Generated command paths", count=len(command_paths))
 
         for path in command_paths:
             name = name_prefix + "-" + name_from_path(path)
 
+            path_log = log.bind(resource=name, path=" | ".join(path))
+            path_log.debug("Creating resource")
+
             jsonrpc_create_params = {
-                "version": 1,
-                "format": "json",
+                **JSONRPC_BASE,
                 "cmds": ["enable", "configure", *path],
             }
             jsonrpc_create = request_json("runCmds", params=jsonrpc_create_params)
 
             remove_path = [*path[:-1], f"no {path[-1]}"]
             jsonrpc_remove_params = {
-                "version": 1,
-                "format": "json",
+                **JSONRPC_BASE,
                 "cmds": ["enable", "configure", *remove_path],
             }
             jsonrpc_remove = request_json("runCmds", params=jsonrpc_remove_params)
@@ -94,19 +97,25 @@ async def RunFunction(
         return rsp
 
 
+def jq_path(cmds: list[str]) -> str:
+    """Create middle part of jq logic expression."""
+    return "".join([f".cmds[{json.dumps(c)}]" for c in cmds])
+
+
 def create_jq_logic_expressions(path: list[str]) -> tuple[str, str]:
-    """Create jq logic exp for Request."""
-    logic = ".response.body.error == null and ("
-    for cmd in path[:-1]:
-        logic = logic + f'.response.body.result[1].cmds."{cmd}".cmds'
-    logic1 = logic + f' | has("{path[-1]}")' + ")"
-    logic2 = logic + f' | has("{path[-1]}")' + " | not)"
+    """Compose jq logic expressions."""
+    base = ".response.body"
+    tree = f"{base}.result[1]{jq_path(path[:-1])}.cmds"
+    check = f"has({json.dumps(path[-1])})"
+
+    expected = f"{base}.error == null and ({tree} | {check})"
+    removed = f"{base}.error == null and ({tree} | {check} | not)"
 
-    return logic1, logic2
+    return expected, removed
 
 
 def walk_cmds(cmds: dict, path: list[str] | None = None) -> list[list[str]]:
-    """walk_cmds function."""
+    """Make command paths from nested command tree."""
     if path is None:
         path = []
 

pyproject.toml

@@ -9,7 +9,7 @@ readme = "README.md"
 requires-python = ">=3.11,<3.12"
 license = "Apache-2.0"
 keywords = []
-authors = [{ name = "Netclab Contributors" }]
+authors = [{ name = "Michal Bakalarski" }]
 classifiers = [
   "Development Status :: 3 - Alpha",
   "Programming Language :: Python",