Add typing to sse.py

.github/workflows/ci.yml

@@ -30,11 +30,11 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        python-version: ["3.7", "3.8", "3.9", "3.10", "3.11"]
+        python-version: ["3.7", "3.8", "3.9", "3.10", "3.11", "3.12"]
         os: [ubuntu-latest, windows-latest, macos-latest]
 
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
     - name: Run spell check on Ubuntu
       if: matrix.os == 'ubuntu-latest'
       uses: codespell-project/actions-codespell@master
@@ -59,7 +59,7 @@ jobs:
         echo "/Users/runner/.local/bin" >> $GITHUB_PATH
     - name: Run unit tests
       run: |
-        python setup.py install
+        pip install -e .
         pytest
     - name: Run functional tests on Ubuntu
       if: matrix.os == 'ubuntu-latest'

minio/api.py

@@ -165,7 +165,8 @@ def __init__(
         )
 
     def __del__(self):
-        self._http.clear()
+        if hasattr(self, "_http"):  # Only required for unit test run
+            self._http.clear()
 
     def _handle_redirect_response(
             self, method, bucket_name, response, retry=False,

minio/error.py

@@ -45,8 +45,8 @@ def __init__(self, code, content_type, body):
         self._content_type = content_type
         self._body = body
         super().__init__(
-            f"non-XML response from server; "
-            f"Response code: {code}, Content-Type: {content_type}, Body: {body}"
+            f"non-XML response from server; Response code: {code}, "
+            f"Content-Type: {content_type}, Body: {body}"
         )
 
     def __reduce__(self):

minio/helpers.py

@@ -123,11 +123,11 @@ def _validate_sizes(object_size, part_size):
     if part_size > 0:
         if part_size < MIN_PART_SIZE:
             raise ValueError(
-                f"part size {part_size} is not supported; minimum allowed 5MiB",
+                f"part size {part_size} is not supported; minimum allowed 5MiB"
             )
         if part_size > MAX_PART_SIZE:
             raise ValueError(
-                f"part size {part_size} is not supported; maximum allowed 5GiB",
+                f"part size {part_size} is not supported; maximum allowed 5GiB"
             )
 
     if object_size >= 0:

minio/select.py

@@ -391,7 +391,8 @@ def _read(self):
 
         if headers.get(":message-type") == "error":
             raise MinioException(
-                f"{headers.get(':error-code')}: {headers.get(':error-message')}"
+                f"{headers.get(':error-code')}: "
+                f"{headers.get(':error-message')}"
             )
 
         if headers.get(":event-type") == "End":

minio/sse.py

@@ -27,29 +27,30 @@
 import base64
 import json
 from abc import ABCMeta, abstractmethod
+from typing import Dict
 
 
 class Sse:
     """Server-side encryption base class."""
     __metaclass__ = ABCMeta
 
     @abstractmethod
-    def headers(self):
+    def headers(self) -> Dict[str, str]:
         """Return headers."""
 
-    def tls_required(self):  # pylint: disable=no-self-use
+    def tls_required(self) -> bool:  # pylint: disable=no-self-use
         """Return TLS required to use this server-side encryption."""
         return True
 
-    def copy_headers(self):  # pylint: disable=no-self-use
+    def copy_headers(self) -> Dict[str, str]:  # pylint: disable=no-self-use
         """Return copy headers."""
         return {}
 
 
 class SseCustomerKey(Sse):
     """ Server-side encryption - customer key type."""
 
-    def __init__(self, key):
+    def __init__(self, key: bytes):
         if len(key) != 32:
             raise ValueError(
                 "SSE-C keys need to be 256 bit base64 encoded",
@@ -71,17 +72,17 @@ def __init__(self, key):
             md5key,
         }
 
-    def headers(self):
+    def headers(self) -> Dict[str, str]:
         return self._headers.copy()
 
-    def copy_headers(self):
+    def copy_headers(self) -> Dict[str, str]:
         return self._copy_headers.copy()
 
 
 class SseKMS(Sse):
     """Server-side encryption - KMS type."""
 
-    def __init__(self, key, context):
+    def __init__(self, key: str, context: Dict):
         self._headers = {
             "X-Amz-Server-Side-Encryption-Aws-Kms-Key-Id": key,
             "X-Amz-Server-Side-Encryption": "aws:kms"
@@ -92,17 +93,17 @@ def __init__(self, key, context):
                 base64.b64encode(data).decode()
             )
 
-    def headers(self):
+    def headers(self) -> Dict[str, str]:
         return self._headers.copy()
 
 
 class SseS3(Sse):
     """Server-side encryption - S3 type."""
 
-    def headers(self):
+    def headers(self) -> Dict[str, str]:
         return {
             "X-Amz-Server-Side-Encryption": "AES256"
         }
 
-    def tls_required(self):
+    def tls_required(self) -> bool:
         return False

minio/time.py

@@ -21,6 +21,12 @@
 import time as ctime
 from datetime import datetime, timezone
 
+try:
+    from datetime import UTC
+    _UTC_IMPORTED = True
+except ImportError:
+    _UTC_IMPORTED = False
+
 _WEEK_DAYS = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
 _MONTHS = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct",
            "Nov", "Dec"]
@@ -102,6 +108,8 @@ def to_amz_date(value):
 
 def utcnow():
     """Timezone-aware wrapper to datetime.utcnow()."""
+    if _UTC_IMPORTED:
+        return datetime.now(UTC).replace(tzinfo=timezone.utc)
     return datetime.utcnow().replace(tzinfo=timezone.utc)
 
 

tests/functional/tests.py

@@ -140,7 +140,9 @@ def _call_test(func, *args, **kwargs):
     if log_entry.get("method"):
         # pylint: disable=deprecated-method
         args_string = ', '.join(getfullargspec(log_entry["method"]).args[1:])
-        log_entry["function"] = f"{log_entry['method'].__name__}({args_string})"
+        log_entry["function"] = (
+            f"{log_entry['method'].__name__}({args_string})"
+        )
     log_entry["args"] = {
         k: v for k, v in log_entry.get("args", {}).items() if v
     }

tests/unit/crypto_test.py

@@ -25,7 +25,10 @@ def test_correct(self):
         plaintext = "Hello MinIO!"
         encrypted = encrypt(plaintext.encode(), secret)
         decrypted = decrypt(encrypted, secret).decode()
-        self.assertEquals(plaintext, decrypted)
+        if hasattr(self, "assertEquals"):
+            self.assertEquals(plaintext, decrypted)
+        else:
+            self.assertEqual(plaintext, decrypted)
 
     def test_wrong(self):
         secret = "topsecret"

tests/unit/minio_mocks.py

@@ -69,6 +69,9 @@ class MockConnection(object):
     def __init__(self):
         self.requests = []
 
+    def clear(self):
+        pass
+
     def mock_add_request(self, request):
         self.requests.append(request)
 

tests/unit/retention_test.py

@@ -20,11 +20,12 @@
 from minio import xml
 from minio.commonconfig import COMPLIANCE, GOVERNANCE
 from minio.retention import Retention
+from minio.time import utcnow
 
 
 class RetentionTest(TestCase):
     def test_config(self):
-        config = Retention(GOVERNANCE, datetime.utcnow() + timedelta(days=10))
+        config = Retention(GOVERNANCE, utcnow() + timedelta(days=10))
         xml.marshal(config)
 
         config = xml.unmarshal(