fix: cleanup code & disable nanotdf tests

.pre-commit-config.yaml

@@ -33,7 +33,7 @@ repos:
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
     # Ruff version.
-    rev: v0.15.20
+    rev: v0.15.21
     hooks:
       # Run the linter.
       - id: ruff-check

conftest.py

@@ -18,7 +18,7 @@ def project_root(request) -> Path:
 
 
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
-def pytest_runtest_makereport(item, call):
+def pytest_runtest_makereport(item):
     """Collect server logs when test fails after each test phase.
 
     This hook automatically collects server logs when a test fails.

packages/otdf-python-proto/tests/test_generate_connect_proto.py

@@ -121,7 +121,7 @@ def _run_with_tag(self, tmp_path: Path, tag: str):
 
         captured: list[list] = []
 
-        def fake_run(cmd, **kwargs):
+        def fake_run(cmd, **_kwargs):
             captured.append(cmd)
             if cmd[0] == "git":
                 # Simulate a successful clone by creating the service dir with one proto.
@@ -155,7 +155,7 @@ def test_default_tag_is_used_when_no_tag_is_provided(self, tmp_path):
 
         captured: list[list] = []
 
-        def fake_run(cmd, **kwargs):
+        def fake_run(cmd, **_kwargs):
             captured.append(cmd)
             if cmd[0] == "git":
                 temp_repo = tmp_path / "temp_platform_repo"
@@ -248,7 +248,7 @@ def test_function_returns_false_on_subprocess_error(self, tmp_path):
         assert result is False
 
     def test_function_returns_false_when_no_protos_copied(self, tmp_path):
-        def fake_run(cmd, **kwargs):
+        def fake_run(cmd, **_kwargs):
             if cmd[0] == "git":
                 # Clone succeeds but leaves an empty service dir (no .proto files)
                 service_dir = tmp_path / "temp_platform_repo" / "service"
@@ -265,7 +265,7 @@ def test_temp_dir_cleaned_up_on_success(self, tmp_path):
         """finally block must clean up temp_platform_repo regardless of outcome."""
         temp_repo = tmp_path / "temp_platform_repo"
 
-        def fake_run(cmd, **kwargs):
+        def fake_run(cmd, **_kwargs):
             if cmd[0] == "git":
                 service_dir = temp_repo / "service" / "kas"
                 service_dir.mkdir(parents=True)
@@ -280,7 +280,7 @@ def fake_run(cmd, **kwargs):
     def test_temp_dir_cleaned_up_on_failure(self, tmp_path):
         temp_repo = tmp_path / "temp_platform_repo"
 
-        def fake_run(cmd, **kwargs):
+        def fake_run(cmd, **_kwargs):
             if cmd[0] == "git":
                 temp_repo.mkdir(exist_ok=True)
                 raise gen.subprocess.CalledProcessError(1, "git")

packages/otdf-python/hatch_build.py

@@ -8,7 +8,7 @@
 class CustomBuildHook(BuildHookInterface):
     """Dynamically resolves proto sources for wheel builds from source tree or sdist."""
 
-    def initialize(self, version, build_data):
+    def initialize(self, _version, build_data):
         """Set force_include for otdf_python_proto based on build context."""
         if self.target_name != "wheel":
             return

packages/otdf-python/src/otdf_python/aesgcm.py

@@ -44,7 +44,9 @@ def encrypt(
     def encrypt_with_iv(
         self,
         iv: bytes,
-        auth_tag_len: int,
+        # Kept for signature parity with the Java SDK; the cryptography lib
+        # derives the tag length itself.
+        auth_tag_len: int,  # noqa: ARG002
         plaintext: bytes,
         offset: int = 0,
         length: int | None = None,
@@ -58,6 +60,11 @@ def decrypt(self, encrypted: "AesGcm.Encrypted") -> bytes:
         return self.aesgcm.decrypt(encrypted.iv, encrypted.ciphertext, None)
 
     def decrypt_with_iv(
-        self, iv: bytes, auth_tag_len: int, cipher_data: bytes
+        self,
+        iv: bytes,
+        # Kept for signature parity with the Java SDK; the cryptography lib
+        # derives the tag length itself.
+        auth_tag_len: int,  # noqa: ARG002
+        cipher_data: bytes,
     ) -> bytes:
         return self.aesgcm.decrypt(iv, cipher_data, None)

packages/otdf-python/src/otdf_python/cli.py

@@ -290,7 +290,7 @@ def create_tdf_config(sdk: SDK, args) -> TDFConfig:
     return config
 
 
-def create_nano_tdf_config(sdk: SDK, args) -> NanoTDFConfig:
+def create_nano_tdf_config(_sdk: SDK, args) -> NanoTDFConfig:
     """Create NanoTDF configuration from CLI arguments."""
     attributes = (
         parse_attributes(args.attributes)

packages/otdf-python/src/otdf_python/collection_store.py

@@ -29,7 +29,7 @@ class NoOpCollectionStore(CollectionStore):
     def store(self, header, key: CollectionKey):
         """Discard key operation (no-op)."""
 
-    def get_key(self, header) -> CollectionKey:
+    def get_key(self, _header) -> CollectionKey:
         return self.NO_PRIVATE_KEY
 
 

packages/otdf-python/src/otdf_python/kas_client.py

@@ -553,7 +553,7 @@ def _normalize_session_key_type(self, session_key_type):
             return RSA_KEY_TYPE
         return session_key_type
 
-    def _prepare_ec_keypair(self, session_key_type):
+    def _prepare_ec_keypair(self, _session_key_type):
         """Prepare EC key pair for unwrapping.
 
         Args:

