chore: cleanup for mirrored workflows

Author: b-long

conftest.py

@@ -5,11 +5,18 @@
 loaded by pytest when running tests.
 """
 
+from pathlib import Path
+
 import pytest
 
 from tests.server_logs import log_server_logs_on_failure
 
 
+@pytest.fixture(scope="session")
+def project_root(request) -> Path:
+    return request.config.rootpath  # Project root
+
+
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
 def pytest_runtest_makereport(item, call):
     """

tests/integration/conftest.py

@@ -9,7 +9,7 @@
 
 import pytest
 
-from tests.support_otdfctl_args import generate_tdf_files_for_target_mode
+from tests.support_otdfctl_args import otdfctl_generate_tdf_files_for_target_mode
 
 logger = logging.getLogger(__name__)
 
@@ -45,7 +45,7 @@ def sample_input_files(test_data_dir):
 @pytest.fixture(scope="session")
 def tdf_v4_2_2_files(temp_credentials_file, test_data_dir, sample_input_files):
     """Generate TDF files with target mode v4.2.2."""
-    tdf_files = generate_tdf_files_for_target_mode(
+    tdf_files = otdfctl_generate_tdf_files_for_target_mode(
         "v4.2.2", temp_credentials_file, test_data_dir, sample_input_files
     )
     yield tdf_files
@@ -54,7 +54,7 @@ def tdf_v4_2_2_files(temp_credentials_file, test_data_dir, sample_input_files):
 @pytest.fixture(scope="session")
 def tdf_v4_3_1_files(temp_credentials_file, test_data_dir, sample_input_files):
     """Generate TDF files with target mode v4.3.1."""
-    tdf_files = generate_tdf_files_for_target_mode(
+    tdf_files = otdfctl_generate_tdf_files_for_target_mode(
         "v4.3.1", temp_credentials_file, test_data_dir, sample_input_files
     )
     yield tdf_files
@@ -67,3 +67,8 @@ def all_target_mode_tdf_files(tdf_v4_2_2_files, tdf_v4_3_1_files):
         "v4.2.2": tdf_v4_2_2_files,
         "v4.3.1": tdf_v4_3_1_files,
     }
+
+
+@pytest.fixture(scope="session")
+def known_target_modes():
+    return ["v4.2.2", "v4.3.1"]

tests/integration/otdfctl_only/test_otdfctl_generated_fixtures.py

@@ -1,7 +1,7 @@
-from pathlib import Path
-
 import pytest
 
+from tests.support_common import validate_tdf3_file
+
 
 @pytest.mark.integration
 def test_test_data_directory_structure(tdf_v4_2_2_files, tdf_v4_3_1_files):
@@ -80,15 +80,15 @@ def test_sample_file_contents(sample_input_files):
 
 
 @pytest.mark.integration
-def test_target_mode_fixtures_exist(all_target_mode_tdf_files):
+def test_target_mode_fixtures_exist(all_target_mode_tdf_files, known_target_modes):
     """Test that target mode fixtures generate TDF files correctly."""
     # Check that we have both versions
     assert "v4.2.2" in all_target_mode_tdf_files
     assert "v4.3.1" in all_target_mode_tdf_files
 
     # Check each version has the expected file types
-    for version in ["v4.2.2", "v4.3.1"]:
-        tdf_files = all_target_mode_tdf_files[version]
+    for target_mode in known_target_modes:
+        tdf_files = all_target_mode_tdf_files[target_mode]
 
         # Check all expected file types exist
         expected_types = [
@@ -97,18 +97,14 @@ def test_target_mode_fixtures_exist(all_target_mode_tdf_files):
             "with_attributes",
         ]  # Consider 'empty' as well
         for file_type in expected_types:
-            assert file_type in tdf_files, f"Missing {file_type} TDF for {version}"
+            assert file_type in tdf_files, f"Missing {file_type} TDF for {target_mode}"
 
             # Check the TDF file exists and is not empty
             tdf_path = tdf_files[file_type]
-            assert isinstance(tdf_path, Path)
-            assert tdf_path.exists(), f"TDF file does not exist: {tdf_path}"
-            assert tdf_path.stat().st_size > 0, f"TDF file is empty: {tdf_path}"
-
-            # Check it's a valid ZIP file (TDF format)
-            with open(tdf_path, "rb") as f:
-                header = f.read(4)
-            assert header == b"PK\x03\x04", f"TDF file is not a valid ZIP: {tdf_path}"
+            validate_tdf3_file(
+                tdf_path,
+                f"otdfctl generated using target-mode {target_mode} {file_type}",
+            )
 
 
 @pytest.mark.integration

tests/integration/otdfctl_to_python/test_cli_comparison.py

@@ -2,21 +2,27 @@
 Test CLI functionality
 """
 
-import subprocess
 import tempfile
 from pathlib import Path
 
 import pytest
 
-from tests.support_cli_args import build_cli_decrypt_command
+from tests.support_cli_args import run_cli_decrypt
+from tests.support_common import (
+    handle_subprocess_error,
+    validate_plaintext_file_created,
+    validate_tdf3_file,
+)
 from tests.support_otdfctl_args import (
-    build_otdfctl_decrypt_command,
-    build_otdfctl_encrypt_command,
+    run_otdfctl_decrypt_command,
+    run_otdfctl_encrypt_command,
 )
 
 
 @pytest.mark.integration
-def test_otdfctl_encrypt_python_decrypt(collect_server_logs, temp_credentials_file):
+def test_otdfctl_encrypt_python_decrypt(
+    collect_server_logs, temp_credentials_file, project_root
+):
     """Integration test that uses otdfctl for encryption and the Python CLI for decryption"""
 
     # Create temporary directory for work
@@ -37,99 +43,57 @@ def test_otdfctl_encrypt_python_decrypt(collect_server_logs, temp_credentials_fi
         cli_decrypt_output = temp_path / "decrypted-by-cli.txt"
 
         # Run otdfctl encrypt first to create a TDF file
-        otdfctl_encrypt_cmd = build_otdfctl_encrypt_command(
+        otdfctl_encrypt_result = run_otdfctl_encrypt_command(
             creds_file=temp_credentials_file,
             input_file=input_file,
             output_file=otdfctl_tdf_output,
             mime_type="text/plain",
+            cwd=temp_path,
         )
 
-        otdfctl_encrypt_result = subprocess.run(
-            otdfctl_encrypt_cmd, capture_output=True, text=True, cwd=temp_path
+        # Fail fast on errors
+        handle_subprocess_error(
+            result=otdfctl_encrypt_result,
+            collect_server_logs=collect_server_logs,
+            scenario_name="otdfctl encrypt",
         )
 
-        # If otdfctl encrypt fails, skip the test (might be server issues)
-        if otdfctl_encrypt_result.returncode != 0:
-            raise Exception(f"otdfctl encrypt failed: {otdfctl_encrypt_result.stderr}")
-
         # 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"
 
         # Now run otdfctl decrypt (this is the reference implementation)
-        otdfctl_decrypt_cmd = build_otdfctl_decrypt_command(
+        otdfctl_decrypt_result = run_otdfctl_decrypt_command(
             temp_credentials_file,
             otdfctl_tdf_output,
             otdfctl_decrypt_output,
+            cwd=temp_path,
         )
 
-        otdfctl_decrypt_result = subprocess.run(
-            otdfctl_decrypt_cmd, capture_output=True, text=True, cwd=temp_path
+        # Fail fast on errors
+        handle_subprocess_error(
+            result=otdfctl_decrypt_result,
+            collect_server_logs=collect_server_logs,
+            scenario_name="otdfctl decrypt",
         )
 
-        # Check that otdfctl decrypt succeeded
-        if otdfctl_decrypt_result.returncode != 0:
-            # Collect server logs for debugging
-            logs = collect_server_logs()
-            print(f"Server logs when otdfctl decrypt failed:\n{logs}")
-
-            # Check if this is a server connectivity issue
-            if (
-                "401 Unauthorized" in otdfctl_decrypt_result.stderr
-                or "token endpoint discovery" in otdfctl_decrypt_result.stderr
-                or "Issuer endpoint must be configured" in otdfctl_decrypt_result.stderr
-            ):
-                pytest.skip(
-                    f"Server connectivity or authentication issue: {otdfctl_decrypt_result.stderr}"
-                )
-            else:
-                assert otdfctl_decrypt_result.returncode == 0, (
-                    f"otdfctl decrypt failed: {otdfctl_decrypt_result.stderr}"
-                )
-
         # Run our Python CLI decrypt on the same TDF
-        cli_decrypt_cmd = build_cli_decrypt_command(
+        cli_decrypt_result = run_cli_decrypt(
             creds_file=temp_credentials_file,
             input_file=otdfctl_tdf_output,
             output_file=cli_decrypt_output,
+            cwd=project_root,
         )
 
-        cli_decrypt_result = subprocess.run(
-            cli_decrypt_cmd,
-            capture_output=True,
-            text=True,
-            cwd=Path(__file__).parent.parent,
+        # Fail fast on errors
+        handle_subprocess_error(
+            result=cli_decrypt_result,
+            collect_server_logs=collect_server_logs,
+            scenario_name="Python CLI decrypt",
         )
 
-        # Check that our CLI succeeded
-        if cli_decrypt_result.returncode != 0:
-            # Collect server logs for debugging
-            logs = collect_server_logs()
-            print(f"Server logs when Python CLI decrypt failed:\n{logs}")
-
-            # Check if this is a server connectivity issue
-            if (
-                "401 Unauthorized" in cli_decrypt_result.stderr
-                or "token endpoint discovery" in cli_decrypt_result.stderr
-                or "Issuer endpoint must be configured" in cli_decrypt_result.stderr
-            ):
-                pytest.skip(
-                    f"Server connectivity or authentication issue: {cli_decrypt_result.stderr}"
-                )
-            else:
-                assert cli_decrypt_result.returncode == 0, (
-                    f"Python CLI decrypt failed: {cli_decrypt_result.stderr}"
-                )
-
-        # 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:
@@ -179,82 +143,40 @@ def test_otdfctl_encrypt_otdfctl_decrypt(collect_server_logs, temp_credentials_f
         otdfctl_decrypt_output = temp_path / "otdfctl-roundtrip-decrypted.txt"
 
         # Run otdfctl encrypt
-        otdfctl_encrypt_cmd = build_otdfctl_encrypt_command(
+        otdfctl_encrypt_result = run_otdfctl_encrypt_command(
             creds_file=temp_credentials_file,
             input_file=input_file,
             output_file=otdfctl_tdf_output,
             mime_type="text/plain",
+            cwd=temp_path,
         )
 
-        otdfctl_encrypt_result = subprocess.run(
-            otdfctl_encrypt_cmd, capture_output=True, text=True, cwd=temp_path
+        # Fail fast on errors
+        handle_subprocess_error(
+            result=otdfctl_encrypt_result,
+            collect_server_logs=collect_server_logs,
+            scenario_name="otdfctl encrypt",
         )
 
-        # If otdfctl encrypt fails, skip the test (might be server issues)
-        if otdfctl_encrypt_result.returncode != 0:
-            # Collect server logs for debugging
-            logs = collect_server_logs()
-            print(f"Server logs when otdfctl encrypt failed:\n{logs}")
-
-            # Check if this is a server connectivity issue
-            if (
-                "401 Unauthorized" in otdfctl_encrypt_result.stderr
-                or "token endpoint discovery" in otdfctl_encrypt_result.stderr
-                or "Issuer endpoint must be configured" in otdfctl_encrypt_result.stderr
-            ):
-                pytest.skip(
-                    f"Server connectivity or authentication issue: {otdfctl_encrypt_result.stderr}"
-                )
-            else:
-                assert otdfctl_encrypt_result.returncode == 0, (
-                    f"otdfctl encrypt failed: {otdfctl_encrypt_result.stderr}"
-                )
-
         # 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_cmd = build_otdfctl_decrypt_command(
+        otdfctl_decrypt_result = run_otdfctl_decrypt_command(
             temp_credentials_file,
             otdfctl_tdf_output,
             otdfctl_decrypt_output,
+            cwd=temp_path,
         )
 
-        otdfctl_decrypt_result = subprocess.run(
-            otdfctl_decrypt_cmd, capture_output=True, text=True, cwd=temp_path
+        # Fail fast on errors
+        handle_subprocess_error(
+            result=otdfctl_decrypt_result,
+            collect_server_logs=collect_server_logs,
+            scenario_name="otdfctl decrypt",
         )
 
-        # Check that otdfctl decrypt succeeded
-        if otdfctl_decrypt_result.returncode != 0:
-            # Collect server logs for debugging
-            logs = collect_server_logs()
-            print(f"Server logs when otdfctl decrypt failed:\n{logs}")
-
-            # Check if this is a server connectivity issue
-            if (
-                "401 Unauthorized" in otdfctl_decrypt_result.stderr
-                or "token endpoint discovery" in otdfctl_decrypt_result.stderr
-                or "Issuer endpoint must be configured" in otdfctl_decrypt_result.stderr
-            ):
-                pytest.skip(
-                    f"Server connectivity or authentication issue: {otdfctl_decrypt_result.stderr}"
-                )
-            else:
-                assert otdfctl_decrypt_result.returncode == 0, (
-                    f"otdfctl decrypt failed: {otdfctl_decrypt_result.stderr}"
-                )
-
-        # 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: