Added TS support

README.md

@@ -89,7 +89,7 @@ print_result(result)
 
 ### Tasks
 
-* __atc__: Sends F5 Automation Toolchain (ATC) declaratives (like AS3, DO and TS) on a BIG-IP/BIG-IQ system.
+* __atc__: Sends F5 Automation Toolchain (ATC) declaratives (like AS3, DO and TS) on a BIG-IP/IQ system.
 * __bigip_cm_config_sync__: Synchronizes the configuration between BIG-IP systems.
 * __bigip_cm_failover_status__: Gets the failover status of the BIG-IP system.
 * __bigip_cm_sync_status__: Gets the configuration synchronization status of the BIG-IP system.
@@ -102,7 +102,6 @@ print_result(result)
 ## Roadmap
 
 * ATC:
-  * Support Telemetry (TS)
   * Support BIG-IQ
 * Support dry-run mode
 

nornir_f5/__init__.py

@@ -1,3 +1,3 @@
 """Nornir F5."""
 
-__version__ = "0.2.0"
+__version__ = "0.3.0"

nornir_f5/plugins/tasks/atc.py

@@ -43,17 +43,17 @@
             },
         }
     },
-    # "Telemetry": {
-    #     "endpoints": {
-    #         "configure": {
-    #             "uri": "/mgmt/shared/telemetry/declare",
-    #             "methods": ["GET", "POST"],
-    #         },
-    #         "info": {"uri": "/mgmt/shared/telemetry/info", "methods": ["GET"]},
-    #     }
-    # },
+    "Telemetry": {
+        "endpoints": {
+            "configure": {
+                "uri": "/mgmt/shared/telemetry/declare",
+                "methods": ["GET", "POST"],
+            },
+            "info": {"uri": "/mgmt/shared/telemetry/info", "methods": ["GET"]},
+        }
+    },
 }
-ATC_SERVICE_OPTIONS = ["AS3", "Device"]  # , "Telemetry"]
+ATC_SERVICE_OPTIONS = ["AS3", "Device", "Telemetry"]
 
 
 def _build_as3_endpoint(
@@ -127,6 +127,14 @@ def _send(
     if atc_service == "Device" and atc_method == "POST":
         resp = client.post(url, json=atc_declaration)
 
+    # Telemetry
+    if atc_service == "Telemetry" and atc_method == "POST":
+        resp = client.post(url, json=atc_declaration)
+
+        message = resp.json()["message"]
+        if message != "success":
+            raise Exception("The declaration deployment failed.")
+
     # GET
     if atc_method == "GET":
         resp = client.get(url)
@@ -272,8 +280,8 @@ def atc(
         atc_service=atc_service,
     ).result
 
-    # If 'atc_method' is 'GET', return the declaration
-    if atc_method == "GET":
+    # If 'Telemetry' or 'GET', return the declaration
+    if atc_service == "Telemetry" or atc_method == "GET":
         return Result(host=task.host, result=atc_send_result)
 
     # Wait for task to complete

pyproject.toml

@@ -17,7 +17,7 @@ license = "Apache-2.0"
 name = "nornir_f5"
 readme = "README.md"
 repository = "https://github.com/erjac77/nornir_f5"
-version = "0.2.0"
+version = "0.3.0"
 
 [tool.poetry.urls]
 Issues = "https://github.com/erjac77/nornir_f5/issues"

tests/declarations/atc/telemetry/default_pull_consumer.json

@@ -0,0 +1,21 @@
+{
+  "class": "Telemetry",
+  "My_Poller": {
+    "class": "Telemetry_System_Poller",
+    "interval": 0
+  },
+  "My_System": {
+    "class": "Telemetry_System",
+    "enable": "true",
+    "systemPoller": [
+      "My_Poller"
+    ]
+  },
+  "My_Pull_Consumer": {
+    "class": "Telemetry_Pull_Consumer",
+    "type": "default",
+    "systemPoller": [
+      "My_Poller"
+    ]
+  }
+}

tests/responses/atc/telemetry/failed.json

@@ -0,0 +1,3 @@
+{
+  "message": "failed"
+}

tests/responses/atc/telemetry/success.json

@@ -0,0 +1,3 @@
+{
+  "message": "success"
+}

tests/responses/atc/telemetry/version_1.17.0.json

@@ -0,0 +1,3 @@
+{
+  "version": "1.17.0"
+}

tests/test_atc_tasks.py

@@ -343,3 +343,73 @@ def get_task_callback(request):
 
     # Assert result
     assert_result(result, expected)
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "resp", "expected"),
+    [
+        # POST TS declaration from file
+        (
+            {
+                "atc_declaration_file": f"{base_decl_dir}/atc/telemetry/default_pull_consumer.json",  # noqa B950
+                "atc_method": "POST",
+                "atc_service": "Telemetry",
+            },
+            {
+                "status_code": 200,
+                "data": f"{base_resp_dir}/atc/telemetry/success.json",
+            },
+            {
+                "result_file": f"{base_resp_dir}/atc/telemetry/success.json",
+            },
+        ),
+        # POST TS declaration, failed
+        (
+            {
+                "atc_declaration": {"class": "Telemetry"},
+                "atc_method": "POST",
+                "atc_service": "Telemetry",
+            },
+            {
+                "status_code": 200,
+                "data": f"{base_resp_dir}/atc/telemetry/failed.json",
+            },
+            {
+                "result": "The declaration deployment failed.",
+                "failed": True,
+            },
+        ),
+    ],
+)
+@responses.activate
+def test_ts_deploy(nornir, kwargs, resp, expected):
+    # Register mock responses
+    # GET TS info
+    responses.add(
+        responses.GET,
+        "https://bigip1.localhost:443/mgmt/shared/telemetry/info",
+        json=load_json(f"{base_resp_dir}/atc/telemetry/version_1.17.0.json"),
+        status=200,
+    )
+
+    if resp:
+        responses_data = load_json(resp["data"])
+        responses.add(
+            kwargs["atc_method"] if "atc_method" in kwargs else "GET",
+            "https://bigip1.localhost:443/mgmt/shared/telemetry/declare",
+            json=responses_data,
+            status=resp["status_code"],
+        )
+
+    # Run task
+    nornir = nornir.filter(name="bigip1.localhost")
+    result = nornir.run(
+        name="Deploy TS Declaration",
+        task=atc,
+        atc_delay=0,
+        atc_retries=3,
+        **kwargs,
+    )
+
+    # Assert result
+    assert_result(result, expected)

tests/test_readme.py

@@ -6,11 +6,7 @@
 from nornir_utils.plugins.functions import print_result
 
 import responses
-from nornir_f5.plugins.tasks import (
-    atc,
-    bigip_cm_config_sync,
-    bigip_cm_failover_status,
-)
+from nornir_f5.plugins.tasks import atc, bigip_cm_config_sync, bigip_cm_failover_status
 
 from .conftest import base_resp_dir, load_json