[autofix.ci] apply automated fixes (attempt 3/3)

This commit is contained in:
autofix-ci[bot] 2025-09-22 17:31:59 +00:00 committed by GitHub
parent 2c343e98cc
commit cf222ecfed
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 19 additions and 20 deletions

View File

@ -797,12 +797,12 @@ class MailConfig(BaseSettings):
)
# Microsoft OAuth 2.0 configuration for SMTP
MICROSOFT_OAUTH2_CLIENT_ID: Optional[str] = Field(
MICROSOFT_OAUTH2_CLIENT_ID: str | None = Field(
description="Microsoft OAuth 2.0 client ID for SMTP authentication",
default=None,
)
MICROSOFT_OAUTH2_CLIENT_SECRET: Optional[str] = Field(
MICROSOFT_OAUTH2_CLIENT_SECRET: str | None = Field(
description="Microsoft OAuth 2.0 client secret for SMTP authentication",
default=None,
)
@ -812,7 +812,7 @@ class MailConfig(BaseSettings):
default="common",
)
MICROSOFT_OAUTH2_ACCESS_TOKEN: Optional[str] = Field(
MICROSOFT_OAUTH2_ACCESS_TOKEN: str | None = Field(
description="Microsoft OAuth 2.0 access token for SMTP authentication",
default=None,
)

View File

@ -83,7 +83,7 @@ class Mail:
case _:
raise ValueError(f"Unsupported mail type {mail_type}")
def _get_oauth_token(self) -> Optional[str]:
def _get_oauth_token(self) -> str | None:
"""Get OAuth access token using client credentials flow"""
try:
from libs.mail.oauth_email import MicrosoftEmailOAuth
@ -110,7 +110,7 @@ class Mail:
logging.warning("Failed to obtain OAuth 2.0 access token: %s", str(e))
return None
def send(self, to: str, subject: str, html: str, from_: Optional[str] = None):
def send(self, to: str, subject: str, html: str, from_: str | None = None):
if not self._client:
raise ValueError("Mail client is not initialized")

View File

@ -23,7 +23,7 @@ class EmailOAuth:
client_id: str,
client_secret: str,
redirect_uri: str,
http_client: Optional[OAuthHTTPClientProtocol] = None,
http_client: OAuthHTTPClientProtocol | None = None,
):
self.client_id = client_id
self.client_secret = client_secret
@ -67,12 +67,12 @@ class MicrosoftEmailOAuth(EmailOAuth):
client_secret: str,
redirect_uri: str,
tenant_id: str = "common",
http_client: Optional[OAuthHTTPClientProtocol] = None,
http_client: OAuthHTTPClientProtocol | None = None,
):
super().__init__(client_id, client_secret, redirect_uri, http_client)
self.tenant_id = tenant_id
def get_authorization_url(self, invite_token: Optional[str] = None) -> str:
def get_authorization_url(self, invite_token: str | None = None) -> str:
"""Generate OAuth authorization URL"""
params = {
"client_id": self.client_id,

View File

@ -11,13 +11,13 @@ class OAuthHTTPClientProtocol(ABC):
@abstractmethod
def post(
self, url: str, data: dict[str, Union[str, int]], headers: Optional[dict[str, str]] = None
self, url: str, data: dict[str, Union[str, int]], headers: dict[str, str] | None = None
) -> dict[str, Union[str, int, dict, list]]:
"""Make a POST request"""
pass
@abstractmethod
def get(self, url: str, headers: Optional[dict[str, str]] = None) -> dict[str, Union[str, int, dict, list]]:
def get(self, url: str, headers: dict[str, str] | None = None) -> dict[str, Union[str, int, dict, list]]:
"""Make a GET request"""
pass
@ -26,7 +26,7 @@ class OAuthHTTPClient(OAuthHTTPClientProtocol):
"""Default implementation using requests library"""
def post(
self, url: str, data: dict[str, Union[str, int]], headers: Optional[dict[str, str]] = None
self, url: str, data: dict[str, Union[str, int]], headers: dict[str, str] | None = None
) -> dict[str, Union[str, int, dict, list]]:
"""Make a POST request"""
response = requests.post(url, data=data, headers=headers or {})
@ -37,7 +37,7 @@ class OAuthHTTPClient(OAuthHTTPClientProtocol):
"headers": dict(response.headers),
}
def get(self, url: str, headers: Optional[dict[str, str]] = None) -> dict[str, Union[str, int, dict, list]]:
def get(self, url: str, headers: dict[str, str] | None = None) -> dict[str, Union[str, int, dict, list]]:
"""Make a GET request"""
response = requests.get(url, headers=headers or {})
response.raise_for_status()

View File

@ -83,12 +83,12 @@ class SMTPClient:
from_addr: str,
use_tls: bool = False,
opportunistic_tls: bool = False,
oauth_access_token: Optional[str] = None,
oauth_access_token: str | None = None,
auth_type: str = "basic",
connection_factory: Optional[SMTPConnectionFactory] = None,
ssl_connection_factory: Optional[SMTPConnectionFactory] = None,
authenticator: Optional[SMTPAuthenticator] = None,
message_builder: Optional[SMTPMessageBuilder] = None,
connection_factory: SMTPConnectionFactory | None = None,
ssl_connection_factory: SMTPConnectionFactory | None = None,
authenticator: SMTPAuthenticator | None = None,
message_builder: SMTPMessageBuilder | None = None,
):
self.server = server
self.port = port

View File

@ -36,7 +36,7 @@ class MockHTTPClient(OAuthHTTPClientProtocol):
self.get_responses.append(json_data)
def post(
self, url: str, data: dict[str, Union[str, int]], headers: Optional[dict[str, str]] = None
self, url: str, data: dict[str, Union[str, int]], headers: dict[str, str] | None = None
) -> dict[str, Union[str, int, dict, list]]:
"""Mock POST request"""
self.post_calls.append({"url": url, "data": data, "headers": headers})
@ -54,7 +54,7 @@ class MockHTTPClient(OAuthHTTPClientProtocol):
"headers": {},
}
def get(self, url: str, headers: Optional[dict[str, str]] = None) -> dict[str, Union[str, int, dict, list]]:
def get(self, url: str, headers: dict[str, str] | None = None) -> dict[str, Union[str, int, dict, list]]:
"""Mock GET request"""
self.get_calls.append({"url": url, "headers": headers})

View File

@ -1,7 +1,6 @@
from unittest.mock import MagicMock, patch
import pytest
from libs.smtp import SMTPClient