test: add telegraf container test

base/images/tests/cases/runtime/container-base/test_telegraf/Dockerfile

@@ -0,0 +1,5 @@
+ARG BASE_IMAGE
+FROM ${BASE_IMAGE}
+
+RUN dnf install -y telegraf && dnf clean all
+COPY configs/telegraf.conf /etc/telegraf/telegraf.conf

base/images/tests/cases/runtime/container-base/test_telegraf/configs/file_output.conf

@@ -0,0 +1,10 @@
+[agent]
+  interval = "1s"
+  flush_interval = "1s"
+  logtarget = "stderr"
+
+[[inputs.mem]]
+
+[[outputs.file]]
+  files = ["/tmp/telegraf-metrics.out"]
+  data_format = "influx"

base/images/tests/cases/runtime/container-base/test_telegraf/configs/telegraf.conf

@@ -0,0 +1,21 @@
+[agent]
+  interval = "10s"
+  round_interval = true
+  metric_batch_size = 1000
+  metric_buffer_limit = 10000
+  collection_jitter = "0s"
+  flush_interval = "10s"
+  flush_jitter = "0s"
+  precision = ""
+  logtarget = "stderr"
+  logfile = ""
+  hostname = ""
+  omit_hostname = false
+
+[[inputs.cpu]]
+  percpu = true
+  totalcpu = true
+  collect_cpu_time = false
+  report_active = false
+
+[[inputs.mem]]

base/images/tests/cases/runtime/container-base/test_telegraf/test_telegraf.py

@@ -0,0 +1,54 @@
+# SPDX-License-Identifier: MIT
+"""Validate Telegraf runtime behavior on container-base images.
+
+Use checked-in config, run in test mode, and verify mem metrics output.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+
+TELEGRAF_CONFIG = "/etc/telegraf/telegraf.conf"
+
+
+@pytest.mark.dockerfile()
+def test_telegraf_emits_mem_metrics(container_exec_shell) -> None:
+    """telegraf --test must emit mem plugin measurement output."""
+    result = container_exec_shell(f"telegraf --config {TELEGRAF_CONFIG} --test")
+    assert result.exit_code == 0, f"telegraf --test failed: {result.output}"
+    assert "mem,host" in result.output
+
+
+@pytest.mark.dockerfile()
+def test_telegraf_reports_version_and_plugin_usage(container_exec_shell) -> None:
+    """telegraf binary should report version and cpu plugin usage details."""
+    result = container_exec_shell("telegraf --version && telegraf --usage cpu")
+    assert result.exit_code == 0, f"telegraf version/usage check failed: {result.output}"
+    assert "Telegraf" in result.output
+    assert "cpu" in result.output.lower()
+
+
+@pytest.mark.dockerfile()
+def test_telegraf_file_output_plugin_writes_metrics(
+    container_exec_shell, write_file_in_container
+) -> None:
+    """telegraf should be able to flush metrics to file output."""
+    config_body = (
+        Path(__file__).parent / "configs" / "file_output.conf"
+    ).read_text(encoding="utf-8")
+    result = write_file_in_container("/tmp/telegraf-file-output.conf", config_body)
+    assert result.exit_code == 0, f"failed writing file-output config: {result.output}"
+
+    result = container_exec_shell(
+        "set -o pipefail; telegraf --config /tmp/telegraf-file-output.conf --once 2>&1 | tee /tmp/telegraf-file.log"
+    )
+    assert result.exit_code == 0, f"telegraf file output run failed: {result.output}"
+
+    result = container_exec_shell(
+        'test -s /tmp/telegraf-metrics.out && grep -q "mem,host" /tmp/telegraf-metrics.out'
+    )
+    assert result.exit_code == 0, f"telegraf file output validation failed: {result.output}"
+

base/images/tests/conftest.py

@@ -394,6 +394,32 @@ def _exec_shell(command: str, *, shell: str = "bash"):
     return _exec_shell
 
 
+@pytest.fixture
+def write_file_in_container(container_exec_shell):
+    """Callable to write file content into the running test container.
+
+    Preserves leading whitespace and content exactly as provided, while
+    normalizing trailing newlines so the written file ends with exactly
+    one trailing newline.
+
+    Usage::
+
+        def test_example(write_file_in_container):
+            result = write_file_in_container("/tmp/example.conf", "key=value")
+            assert result.exit_code == 0
+    """
+
+    def _write(path: str, content: str):
+        normalized_content = content.rstrip("\n") + "\n"
+        write_cmd = (
+            f"printf %s {shlex.quote(normalized_content)} > "
+            f"{shlex.quote(path)}"
+        )
+        return container_exec_shell(write_cmd)
+
+    return _write
+
+
 @pytest.fixture
 def wait_for_http(container_exec_shell):
     """Callable that polls an in-container HTTP endpoint until it responds.