packages/otdf-python/src/otdf_python/nanotdf.py

@@ -468,7 +468,7 @@ def create_nano_tdf(
         return len(header_bytes) + 3 + payload_length
 
     def _kas_unwrap(
-        self, nano_tdf_data: bytes, header_len: int, wrapped_key: bytes
+        self, nano_tdf_data: bytes, header_len: int, _wrapped_key: bytes
     ) -> bytes | None:
         try:
             # For NanoTDF, send the entire header to KAS
@@ -622,7 +622,7 @@ def read_nano_tdf(  # noqa: C901
         # Try KAS unwrap first if services available
         if self.services:
             try:
-                key = self._kas_unwrap(nano_tdf_data, header_len, wrapped_key=b"")
+                key = self._kas_unwrap(nano_tdf_data, header_len, b"")
                 if key:
                     logging.info(
                         "Successfully unwrapped NanoTDF key via KAS (ECDH mode)"

packages/otdf-python/src/otdf_python/policy_binding_serializer.py

@@ -25,7 +25,7 @@ class PolicyBindingSerializer:
 
     @staticmethod
     def deserialize(
-        json_data: Any, typeofT: type | None = None, context: Any = None
+        json_data: Any, _typeof_t: type | None = None, _context: Any = None
     ) -> Any:
         if isinstance(json_data, dict):
             return PolicyBinding(**json_data)
@@ -35,7 +35,7 @@ def deserialize(
 
     @staticmethod
     def serialize(
-        src: Any, typeofSrc: type | None = None, context: Any = None
+        src: Any, _typeof_src: type | None = None, _context: Any = None
     ) -> dict | str:
         if isinstance(src, PolicyBinding):
             return vars(src)

packages/otdf-python/src/otdf_python/policy_info.py

@@ -43,7 +43,7 @@ def write_into_buffer(self, buffer: bytearray, offset: int = 0) -> int:
         return offset - start
 
     @staticmethod
-    def from_bytes_with_size(buffer: bytes, ecc_mode):
+    def from_bytes_with_size(buffer: bytes, _ecc_mode):
         # Parse policy_type (1 byte), body_len (2 bytes), body
         # Note: binding is NOT part of PolicyInfo - it's read separately in Header
         offset = 0

packages/otdf-python/src/otdf_python/sdk.py

@@ -97,9 +97,9 @@ def unwrap(self, key_access: Any, policy: str, session_key_type: Any) -> bytes:
 
     def unwrap_nanotdf(
         self,
-        curve: Any,
-        header: str,
-        kas_url: str,
+        curve: Any,  # noqa: ARG002
+        header: str,  # noqa: ARG002
+        kas_url: str,  # noqa: ARG002
         wrapped_key: bytes | None = None,
         kas_private_key: str | None = None,
         mock: bool = False,

pyproject.toml

@@ -22,6 +22,7 @@ lint.ignore = [
     "E501",
 ]
 lint.select = [
+    "ARG",    # flake8-unused-arguments
     "B",      # flake8-bugbear
     "C4",     # flake8-comprehensions
     "C90",    # McCabe complexity

tests/integration/otdfctl_to_python/test_nanotdf_cli_comparison.py

@@ -22,6 +22,8 @@
     run_otdfctl_encrypt_command,
 )
 
+pytestmark = pytest.mark.skip(reason="NanoTDF tests are disabled")
+
 logger = logging.getLogger(__name__)
 
 

tests/integration/otdfctl_to_python/test_python_nanotdf_only.py

@@ -15,6 +15,8 @@
     validate_plaintext_file_created,
 )
 
+pytestmark = pytest.mark.skip(reason="NanoTDF tests are disabled")
+
 logger = logging.getLogger(__name__)
 
 

tests/test_ecdh.py

@@ -18,6 +18,8 @@
     get_curve,
 )
 
+pytestmark = pytest.mark.skip(reason="NanoTDF tests are disabled")
+
 
 class TestCurveOperations:
     """Test basic curve operations."""

tests/test_header.py

@@ -1,11 +1,14 @@
 import unittest
 
+import pytest
 from otdf_python.ecc_mode import ECCMode
 from otdf_python.header import Header
 from otdf_python.policy_info import PolicyInfo
 from otdf_python.resource_locator import ResourceLocator
 from otdf_python.symmetric_and_payload_config import SymmetricAndPayloadConfig
 
+pytestmark = pytest.mark.skip(reason="NanoTDF tests are disabled")
+
 
 class TestHeader(unittest.TestCase):
     def test_header_fields(self):

tests/test_kas_client.py

@@ -65,7 +65,7 @@ def test_get_public_key_fetches_and_caches(mock_access_service_client):
     mock_rpc_response.kid = "kid2"
     mock_rpc_response.public_key = "public-key-data"
 
-    def mock_public_key_call(*args, **kwargs):
+    def mock_public_key_call(*_args, **_kwargs):
         return mock_rpc_response
 
     mock_rpc_client_instance.public_key = mock_public_key_call

tests/test_nanotdf.py

@@ -6,6 +6,8 @@
 from otdf_python.config import NanoTDFConfig
 from otdf_python.nanotdf import InvalidNanoTDFConfig, NanoTDF, NanoTDFMaxSizeLimit
 
+pytestmark = pytest.mark.skip(reason="NanoTDF tests are disabled")
+
 
 def test_nanotdf_roundtrip():
     """Test NanoTDF encrypt and decrypt roundtrip."""

tests/test_nanotdf_ecdh.py

@@ -8,6 +8,8 @@
 from otdf_python.ecdh import generate_ephemeral_keypair
 from otdf_python.nanotdf import NanoTDF
 
+pytestmark = pytest.mark.skip(reason="NanoTDF tests are disabled")
+
 
 class TestNanoTDFWithECDH:
     """Test NanoTDF encryption/decryption using ECDH key exchange."""

tests/test_nanotdf_ecdsa_struct.py

@@ -6,6 +6,8 @@
     NanoTDFECDSAStruct,
 )
 
+pytestmark = pytest.mark.skip(reason="NanoTDF tests are disabled")
+
 
 def test_from_bytes():
     """Test creating a NanoTDFECDSAStruct from bytes."""

tests/test_nanotdf_integration.py

@@ -8,6 +8,8 @@
 from otdf_python.config import KASInfo, NanoTDFConfig
 from otdf_python.nanotdf import NanoTDF
 
+pytestmark = pytest.mark.skip(reason="NanoTDF tests are disabled")
+
 
 @pytest.mark.integration
 def test_nanotdf_kas_roundtrip():

tests/test_nanotdf_type.py

@@ -2,6 +2,7 @@
 
 import unittest
 
+import pytest
 from otdf_python.nanotdf_type import (
     Cipher,
     ECCurve,
@@ -10,6 +11,8 @@
     Protocol,
 )
 
+pytestmark = pytest.mark.skip(reason="NanoTDF tests are disabled")
+
 
 class TestNanoTDFType(unittest.TestCase):
     """Tests for NanoTDF type enums."""

tests/test_sdk_builder.py

@@ -131,7 +131,7 @@ def test_ssl_context_from_directory():
 def test_get_token_from_client_credentials(mock_post, mock_get):
     """Test getting OAuth token from client credentials."""
 
-    def get_side_effect(url, **kwargs):
+    def get_side_effect(url, **_kwargs):
         resp = MagicMock()
         if "openid-configuration" in url:
             resp.status_code = 200
@@ -165,7 +165,7 @@ def get_side_effect(url, **kwargs):
 def test_get_token_failure(mock_post, mock_get):
     """Test handling of token acquisition failure."""
 
-    def get_side_effect(url, **kwargs):
+    def get_side_effect(url, **_kwargs):
         resp = MagicMock()
         if "openid-configuration" in url:
             resp.status_code = 200

tests/test_sdk_mock.py

@@ -6,19 +6,19 @@
 class MockKAS(KAS):
     """Mock KAS implementation for testing."""
 
-    def get_public_key(self, kas_info):
+    def get_public_key(self, _kas_info):
         """Return mock public key."""
         return "mock-public-key"
 
-    def get_ec_public_key(self, kas_info, curve):
+    def get_ec_public_key(self, _kas_info, _curve):
         """Return mock EC public key."""
         return "mock-ec-public-key"
 
-    def unwrap(self, key_access, policy, session_key_type):
+    def unwrap(self, _key_access, _policy, _session_key_type):
         """Return mock unwrapped key."""
         return b"mock-unwrapped-key"
 
-    def unwrap_nanotdf(self, curve, header, kas_url):
+    def unwrap_nanotdf(self, _curve, _header, _kas_url):
         """Return mock unwrapped NanoTDF key."""
         return b"mock-unwrapped-nanotdf"