tests: memfault: Use chronicler API

The old eventlog API was deprecated.

Signed-off-by: Jorgen Kvalvaag <jorgen.kvalvaag@nordicsemi.no>

Author: jorgenmk

tests/on_target/tests/test_functional/test_memfault.py

@@ -5,7 +5,7 @@
 
 import os
 import requests
-import datetime
+from datetime import datetime, timezone, timedelta
 import time
 from utils.flash_tools import flash_device, reset_device
 from utils.logger import get_logger
@@ -21,36 +21,56 @@
 url = "https://api.memfault.com/api/v0"
 auth = ("", MEMFAULT_ORG_TOKEN)
 
-def get_event_log():
-    response = requests.get(f"{url}/organizations/{MEMFAULT_ORG}/projects/{MEMFAULT_PROJ}/event-log", auth=auth)
-    response.raise_for_status()
-    return response.json()
+def get_chronicler_events(
+        device_serial=None,
+        device_id=None,
+        event_types=("ReceivedDataRebootEvent", "ReceivedDataHeartbeatEvent", "ReceivedDataTraceEvent"),
+        levels=("INFO", "ERROR"),
+        start=None, end=None
+):
+    if not start:
+        start = datetime.now(timezone.utc) - timedelta(hours=1)
+    if not end:
+        end = datetime.now(timezone.utc)
 
-def get_latest_events(event_type, device_id):
-    data = get_event_log()
-    latest_events = [
-        x
-        for x in data["data"]
-        if x["device_serial"] == str(device_id) and x["type"] == event_type
-    ]
-    return latest_events
+    params = {
+        "device_id": device_id,
+        "device_serial": device_serial,
+        "event_types": event_types,
+        "levels": levels,
+        "start": datetime.isoformat(start),
+        "end": datetime.isoformat(end)
+    }
+    
+    r = requests.get(
+        f"{url}/organizations/{MEMFAULT_ORG}/projects/{MEMFAULT_PROJ}/chronicler-logs",
+        auth=auth,
+        params=params,
+        timeout=10
+    )
+    r.raise_for_status()
+    return r.json()["data"]
+    
+def get_latest_heartbeat_events(device_serial):
+    return get_chronicler_events(
+        device_serial=device_serial,
+        event_types=["ReceivedDataHeartbeatEvent"]
+    )
 
 def timestamp(event):
-    return datetime.datetime.strptime(
-        event["captured_date"], "%Y-%m-%dT%H:%M:%S.%f%z"
-    )
+    return datetime.fromisoformat(event["event_data"]["received_event"]["captured_date"])
 
 def wait_for_heartbeat(timestamp_old_heartbeat_evt):
     new_heartbeat_found = False
     start = time.time()
     while time.time() - start < MEMFAULT_TIMEOUT:
         logger.debug("Looking for latest heartbeat events")
         time.sleep(5)
-        new_heartbeat_events = get_latest_events("HEARTBEAT", IMEI)
+        new_heartbeat_events = get_latest_heartbeat_events(IMEI)
         if not new_heartbeat_events:
             continue
         try:
-            unexpected_reboot_count = new_heartbeat_events[0]["summary"]["event_info"]["metrics"]["MemfaultSdkMetric_UnexpectedRebootCount"]
+            unexpected_reboot_count = new_heartbeat_events[0]["event_data"]["received_event"]["event_info"]["metrics"]["MemfaultSdkMetric_UnexpectedRebootCount"]
         except (KeyError, ValueError):
             continue
         if unexpected_reboot_count < 1:
@@ -75,7 +95,7 @@ def test_memfault(t91x_board, hex_file):
     ]
 
     # Save timestamp of latest heartbeat event
-    heartbeat_events = get_latest_events("HEARTBEAT", IMEI)
+    heartbeat_events = get_latest_heartbeat_events(IMEI)
     timestamp_old_heartbeat_evt = timestamp(heartbeat_events[0]) if heartbeat_events else  None
 
     t91x_board.uart.flush()