Automated nbdkit data plugin supports base64 in format strings

Signed-off-by: Ganesh Hubale <ghubale@redhat.com>

Author: ganeshhubale

v2v/tests/cfg/nbdkit/nbdkit.cfg

@@ -165,6 +165,16 @@
               - constraints:
                 checkpoint = "blocksize_constraints"
                 expected_fields = "block_size_preferred block_size_maximum block_size_minimum"
+          - data:
+            only source_none..dest_none
+            version_required = "[nbdkit-1.46,)"
+            checkpoint = "test_data_plugin_supports_base64_in_format_string"
+            variants:
+              - positive_test:
+                test_state= 'positive'
+              - negative_test:
+                test_state= 'negative'
+                expected_err_msg= 'Base64 decoding error.'
       - run:
         checkpoint = 'has_run_againt_vddk'
         vpx_passwd_file = "/tmp/v2v_vpx_passwd"

v2v/tests/src/nbdkit/nbdkit.py

@@ -809,6 +809,33 @@ def test_nbdkit_instance_name():
                         # Force kill if it's being stubborn
                         p.kill()
 
+    def test_data_plugin_supports_base64_in_format_string():
+        import base64
+        test_state = params_get(params, "test_state")
+        if test_state == "positive":
+            # Prepare the data
+            original_text = b"nbdkit-test123"
+            # Encode bytes to a base64 string
+            b64_input = base64.b64encode(original_text).decode()
+            cmd_str = f"nbdkit -U - data \'base64:{b64_input}\' --run 'nbdcopy $uri -'"
+            # Execute command
+            result = process.run(cmd_str, shell=True, ignore_status=True)
+            # Result.stdout contains the raw binary output from nbdcopy
+            cmd_output = result.stdout
+            if original_text not in cmd_output:
+                test.fail(f"Base64 decoding failed. Expected {original_text!r}, got {cmd_output!r}")
+        elif test_state == "negative":
+            # Negative scenario - Check with invalid value
+            invalid_b64 = "base64:SGVsbG8@@@"
+            cmd_str = f"nbdkit -U - data \'{invalid_b64}\' --run 'nbdcopy $uri -'"
+            # Execute command
+            result = process.run(cmd_str, shell=True, ignore_status=True)
+            # Verify the error message
+            cmd_output = result.stderr_text
+            err_msg = params_get(params, "expected_err_msg")
+            if err_msg not in cmd_output:
+                test.fail(f"Error message - \'{err_msg}\' not appeared.")
+
     if version_required and not multiple_versions_compare(
             version_required):
         test.cancel("Testing requires version: %s" % version_required)
@@ -889,5 +916,7 @@ def test_nbdkit_instance_name():
         check_blocksize_constraints()
     elif checkpoint == 'test_nbdkit_instance_name':
         test_nbdkit_instance_name()
+    elif checkpoint == 'test_data_plugin_supports_base64_in_format_string':
+        test_data_plugin_supports_base64_in_format_string()
     else:
         test.error('Not found testcase: %s' % checkpoint)