Automated nbdkit count filter reports number of bytes read, write, zeroed, trimmed in debug log

v2v/tests/cfg/nbdkit/nbdkit.cfg

@@ -81,6 +81,9 @@
             option_tar_limit = '0 100K 100G'
             option_tar_entry = '0 latest-rhel9.img fake-entry'
             checkpoint = 'test_tar_filter'
+          - count:
+            version_required = "[nbdkit-1.46,)"
+            checkpoint = 'test_count_filter'
       - plugin:
         variants:
           - file:

v2v/tests/src/nbdkit/nbdkit.py

@@ -809,6 +809,59 @@ def test_nbdkit_instance_name():
                         # Force kill if it's being stubborn
                         p.kill()
 
+    def test_count_filter():
+        # p is our SubProcess object
+        p = None
+        try:
+            LOG.info("Starting nbdkit")
+            # Define the command.
+            cmd = "nbdkit -f -v memory 10M --filter=count"
+
+            # Instantiate the SubProcess object
+            # Note: We pass the LOG so it knows where to send output
+            p = process.SubProcess(cmd, shell=True, logger=LOG)
+
+            # Start it in the background
+            p.start()
+
+            # Check if the process actually started (has a PID)
+            if not p.get_pid():
+                test.fail("nbdkit failed to start!")
+
+            # Allow nbdkit instance time to start listening
+            import time
+            time.sleep(1)
+
+            # Trigger activity with qemu-io (Synchronous run is fine here)
+            LOG.info("Writing data to nbdkit")
+            qemu_cmd = "qemu-io -f raw nbd://localhost -c 'write 0 1M' -c 'read 0 1M' -c 'write -z 1M 1M' -c 'discard 0 1M'"
+            result = process.run(qemu_cmd, shell=True)
+            if result.exit_status != 0:
+                test.fail(f"qemu-io failed with status {result.exit_status}")
+
+            # VALIDATION: Check the captured output
+            # SubProcess captures stdout and stderr separately as bytes
+            log_content = (p.get_stdout() + p.get_stderr()).decode('utf-8', errors='replace')
+            LOG.info(f"Captured output: {log_content}")
+
+            expected_patterns = [
+                "count: pwrite count=1048576",
+                "count: pread count=1048576",
+                "count: zero count=1048576",
+                "count: trim count=1048576"
+            ]
+            for pattern in expected_patterns:
+                if pattern not in log_content:
+                    # We show the log content to help debug why it failed
+                    test.fail(f"'{pattern}' was not found in logs!")
+                LOG.info(f"Pattern '{pattern}' found.")
+        finally:
+            # Cleanup using the class methods
+            if p and p.poll() is None:
+                LOG.info("Cleanup: Stopping nbdkit process")
+                # stop() is better than terminate() here as it handles the wait() logic
+                p.stop(timeout=5)
+
     if version_required and not multiple_versions_compare(
             version_required):
         test.cancel("Testing requires version: %s" % version_required)
@@ -889,5 +942,7 @@ def test_nbdkit_instance_name():
         check_blocksize_constraints()
     elif checkpoint == 'test_nbdkit_instance_name':
         test_nbdkit_instance_name()
+    elif checkpoint == 'test_count_filter':
+        test_count_filter()
     else:
         test.error('Not found testcase: %s' % checkpoint)