Added get_metric_value helper function

Author: JoukoVirtanen

tests/test_path_rmdir.py

@@ -2,9 +2,9 @@
 import shutil
 
 import pytest
-import requests
 
 from event import Event, EventType, Process
+from utils import get_metric_value
 
 
 def get_inode_removed_count(fact_config):
@@ -17,18 +17,8 @@ def get_inode_removed_count(fact_config):
     Returns:
         The current value of host_scanner_scan{label="InodeRemoved"} metric.
     """
-    config, _ = fact_config
-    response = requests.get(f'http://{config["endpoint"]["address"]}/metrics')
-    assert response.status_code == 200
-
-    for line in response.text.split('\n'):
-        if 'host_scanner_scan' in line and 'label="InodeRemoved"' in line:
-            # Format: host_scanner_scan{label="InodeRemoved"} 42
-            parts = line.split()
-            if len(parts) >= 2:
-                return int(parts[-1])
-
-    return 0
+    value = get_metric_value(fact_config, "host_scanner_scan", {"label": "InodeRemoved"})
+    return int(value) if value is not None else 0
 
 
 @pytest.mark.parametrize("dirname", [

tests/utils.py

@@ -1,6 +1,8 @@
 import os
 import re
 
+import requests
+
 
 def join_path_with_filename(directory, filename):
     """
@@ -66,8 +68,40 @@ def rust_style_join(args):
     """
     Concatenate arguments after quoting them. Each argument is separated
     by a single space.
-    
+
     Args:
         args: The string to quote
     """
     return ' '.join(rust_style_quote(arg) for arg in args)
+
+
+def get_metric_value(fact_config, metric_name, labels=None):
+    """
+    Query Prometheus metrics endpoint to get the value of a metric.
+
+    Args:
+        fact_config: The fact configuration tuple (config dict, config file path).
+        metric_name: Name of the metric to query (e.g., "host_scanner_scan").
+        labels: Optional dict of label filters (e.g., {"label": "InodeRemoved"}).
+
+    Returns:
+        The metric value as a string if found, None otherwise.
+    """
+    config, _ = fact_config
+    response = requests.get(f'http://{config["endpoint"]["address"]}/metrics')
+    assert response.status_code == 200
+
+    labels = labels or {}
+
+    for line in response.text.split('\n'):
+        if metric_name not in line:
+            continue
+
+        # Check if all label filters match
+        if all(f'{k}="{v}"' in line for k, v in labels.items()):
+            # Format: metric_name{label="value"} 42
+            parts = line.split()
+            if len(parts) >= 2:
+                return parts[-1]
+
+    return None