chore: cleanup for mirrored workflows

Author: b-long

tests/integration/otdfctl_to_python/test_cli_comparison.py

@@ -8,7 +8,11 @@
 import pytest
 
 from tests.support_cli_args import run_cli_decrypt
-from tests.support_common import handle_subprocess_error
+from tests.support_common import (
+    handle_subprocess_error,
+    validate_plaintext_file_created,
+    validate_tdf3_file,
+)
 from tests.support_otdfctl_args import (
     run_otdfctl_decrypt_command,
     run_otdfctl_encrypt_command,
@@ -88,15 +92,8 @@ def test_otdfctl_encrypt_python_decrypt(
             scenario_name="Python CLI decrypt",
         )
 
-        # Verify both decrypted files were created
-        assert otdfctl_decrypt_output.exists(), "otdfctl did not create decrypted file"
-        assert otdfctl_decrypt_output.stat().st_size > 0, (
-            "otdfctl created empty decrypted file"
-        )
-        assert cli_decrypt_output.exists(), "Python CLI did not create decrypted file"
-        assert cli_decrypt_output.stat().st_size > 0, (
-            "Python CLI created empty decrypted file"
-        )
+        validate_plaintext_file_created(path=otdfctl_decrypt_output, scenario="otdfctl")
+        validate_plaintext_file_created(path=cli_decrypt_output, scenario="Python CLI")
 
         # Verify both tools produce the same decrypted content
         with open(otdfctl_decrypt_output) as f:
@@ -162,13 +159,7 @@ def test_otdfctl_encrypt_otdfctl_decrypt(collect_server_logs, temp_credentials_f
         )
 
         # Verify the TDF file was created
-        assert otdfctl_tdf_output.exists(), "otdfctl did not create TDF file"
-        assert otdfctl_tdf_output.stat().st_size > 0, "otdfctl created empty TDF file"
-
-        # Verify TDF file has correct ZIP signature
-        with open(otdfctl_tdf_output, "rb") as f:
-            tdf_header = f.read(4)
-        assert tdf_header == b"PK\x03\x04", "otdfctl output is not a valid ZIP file"
+        validate_tdf3_file(tdf_path=otdfctl_tdf_output, tool_name="otdfctl")
 
         # Run otdfctl decrypt
         otdfctl_decrypt_result = run_otdfctl_decrypt_command(
@@ -185,11 +176,7 @@ def test_otdfctl_encrypt_otdfctl_decrypt(collect_server_logs, temp_credentials_f
             scenario_name="otdfctl decrypt",
         )
 
-        # Verify the decrypted file was created
-        assert otdfctl_decrypt_output.exists(), "otdfctl did not create decrypted file"
-        assert otdfctl_decrypt_output.stat().st_size > 0, (
-            "otdfctl created empty decrypted file"
-        )
+        validate_plaintext_file_created(path=otdfctl_decrypt_output, scenario="otdfctl")
 
         # Verify the decrypted content matches the original
         with open(otdfctl_decrypt_output) as f:

tests/integration/test_cli_integration.py

@@ -11,6 +11,7 @@
 from tests.support_common import (
     compare_tdf3_file_size,
     handle_subprocess_error,
+    validate_plaintext_file_created,
     validate_tdf3_file,
 )
 from tests.support_otdfctl_args import (
@@ -78,10 +79,8 @@ def test_cli_decrypt_otdfctl_tdf(
             scenario_name="Python CLI decrypt",
         )
 
-        # Verify the decrypted file was created
-        assert cli_decrypt_output.exists(), "Python CLI did not create decrypted file"
-        assert cli_decrypt_output.stat().st_size > 0, (
-            "Python CLI created empty decrypted file"
+        validate_plaintext_file_created(
+            path=cli_decrypt_output, scenario="Python decrypt"
         )
 
         # Verify the content matches the original
@@ -168,15 +167,8 @@ def test_otdfctl_decrypt_comparison(
             scenario_name="Python CLI decrypt",
         )
 
-        # Verify both decrypted files were created
-        assert otdfctl_decrypt_output.exists(), "otdfctl did not create decrypted file"
-        assert otdfctl_decrypt_output.stat().st_size > 0, (
-            "otdfctl created empty decrypted file"
-        )
-        assert cli_decrypt_output.exists(), "Python CLI did not create decrypted file"
-        assert cli_decrypt_output.stat().st_size > 0, (
-            "Python CLI created empty decrypted file"
-        )
+        validate_plaintext_file_created(path=otdfctl_decrypt_output, scenario="otdfctl")
+        validate_plaintext_file_created(path=cli_decrypt_output, scenario="Python CL")
 
         # Verify both tools produce the same decrypted content
         with open(otdfctl_decrypt_output) as f:
@@ -261,11 +253,7 @@ def test_otdfctl_encrypt_decrypt_roundtrip(collect_server_logs, temp_credentials
             scenario_name="otdfctl decrypt",
         )
 
-        # Verify the decrypted file was created
-        assert otdfctl_decrypt_output.exists(), "otdfctl did not create decrypted file"
-        assert otdfctl_decrypt_output.stat().st_size > 0, (
-            "otdfctl created empty decrypted file"
-        )
+        validate_plaintext_file_created(path=otdfctl_decrypt_output, scenario="otdfctl")
 
         # Verify the decrypted content matches the original
         with open(otdfctl_decrypt_output) as f:

tests/integration/test_cli_tdf_validation.py

@@ -16,6 +16,7 @@
 )
 from tests.support_common import (
     handle_subprocess_error,
+    validate_plaintext_file_created,
     validate_tdf3_file,
 )
 from tests.support_otdfctl_args import (
@@ -249,9 +250,7 @@ def _run_otdfctl_decrypt(
         otdfctl_decrypt_result, collect_server_logs, "otdfctl decrypt"
     )
 
-    # Verify the decrypted file was created
-    assert decrypt_output.exists(), "otdfctl did not create decrypted file"
-    assert decrypt_output.stat().st_size > 0, "otdfctl created empty decrypted file"
+    validate_plaintext_file_created(path=decrypt_output, scenario="otdfctl")
 
     # Verify the decrypted content matches expected
     with open(decrypt_output) as f:
@@ -285,9 +284,7 @@ def _run_python_cli_decrypt(
         python_decrypt_result, collect_server_logs, "Python CLI decrypt"
     )
 
-    # Verify the decrypted file was created
-    assert decrypt_output.exists(), "Python CLI did not create decrypted file"
-    assert decrypt_output.stat().st_size > 0, "Python CLI created empty decrypted file"
+    validate_plaintext_file_created(path=decrypt_output, scenario="Python CLI")
 
     # Verify the decrypted content matches expected
     with open(decrypt_output) as f:

tests/support_common.py

@@ -60,6 +60,12 @@ def validate_tdf3_file(tdf_path: Path, tool_name: str) -> None:
     assert tdf_header == b"PK\x03\x04", f"{tool_name} output is not a valid ZIP file"
 
 
+def validate_plaintext_file_created(path: Path, scenario: str) -> None:
+    """Validate that a non-empty file was created"""
+    assert path.exists(), f"{scenario=} did not create decrypted file"
+    assert path.stat().st_size > 0, f"{scenario=} created empty decrypted file"
+
+
 def compare_tdf3_file_size(otdfctl_tdf_path: Path, py_cli_tdf_path: Path) -> None:
     """Compare the file sizes of two TDF files (tdf_type="tdf3"), assert within 30% of each other."""
     size_otdfctl_tdf = otdfctl_tdf_path.stat().st_size