diff --git a/eng/tools/azure-sdk-tools/ci_tools/functions.py b/eng/tools/azure-sdk-tools/ci_tools/functions.py index c2cc9cdcb216..154f6f6feeb8 100644 --- a/eng/tools/azure-sdk-tools/ci_tools/functions.py +++ b/eng/tools/azure-sdk-tools/ci_tools/functions.py @@ -745,7 +745,20 @@ def resolve_compatible_package(package_name: str, immutable_requirements: List[R versions = [v for v in versions if not v.is_prerelease] for version in versions: - version_release = pypi.project_release(package_name, version).get("info", {}).get("requires_dist", []) + # ``project_release`` reads per-version metadata from the PyPI JSON API. Under the + # ``CFSClean`` network isolation policy pypi.org is unreachable, so this best-effort + # compatibility walk cannot fetch metadata. Treat an unreachable endpoint as "no info" + # and continue rather than hard-failing the entire dependency check. + try: + version_release = pypi.project_release(package_name, version).get("info", {}).get("requires_dist", []) + except Exception as e: + logging.warning( + "Unable to retrieve release metadata for %s==%s while resolving a compatible version: %s", + package_name, + version, + e, + ) + continue if version_release: requirements_for_dev_req = [Requirement(r) for r in version_release] diff --git a/eng/tools/azure-sdk-tools/ci_tools/scenario/dependency_resolution.py b/eng/tools/azure-sdk-tools/ci_tools/scenario/dependency_resolution.py index 10c2951ada61..1c676b781076 100644 --- a/eng/tools/azure-sdk-tools/ci_tools/scenario/dependency_resolution.py +++ b/eng/tools/azure-sdk-tools/ci_tools/scenario/dependency_resolution.py @@ -29,6 +29,29 @@ NEW_DEV_REQ_FILE = "new_dev_requirements.txt" PKGS_TXT_FILE = "packages.txt" +# Internal Azure Artifacts (CFS) feed used when no usable index is configured. Version resolution +# below builds a ``PyPIClient`` whose backend is chosen from ``PIP_INDEX_URL``: an Azure Artifacts +# URL selects the AzDO REST backend, anything else falls back to pypi.org. Under the ``CFSClean`` +# network isolation policy, pypi.org egress is blocked, so an unset or pypi.org index would make +# resolution fail. Mirror the default applied by ``azpysdk.main`` so shared resolution always +# targets the CFS feed unless an Azure Artifacts index is already set. +CFS_INDEX_URL = "https://pkgs.dev.azure.com/azure-sdk/public/_packaging/azure-sdk-for-python/pypi/simple/" + + +def _ensure_resolution_index() -> None: + """Point ``PIP_INDEX_URL`` at the CFS feed unless an Azure Artifacts index is already set. + + This keeps ``PyPIClient`` on its AzDO backend during version resolution, which is required + under the ``CFSClean`` policy that blocks pypi.org. + """ + + current = os.environ.get("PIP_INDEX_URL", "") + if "pkgs.dev.azure.com" in current: + return + if current: + logger.info("Overriding PIP_INDEX_URL %s with CFS feed for dependency resolution.", current) + os.environ["PIP_INDEX_URL"] = CFS_INDEX_URL + # GENERIC_OVERRIDES dictionaries pair a specific dependency with a MINIMUM or MAXIMUM inclusive bound. # During LATEST and MINIMUM dependency checks, we sometimes need to ignore versions for various compatibility # reasons. @@ -121,6 +144,8 @@ def install_dependent_packages( python_exe = python_executable or sys.executable + _ensure_resolution_index() + released_packages = find_released_packages(setup_py_file_path, dependency_type) override_added_packages: List[str] = [] diff --git a/sdk/eventhub/azure-eventhub/CHANGELOG.md b/sdk/eventhub/azure-eventhub/CHANGELOG.md index f287cb0b8fde..7ebacae7524f 100644 --- a/sdk/eventhub/azure-eventhub/CHANGELOG.md +++ b/sdk/eventhub/azure-eventhub/CHANGELOG.md @@ -1,5 +1,11 @@ # Release History +## 5.15.2 (Unreleased) + +### Bugs Fixed + +- Fixed a bug where the async pure-Python AMQP transport failed to connect with `[Errno 22] Invalid argument` (`amqp:socket-error`) inside containerized/virtualized environments such as Docker Desktop on macOS. The transport no longer reads back and re-applies platform-negotiated TCP options (e.g. `TCP_MAXSEG`) that some platforms reject via `setsockopt`. Also fixed the async transport to apply default TCP socket settings even when no custom `socket_settings` are provided. ([#45394](https://github.com/Azure/azure-sdk-for-python/issues/45394)) + ## 5.15.1 (2025-11-11) ### Bugs Fixed diff --git a/sdk/eventhub/azure-eventhub/api.md b/sdk/eventhub/azure-eventhub/api.md new file mode 100644 index 000000000000..2465799687a8 --- /dev/null +++ b/sdk/eventhub/azure-eventhub/api.md @@ -0,0 +1,1032 @@ +```py +namespace azure.eventhub + + def azure.eventhub.parse_connection_string(conn_str: str) -> EventHubConnectionStringProperties: ... + + + class azure.eventhub.CheckpointStore: + + @abstractmethod + def claim_ownership( + self, + ownership_list: Iterable[Dict[str, Any]], + **kwargs: Any + ) -> Iterable[Dict[str, Any]]: ... + + @abstractmethod + def list_checkpoints( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + **kwargs: Any + ) -> Iterable[Dict[str, Any]]: ... + + @abstractmethod + def list_ownership( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + **kwargs: Any + ) -> Iterable[Dict[str, Any]]: ... + + @abstractmethod + def update_checkpoint( + self, + checkpoint: Dict[str, Optional[Union[str, int]]], + **kwargs: Any + ) -> None: ... + + + class azure.eventhub.CloseReason(Enum): + OWNERSHIP_LOST = 1 + SHUTDOWN = 0 + + + class azure.eventhub.EventData: + property body: PrimitiveTypes # Read-only + property body_type: AmqpMessageBodyType # Read-only + property content_type: Optional[str] + property correlation_id: Optional[str] + property enqueued_time: Optional[datetime] # Read-only + property message: Union[Message, LegacyMessage] + property message_id: Optional[str] + property offset: Optional[str] # Read-only + property partition_key: Optional[bytes] # Read-only + property properties: Dict[Union[str, bytes], Any] + property raw_amqp_message: AmqpAnnotatedMessage # Read-only + property sequence_number: Optional[int] # Read-only + property system_properties: Dict[bytes, Any] # Read-only + + def __init__(self, body: Optional[Union[str, bytes, List[AnyStr]]] = None) -> None: ... + + def __message_content__(self) -> MessageContent: ... + + def __repr__(self) -> str: ... + + def __str__(self) -> str: ... + + @classmethod + def from_bytes(cls, message: bytes) -> EventData: ... + + @classmethod + def from_message_content( + cls, + content: bytes, + content_type: str, + **kwargs: Any + ) -> EventData: ... + + def body_as_json(self, encoding: str = "UTF-8") -> Dict[str, Any]: ... + + def body_as_str(self, encoding: str = "UTF-8") -> str: ... + + + class azure.eventhub.EventDataBatch: + property message: Union[BatchMessage, LegacyBatchMessage] + property size_in_bytes: int # Read-only + + def __init__( + self, + max_size_in_bytes: Optional[int] = None, + partition_id: Optional[str] = None, + partition_key: Optional[Union[str, bytes]] = None, + **kwargs: Any + ) -> None: ... + + def __len__(self) -> int: ... + + def __repr__(self) -> str: ... + + def add(self, event_data: Union[EventData, AmqpAnnotatedMessage]) -> None: ... + + + class azure.eventhub.EventHubConnectionStringProperties(DictMixin): + property endpoint: str # Read-only + property eventhub_name: Optional[str] # Read-only + property fully_qualified_namespace: str # Read-only + property shared_access_key: Optional[str] # Read-only + property shared_access_key_name: Optional[str] # Read-only + property shared_access_signature: Optional[str] # Read-only + + def __contains__(self, key: str) -> bool: ... + + def __delitem__(self, key: str) -> None: ... + + def __eq__(self, other: Any) -> bool: ... + + def __getitem__(self, key: str) -> Any: ... + + def __init__( + self, + *, + endpoint: str, + eventhub_name: Optional[str] = ..., + fully_qualified_namespace: str, + shared_access_key: Optional[str] = ..., + shared_access_key_name: Optional[str] = ..., + shared_access_signature: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + def __len__(self) -> int: ... + + def __ne__(self, other: Any) -> bool: ... + + def __repr__(self) -> str: ... + + def __setitem__( + self, + key: str, + item: Any + ) -> None: ... + + def __str__(self) -> str: ... + + def get( + self, + key: str, + default: Optional[Any] = None + ) -> Any: ... + + def has_key(self, k: str) -> bool: ... + + def items(self) -> List[Tuple[str, Any]]: ... + + def keys(self) -> List[str]: ... + + def update( + self, + *args: Any, + **kwargs: Any + ) -> None: ... + + def values(self) -> List[Any]: ... + + + class azure.eventhub.EventHubConsumerClient(ClientBase): implements ContextManager + + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + credential: CredentialTypes, + *, + auth_timeout: Optional[float] = ..., + checkpoint_store: Union[CheckpointStore, None] = ..., + connection_verify: Union[str, None] = ..., + custom_endpoint_address: Union[str, None] = ..., + idle_timeout: Optional[float] = ..., + load_balancing_interval: Optional[float] = ..., + load_balancing_strategy: Union[str, LoadBalancingStrategy] = ..., + logging_enable: Optional[bool] = ..., + partition_ownership_expiration_interval: Optional[float] = ..., + retry_backoff_factor: Optional[float] = ..., + retry_backoff_max: Optional[float] = ..., + retry_mode: str = ..., + retry_total: int = ..., + socket_timeout: Optional[float] = ..., + ssl_context: Union[SSLContext, None] = ..., + transport_type: TransportType = ..., + uamqp_transport: bool = ..., + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + @classmethod + def from_connection_string( + cls, + conn_str: str, + consumer_group: str, + *, + auth_timeout: float = 60, + checkpoint_store: Optional[CheckpointStore] = ..., + connection_verify: Optional[str] = ..., + custom_endpoint_address: Optional[str] = ..., + eventhub_name: Optional[str] = ..., + http_proxy: Optional[Dict[str, Union[str, int]]] = ..., + idle_timeout: Optional[float] = ..., + load_balancing_interval: float = 30, + load_balancing_strategy: Union[str, LoadBalancingStrategy] = LoadBalancingStrategy.GREEDY, + logging_enable: bool = False, + partition_ownership_expiration_interval: Optional[float] = ..., + retry_backoff_factor: float = 0.8, + retry_backoff_max: float = 120, + retry_mode: Literal["exponential", "fixed"] = "exponential", + retry_total: int = 3, + ssl_context: Optional[SSLContext] = ..., + transport_type: TransportType = TransportType.Amqp, + uamqp_transport: bool = False, + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> EventHubConsumerClient: ... + + def close(self) -> None: ... + + def get_eventhub_properties(self) -> Dict[str, Any]: ... + + def get_partition_ids(self) -> List[str]: ... + + def get_partition_properties(self, partition_id: str) -> Dict[str, Any]: ... + + def receive( + self, + on_event: Callable[[PartitionContext, Optional[EventData]], None], + *, + max_wait_time: Optional[float] = ..., + on_error: Optional[Callable[[PartitionContext, Exception], None]] = ..., + on_partition_close: Optional[Callable[[PartitionContext, CloseReason], None]] = ..., + on_partition_initialize: Optional[Callable[[PartitionContext], None]] = ..., + owner_level: Optional[int] = ..., + partition_id: Optional[str] = ..., + prefetch: int = 300, + starting_position: Optional[Union[str, int, datetime, Dict[str, Any]]] = ..., + starting_position_inclusive: Union[bool, Dict[str, bool]] = False, + track_last_enqueued_event_properties: bool = False, + **kwargs: Any + ) -> None: ... + + def receive_batch( + self, + on_event_batch: Callable[[PartitionContext, List[EventData]], None], + *, + max_batch_size: int = 300, + max_wait_time: Optional[float] = ..., + on_error: Optional[Callable[[PartitionContext, Exception], None]] = ..., + on_partition_close: Optional[Callable[[PartitionContext, CloseReason], None]] = ..., + on_partition_initialize: Optional[Callable[[PartitionContext], None]] = ..., + owner_level: Optional[int] = ..., + partition_id: Optional[str] = ..., + prefetch: int = 300, + starting_position: Optional[Union[str, int, datetime, Dict[str, Any]]] = ..., + starting_position_inclusive: Union[bool, Dict[str, bool]] = False, + track_last_enqueued_event_properties: bool = False, + **kwargs: Any + ) -> None: ... + + + class azure.eventhub.EventHubProducerClient(ClientBase): implements ContextManager + property total_buffered_event_count: Optional[int] # Read-only + + @overload + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + credential: CredentialTypes, + *, + auth_timeout: Optional[float] = ..., + buffer_concurrency: Union[ThreadPoolExecutor, int, None] = ..., + buffered_mode: Literal[False] = False, + connection_verify: Optional[str] = ..., + custom_endpoint_address: Optional[str] = ..., + http_proxy: Optional[Dict] = ..., + idle_timeout: Optional[float] = ..., + logging_enable: Optional[bool] = ..., + max_buffer_length: Optional[int] = ..., + max_wait_time: Optional[float] = ..., + on_error: Optional[Callable[[SendEventTypes, Optional[str], Exception], None]] = ..., + on_success: Optional[Callable[[SendEventTypes, Optional[str]], None]] = ..., + retry_backoff_factor: Optional[float] = ..., + retry_backoff_max: Optional[float] = ..., + retry_mode: str = ..., + retry_total: int = ..., + socket_timeout: Optional[float] = ..., + ssl_context: Union[SSLContext, None] = ..., + transport_type: TransportType = ..., + uamqp_transport: bool = ..., + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + @overload + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + credential: CredentialTypes, + *, + auth_timeout: Optional[float] = ..., + buffer_concurrency: Optional[Union[ThreadPoolExecutor, int]] = ..., + buffered_mode: Literal[True], + connection_verify: Optional[str] = ..., + custom_endpoint_address: Optional[str] = ..., + http_proxy: Optional[Dict] = ..., + idle_timeout: Optional[float] = ..., + logging_enable: Optional[bool] = ..., + max_buffer_length: int = 1500, + max_wait_time: float = 1, + on_error: Callable[[SendEventTypes, Optional[str], Exception], None], + on_success: Callable[[SendEventTypes, Optional[str]], None], + retry_backoff_factor: Optional[float] = ..., + retry_backoff_max: Optional[float] = ..., + retry_mode: str = ..., + retry_total: int = ..., + socket_timeout: Optional[float] = ..., + ssl_context: Union[SSLContext, None] = ..., + transport_type: TransportType = ..., + uamqp_transport: bool = ..., + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + @classmethod + @overload + def from_connection_string( + cls, + conn_str: str, + *, + buffered_mode: Literal[False] = False, + eventhub_name: Optional[str] = ..., + **kwargs: Any + ) -> EventHubProducerClient: ... + + @classmethod + @overload + def from_connection_string( + cls, + conn_str: str, + *, + buffer_concurrency: Optional[Union[ThreadPoolExecutor, int]] = ..., + buffered_mode: Literal[True], + eventhub_name: Optional[str] = ..., + max_buffer_length: int = 1500, + max_wait_time: float = 1, + on_error: Callable[[SendEventTypes, Optional[str], Exception], None], + on_success: Callable[[SendEventTypes, Optional[str]], None], + **kwargs: Any + ) -> EventHubProducerClient: ... + + def close( + self, + *, + flush: bool = True, + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + def create_batch( + self, + *, + max_size_in_bytes: Optional[int] = ..., + partition_id: Optional[str] = ..., + partition_key: Optional[str] = ..., + **kwargs: Any + ) -> EventDataBatch: ... + + def flush( + self, + *, + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + def get_buffered_event_count(self, partition_id: str) -> Optional[int]: ... + + def get_eventhub_properties(self) -> Dict[str, Any]: ... + + def get_partition_ids(self) -> List[str]: ... + + def get_partition_properties(self, partition_id: str) -> Dict[str, Any]: ... + + def send_batch( + self, + event_data_batch: Union[EventDataBatch, SendEventTypes], + *, + partition_id: Optional[str] = ..., + partition_key: Optional[str] = ..., + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + def send_event( + self, + event_data: Union[EventData, AmqpAnnotatedMessage], + *, + partition_id: Optional[str] = ..., + partition_key: Optional[str] = ..., + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + + class azure.eventhub.EventHubSharedKeyCredential: + + def __init__( + self, + policy: str, + key: str + ) -> None: ... + + def get_token( + self, + *scopes: str, + **kwargs: Any + ) -> AccessToken: ... + + + class azure.eventhub.LoadBalancingStrategy(Enum): + BALANCED = "balanced" + GREEDY = "greedy" + + + class azure.eventhub.PartitionContext: + property last_enqueued_event_properties: Optional[Dict[str, Any]] # Read-only + + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + partition_id: str, + checkpoint_store: Optional[CheckpointStore] = None + ) -> None: ... + + def update_checkpoint( + self, + event: Optional[EventData] = None, + **kwargs: Any + ) -> None: ... + + + class azure.eventhub.TransportType(Enum): + Amqp = 1 + AmqpOverWebsocket = 2 + + +namespace azure.eventhub.aio + + class azure.eventhub.aio.CheckpointStore(ABC): + + @abstractmethod + async def claim_ownership( + self, + ownership_list: Iterable[Dict[str, Any]], + **kwargs: Any + ) -> Iterable[Dict[str, Any]]: ... + + @abstractmethod + async def list_checkpoints( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + **kwargs: Any + ) -> Iterable[Dict[str, Any]]: ... + + @abstractmethod + async def list_ownership( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + **kwargs: Any + ) -> Iterable[Dict[str, Any]]: ... + + @abstractmethod + async def update_checkpoint( + self, + checkpoint: Dict[str, Optional[Union[str, int]]], + **kwargs: Any + ) -> None: ... + + + class azure.eventhub.aio.EventHubConsumerClient(ClientBaseAsync): implements AsyncContextManager + + def __enter__(self) -> None: ... + + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + credential: CredentialTypes, + *, + auth_timeout: Optional[float] = ..., + checkpoint_store: Optional[CheckpointStore] = ..., + connection_verify: Optional[str] = ..., + custom_endpoint_address: Optional[str] = ..., + http_proxy: Union[dict[str, str], dict[str, int], None] = ..., + idle_timeout: Optional[float] = ..., + load_balancing_interval: Optional[float] = ..., + load_balancing_strategy: Union[str, LoadBalancingStrategy] = ..., + logging_enable: Optional[bool] = ..., + partition_ownership_expiration_interval: Optional[float] = ..., + retry_backoff_factor: Optional[float] = ..., + retry_backoff_max: Optional[float] = ..., + retry_mode: str = ..., + retry_total: int = ..., + socket_timeout: Optional[float] = ..., + ssl_context: Union[SSLContext, None] = ..., + transport_type: TransportType = ..., + uamqp_transport: bool = ..., + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + @classmethod + def from_connection_string( + cls, + conn_str: str, + consumer_group: str, + *, + auth_timeout: float = 60, + checkpoint_store: Optional[CheckpointStore] = ..., + connection_verify: Optional[str] = ..., + custom_endpoint_address: Optional[str] = ..., + eventhub_name: Optional[str] = ..., + http_proxy: Optional[Dict[str, Union[str, int]]] = ..., + idle_timeout: Optional[float] = ..., + load_balancing_interval: float = 30, + load_balancing_strategy: Union[str, LoadBalancingStrategy] = LoadBalancingStrategy.GREEDY, + logging_enable: bool = False, + partition_ownership_expiration_interval: Optional[float] = ..., + retry_backoff_factor: float = 0.8, + retry_backoff_max: float = 120, + retry_mode: Literal["exponential", "fixed"] = "exponential", + retry_total: int = 3, + ssl_context: Optional[SSLContext] = ..., + transport_type: TransportType = TransportType.Amqp, + uamqp_transport: bool = False, + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> EventHubConsumerClient: ... + + async def close(self) -> None: ... + + async def get_eventhub_properties(self) -> Dict[str, Any]: ... + + async def get_partition_ids(self) -> List[str]: ... + + async def get_partition_properties(self, partition_id: str) -> Dict[str, Any]: ... + + async def receive( + self, + on_event: Callable[[PartitionContext, Optional[EventData]], Awaitable[None]], + *, + max_wait_time: Optional[float] = ..., + on_error: Optional[Callable[[PartitionContext, Exception], Awaitable[None]]] = ..., + on_partition_close: Optional[Callable[[PartitionContext, CloseReason], Awaitable[None]]] = ..., + on_partition_initialize: Optional[Callable[[PartitionContext], Awaitable[None]]] = ..., + owner_level: Optional[int] = ..., + partition_id: Optional[str] = ..., + prefetch: int = 300, + starting_position: Optional[Union[str, int, datetime, Dict[str, Any]]] = ..., + starting_position_inclusive: Union[bool, Dict[str, bool]] = False, + track_last_enqueued_event_properties: bool = False + ) -> None: ... + + async def receive_batch( + self, + on_event_batch: Callable[[PartitionContext, List[EventData]], Awaitable[None]], + *, + max_batch_size: int = 300, + max_wait_time: Optional[float] = ..., + on_error: Optional[Callable[[PartitionContext, Exception], Awaitable[None]]] = ..., + on_partition_close: Optional[Callable[[PartitionContext, CloseReason], Awaitable[None]]] = ..., + on_partition_initialize: Optional[Callable[[PartitionContext], Awaitable[None]]] = ..., + owner_level: Optional[int] = ..., + partition_id: Optional[str] = ..., + prefetch: int = 300, + starting_position: Optional[Union[str, int, datetime, Dict[str, Any]]] = ..., + starting_position_inclusive: Union[bool, Dict[str, bool]] = False, + track_last_enqueued_event_properties: bool = False + ) -> None: ... + + + class azure.eventhub.aio.EventHubProducerClient(ClientBaseAsync): implements AsyncContextManager + property total_buffered_event_count: Optional[int] # Read-only + + def __enter__(self) -> None: ... + + @overload + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + credential: CredentialTypes, + *, + auth_timeout: Optional[float] = ..., + buffered_mode: Literal[False] = False, + connection_verify: Optional[str] = ..., + custom_endpoint_address: Optional[str] = ..., + http_proxy: Optional[dict] = ..., + idle_timeout: Optional[float] = ..., + logging_enable: Optional[bool] = ..., + max_buffer_length: Optional[int] = ..., + max_wait_time: Optional[float] = ..., + on_error: Optional[Callable[[SendEventTypes, Optional[str], Exception], Awaitable[None]]] = ..., + on_success: Optional[Callable[[SendEventTypes, Optional[str]], Awaitable[None]]] = ..., + retry_backoff_factor: Optional[float] = ..., + retry_backoff_max: Optional[float] = ..., + retry_mode: str = ..., + retry_total: int = ..., + socket_timeout: Optional[float] = ..., + ssl_context: Union[SSLContext, None] = ..., + transport_type: TransportType = ..., + uamqp_transport: bool = ..., + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + @overload + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + credential: CredentialTypes, + *, + auth_timeout: Optional[float] = ..., + buffered_mode: Literal[True], + connection_verify: Optional[str] = ..., + custom_endpoint_address: Optional[str] = ..., + http_proxy: Optional[dict] = ..., + idle_timeout: Optional[float] = ..., + logging_enable: Optional[bool] = ..., + max_buffer_length: int = 1500, + max_wait_time: float = 1, + on_error: Callable[[SendEventTypes, Optional[str], Exception], Awaitable[None]], + on_success: Callable[[SendEventTypes, Optional[str]], Awaitable[None]], + retry_backoff_factor: Optional[float] = ..., + retry_backoff_max: Optional[float] = ..., + retry_mode: str = ..., + retry_total: int = ..., + socket_timeout: Optional[float] = ..., + ssl_context: Union[SSLContext, None] = ..., + transport_type: TransportType = ..., + uamqp_transport: bool = ..., + user_agent: Optional[str] = ..., + **kwargs: Any + ) -> None: ... + + @classmethod + @overload + def from_connection_string( + cls, + conn_str: str, + *, + buffered_mode: Literal[False] = False, + eventhub_name: Optional[str] = ..., + **kwargs: Any + ) -> EventHubProducerClient: ... + + @classmethod + @overload + def from_connection_string( + cls, + conn_str: str, + *, + buffered_mode: Literal[True], + eventhub_name: Optional[str] = ..., + max_buffer_length: int = 1500, + max_wait_time: float = 1, + on_error: Callable[[SendEventTypes, Optional[str], Exception], Awaitable[None]], + on_success: Callable[[SendEventTypes, Optional[str]], Awaitable[None]], + **kwargs: Any + ) -> EventHubProducerClient: ... + + async def close( + self, + *, + flush: bool = True, + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + async def create_batch( + self, + *, + max_size_in_bytes: Optional[int] = ..., + partition_id: Optional[str] = ..., + partition_key: Optional[str] = ... + ) -> EventDataBatch: ... + + async def flush( + self, + *, + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + def get_buffered_event_count(self, partition_id: str) -> Optional[int]: ... + + async def get_eventhub_properties(self) -> Dict[str, Any]: ... + + async def get_partition_ids(self) -> List[str]: ... + + async def get_partition_properties(self, partition_id: str) -> Dict[str, Any]: ... + + async def send_batch( + self, + event_data_batch: Union[EventDataBatch, SendEventTypes], + *, + partition_id: Optional[str] = ..., + partition_key: Optional[str] = ..., + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + async def send_event( + self, + event_data: Union[EventData, AmqpAnnotatedMessage], + *, + partition_id: Optional[str] = ..., + partition_key: Optional[str] = ..., + timeout: Optional[float] = ..., + **kwargs: Any + ) -> None: ... + + + class azure.eventhub.aio.EventHubSharedKeyCredential: + + def __init__( + self, + policy: str, + key: str + ): ... + + async def get_token( + self, + *scopes: str, + **kwargs: Any + ) -> AccessToken: ... + + + class azure.eventhub.aio.PartitionContext: + property last_enqueued_event_properties: Optional[Dict[str, Any]] # Read-only + + def __init__( + self, + fully_qualified_namespace: str, + eventhub_name: str, + consumer_group: str, + partition_id: str, + checkpoint_store: Optional[CheckpointStore] = None + ) -> None: ... + + async def update_checkpoint( + self, + event: Optional[EventData] = None, + **kwargs: Any + ) -> None: ... + + +namespace azure.eventhub.amqp + + class azure.eventhub.amqp.AmqpAnnotatedMessage: + property annotations: Optional[Dict[Union[str, bytes], Any]] + property application_properties: Optional[Dict[Union[str, bytes], Any]] + property body: Any # Read-only + property body_type: AmqpMessageBodyType # Read-only + property delivery_annotations: Optional[Dict[Union[str, bytes], Any]] + property footer: Optional[Dict[Any, Any]] + property header: Optional[AmqpMessageHeader] + property properties: Optional[AmqpMessageProperties] + + def __init__( + self, + *, + annotations: Optional[Dict] = ..., + application_properties: Optional[Dict] = ..., + data_body: Union[str, bytes, List[Union[str, bytes]]] = ..., + delivery_annotations: Optional[Dict] = ..., + footer: Optional[Dict] = ..., + header: Optional[AmqpMessageHeader] = ..., + properties: Optional[AmqpMessageProperties] = ..., + sequence_body: List[Any] = ..., + value_body: Any = ..., + **kwargs: Any + ) -> None: ... + + def __repr__(self) -> str: ... + + def __str__(self) -> str: ... + + + class azure.eventhub.amqp.AmqpMessageBodyType(str, Enum, metaclass=CaseInsensitiveEnumMeta): + DATA = "data" + SEQUENCE = "sequence" + VALUE = "value" + + + class azure.eventhub.amqp.AmqpMessageHeader(DictMixin): + delivery_count: Optional[int] + durable: Optional[bool] + first_acquirer: Optional[bool] + priority: Optional[int] + time_to_live: Optional[int] + + def __contains__(self, key: str) -> bool: ... + + def __delitem__(self, key: str) -> None: ... + + def __eq__(self, other: Any) -> bool: ... + + def __getitem__(self, key: str) -> Any: ... + + def __init__( + self, + *, + delivery_count: Optional[int] = ..., + durable: Optional[bool] = ..., + first_acquirer: Optional[bool] = ..., + priority: Optional[int] = ..., + time_to_live: Optional[int] = ..., + **kwargs + ): ... + + def __len__(self) -> int: ... + + def __ne__(self, other: Any) -> bool: ... + + def __repr__(self) -> str: ... + + def __setitem__( + self, + key: str, + item: Any + ) -> None: ... + + def __str__(self) -> str: ... + + def get( + self, + key: str, + default: Optional[Any] = None + ) -> Any: ... + + def has_key(self, k: str) -> bool: ... + + def items(self) -> List[Tuple[str, Any]]: ... + + def keys(self) -> List[str]: ... + + def update( + self, + *args: Any, + **kwargs: Any + ) -> None: ... + + def values(self) -> List[Any]: ... + + + class azure.eventhub.amqp.AmqpMessageProperties(DictMixin): + absolute_expiry_time: Optional[int] + content_encoding: Optional[bytes] + content_type: Optional[bytes] + correlation_id: Optional[bytes] + creation_time: Optional[int] + group_id: Optional[bytes] + group_sequence: Optional[int] + message_id: Optional[bytes] + reply_to: Optional[bytes] + reply_to_group_id: Optional[bytes] + subject: Optional[bytes] + to: Optional[bytes] + user_id: Optional[bytes] + + def __contains__(self, key: str) -> bool: ... + + def __delitem__(self, key: str) -> None: ... + + def __eq__(self, other: Any) -> bool: ... + + def __getitem__(self, key: str) -> Any: ... + + def __init__( + self, + *, + absolute_expiry_time: Optional[int] = ..., + content_encoding: Optional[Union[str, bytes]] = ..., + content_type: Optional[Union[str, bytes]] = ..., + correlation_id: Optional[Union[str, bytes]] = ..., + creation_time: Optional[int] = ..., + group_id: Optional[Union[str, bytes]] = ..., + group_sequence: Optional[int] = ..., + message_id: Optional[Union[str, bytes, uuid.UUID]] = ..., + reply_to: Optional[Union[str, bytes]] = ..., + reply_to_group_id: Optional[Union[str, bytes]] = ..., + subject: Optional[Union[str, bytes]] = ..., + to: Optional[Union[str, bytes]] = ..., + user_id: Optional[Union[str, bytes]] = ..., + **kwargs + ): ... + + def __len__(self) -> int: ... + + def __ne__(self, other: Any) -> bool: ... + + def __repr__(self) -> str: ... + + def __setitem__( + self, + key: str, + item: Any + ) -> None: ... + + def __str__(self) -> str: ... + + def get( + self, + key: str, + default: Optional[Any] = None + ) -> Any: ... + + def has_key(self, k: str) -> bool: ... + + def items(self) -> List[Tuple[str, Any]]: ... + + def keys(self) -> List[str]: ... + + def update( + self, + *args: Any, + **kwargs: Any + ) -> None: ... + + def values(self) -> List[Any]: ... + + +namespace azure.eventhub.exceptions + + class azure.eventhub.exceptions.AuthenticationError(ConnectError): + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.ClientClosedError(EventHubError): + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.ConnectError(EventHubError): + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.ConnectionLostError(EventHubError): + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.EventDataError(EventHubError): + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.EventDataSendError(EventHubError): + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.EventHubError(Exception): + details: list[str] + error: str + message: str + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.OperationTimeoutError(EventHubError): + + def __init__( + self, + message: str, + details: Optional[List[str]] = None + ) -> None: ... + + + class azure.eventhub.exceptions.OwnershipLostError(Exception): + + +``` \ No newline at end of file diff --git a/sdk/eventhub/azure-eventhub/api.metadata.yml b/sdk/eventhub/azure-eventhub/api.metadata.yml new file mode 100644 index 000000000000..e3d540096229 --- /dev/null +++ b/sdk/eventhub/azure-eventhub/api.metadata.yml @@ -0,0 +1,3 @@ +apiMdSha256: c54b169f6a986c1d8193fc3420e451cc005b078a2c347782b9ef349b66633663 +parserVersion: 0.3.28 +pythonVersion: 3.12.12 diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_platform.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_platform.py index 9289fb1e34f3..3a836317cc59 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_platform.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_platform.py @@ -31,20 +31,19 @@ def _versionatom(s): return int(match.groups()[0]) if match else 0 -# available socket options for TCP level +# TCP socket options that are safe to set on this platform. +# Used by _transport.py to filter DEFAULT_SOCKET_SETTINGS — any option +# not in this set is skipped during socket configuration. +# +# Not all platforms support all TCP options via setsockopt(). For example, +# Windows only supports TCP_NODELAY, and macOS/BSD do not support +# TCP_USER_TIMEOUT. This set is trimmed per-platform below. KNOWN_TCP_OPTS = { - "TCP_CORK", - "TCP_DEFER_ACCEPT", "TCP_KEEPCNT", "TCP_KEEPIDLE", "TCP_KEEPINTVL", - "TCP_LINGER2", - "TCP_MAXSEG", "TCP_NODELAY", - "TCP_QUICKACK", - "TCP_SYNCNT", "TCP_USER_TIMEOUT", - "TCP_WINDOW_CLAMP", } LINUX_VERSION = None @@ -64,23 +63,13 @@ def _versionatom(s): elif "bsd" in sys.platform: KNOWN_TCP_OPTS.remove("TCP_USER_TIMEOUT") -# According to MSDN Windows platforms support getsockopt(TCP_MAXSSEG) but not -# setsockopt(TCP_MAXSEG) on IPPROTO_TCP sockets. elif sys.platform.startswith("win"): KNOWN_TCP_OPTS = {"TCP_NODELAY"} elif sys.platform.startswith("cygwin"): KNOWN_TCP_OPTS = {"TCP_NODELAY"} -# illumos does not allow to set the TCP_MAXSEG socket option, -# even if the Oracle documentation says otherwise. -elif sys.platform.startswith("sunos"): - KNOWN_TCP_OPTS.remove("TCP_MAXSEG") - -# aix does not allow to set the TCP_MAXSEG -# or the TCP_USER_TIMEOUT socket options. elif sys.platform.startswith("aix"): - KNOWN_TCP_OPTS.remove("TCP_MAXSEG") KNOWN_TCP_OPTS.remove("TCP_USER_TIMEOUT") pack = struct.pack diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_transport.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_transport.py index bfb1f9561066..ac5faa8a89ec 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_transport.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/_transport.py @@ -99,6 +99,9 @@ def set_cloexec(fd, cloexec): # noqa # pylint: disable=inconsistent-return-stat # Match things like: [fe80::1]:5432, from RFC 2732 IPV6_LITERAL = re.compile(r"\[([\.0-9a-f:]+)\](?::(\d+))?") +# TCP socket options applied to every new connection. +# Any option added here must also be present in KNOWN_TCP_OPTS +# (in _platform.py) for the platforms where it should be set. DEFAULT_SOCKET_SETTINGS = { "TCP_NODELAY": 1, "TCP_USER_TIMEOUT": 1000, @@ -213,11 +216,11 @@ def connect(self): @contextmanager def block(self): - bocking_timeout = None + blocking_timeout = None sock = self.sock prev = sock.gettimeout() - if prev != bocking_timeout: - sock.settimeout(bocking_timeout) + if prev != blocking_timeout: + sock.settimeout(blocking_timeout) try: yield self.sock except SSLError as exc: @@ -233,16 +236,16 @@ def block(self): raise socket.timeout() raise finally: - if bocking_timeout != prev: + if blocking_timeout != prev: sock.settimeout(prev) @contextmanager def non_blocking(self): - non_bocking_timeout = 0.0 + non_blocking_timeout = 0.0 sock = self.sock prev = sock.gettimeout() - if prev != non_bocking_timeout: - sock.settimeout(non_bocking_timeout) + if prev != non_blocking_timeout: + sock.settimeout(non_blocking_timeout) try: yield self.sock except SSLError as exc: @@ -258,7 +261,7 @@ def non_blocking(self): raise socket.timeout() raise finally: - if non_bocking_timeout != prev: + if non_blocking_timeout != prev: sock.settimeout(prev) @@ -280,28 +283,25 @@ def _init_socket(self, socket_settings, socket_timeout): self._setup_transport() self.sock.settimeout(socket_timeout) # set socket to blocking with timeout - def _get_tcp_socket_defaults(self, sock): + def _get_tcp_socket_defaults(self): tcp_opts = {} - for opt in KNOWN_TCP_OPTS: + for opt, val in DEFAULT_SOCKET_SETTINGS.items(): + if opt not in KNOWN_TCP_OPTS: + continue enum = None if opt == "TCP_USER_TIMEOUT": try: from socket import TCP_USER_TIMEOUT as enum except ImportError: - # should be in Python 3.6+ on Linux. enum = 18 elif hasattr(socket, opt): enum = getattr(socket, opt) - if enum: - if opt in DEFAULT_SOCKET_SETTINGS: - tcp_opts[enum] = DEFAULT_SOCKET_SETTINGS[opt] - elif hasattr(socket, opt): - tcp_opts[enum] = sock.getsockopt(SOL_TCP, getattr(socket, opt)) + tcp_opts[enum] = val return tcp_opts def _set_socket_options(self, socket_settings): - tcp_opts = self._get_tcp_socket_defaults(self.sock) + tcp_opts = self._get_tcp_socket_defaults() if socket_settings: tcp_opts.update(socket_settings) for opt, val in tcp_opts.items(): diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_client_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_client_async.py index 18085bf2bdb0..087a0c48b243 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_client_async.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_client_async.py @@ -4,7 +4,7 @@ # license information. # -------------------------------------------------------------------------- # TODO: Check types of kwargs (issue exists for this) -import asyncio # pylint:disable=do-not-import-asyncio +import asyncio # pylint:disable=do-not-import-asyncio import logging import time import queue diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py index 0478f41324bf..6b96e090c26b 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_transport_async.py @@ -32,7 +32,7 @@ # THE POSSIBILITY OF SUCH DAMAGE. # ------------------------------------------------------------------------- -import asyncio # pylint:disable=do-not-import-asyncio +import asyncio # pylint:disable=do-not-import-asyncio import errno import socket import ssl @@ -273,32 +273,29 @@ async def connect(self): self.connected = False raise - def _get_tcp_socket_defaults(self, sock): + def _get_tcp_socket_defaults(self): tcp_opts = {} - for opt in KNOWN_TCP_OPTS: + for opt, val in DEFAULT_SOCKET_SETTINGS.items(): + if opt not in KNOWN_TCP_OPTS: + continue enum = None if opt == "TCP_USER_TIMEOUT": try: from socket import TCP_USER_TIMEOUT as enum except ImportError: - # should be in Python 3.6+ on Linux. enum = 18 elif hasattr(socket, opt): enum = getattr(socket, opt) - if enum: - if opt in DEFAULT_SOCKET_SETTINGS: - tcp_opts[enum] = DEFAULT_SOCKET_SETTINGS[opt] - elif hasattr(socket, opt): - tcp_opts[enum] = sock.getsockopt(SOL_TCP, getattr(socket, opt)) + tcp_opts[enum] = val return tcp_opts def _set_socket_options(self, sock, socket_settings): + tcp_opts = self._get_tcp_socket_defaults() if socket_settings: - tcp_opts = self._get_tcp_socket_defaults(sock) tcp_opts.update(socket_settings) - for opt, val in tcp_opts.items(): - sock.setsockopt(SOL_TCP, opt, val) + for opt, val in tcp_opts.items(): + sock.setsockopt(SOL_TCP, opt, val) async def _read( self, diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_version.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_version.py index 5bb5c30266da..5d660bbf8ea3 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_version.py +++ b/sdk/eventhub/azure-eventhub/azure/eventhub/_version.py @@ -3,4 +3,4 @@ # Licensed under the MIT License. # ------------------------------------ -VERSION = "5.15.1" +VERSION = "5.15.2" diff --git a/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_transport.py b/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_transport.py new file mode 100644 index 000000000000..651f9e99256a --- /dev/null +++ b/sdk/eventhub/azure-eventhub/tests/pyamqp_tests/unittest/test_transport.py @@ -0,0 +1,55 @@ +import socket +from unittest.mock import MagicMock, patch + +from azure.eventhub._pyamqp._transport import ( + DEFAULT_SOCKET_SETTINGS, + _AbstractTransport, +) +from azure.eventhub._pyamqp._platform import KNOWN_TCP_OPTS, SOL_TCP +from azure.eventhub._pyamqp.aio._transport_async import AsyncTransport + + +def test_get_defaults_returns_only_configured_settings(): + t = _AbstractTransport.__new__(_AbstractTransport) + result = t._get_tcp_socket_defaults() + for name, expected_val in DEFAULT_SOCKET_SETTINGS.items(): + if name not in KNOWN_TCP_OPTS: + continue + enum = getattr(socket, name, 18 if name == "TCP_USER_TIMEOUT" else None) + if enum is not None: + assert enum in result + assert result[enum] == expected_val + # no extra keys beyond what DEFAULT_SOCKET_SETTINGS defines + assert len(result) <= len(DEFAULT_SOCKET_SETTINGS) + + +def test_get_defaults_skips_platform_unsupported_opts(): + # Simulate a platform (e.g. Windows) that only supports TCP_NODELAY + t = _AbstractTransport.__new__(_AbstractTransport) + with patch( + "azure.eventhub._pyamqp._transport.KNOWN_TCP_OPTS", + {"TCP_NODELAY"}, + ): + result = t._get_tcp_socket_defaults() + assert len(result) == 1 + assert result[socket.TCP_NODELAY] == DEFAULT_SOCKET_SETTINGS["TCP_NODELAY"] + + +def test_sync_set_socket_options_applies_defaults(): + t = _AbstractTransport.__new__(_AbstractTransport) + t.sock = MagicMock(spec=socket.socket) + t._set_socket_options(None) + defaults = t._get_tcp_socket_defaults() + assert t.sock.setsockopt.call_count == len(defaults) + for enum, val in defaults.items(): + t.sock.setsockopt.assert_any_call(SOL_TCP, enum, val) + + +def test_async_set_socket_options_applies_defaults(): + t = AsyncTransport.__new__(AsyncTransport) + mock_sock = MagicMock(spec=socket.socket) + t._set_socket_options(mock_sock, None) + defaults = t._get_tcp_socket_defaults() + assert mock_sock.setsockopt.call_count == len(defaults) + for enum, val in defaults.items(): + mock_sock.setsockopt.assert_any_call(SOL_TCP, enum, val) diff --git a/sdk/servicebus/azure-servicebus/CHANGELOG.md b/sdk/servicebus/azure-servicebus/CHANGELOG.md index 0d105a35dc45..1fdc4a3f450a 100644 --- a/sdk/servicebus/azure-servicebus/CHANGELOG.md +++ b/sdk/servicebus/azure-servicebus/CHANGELOG.md @@ -13,6 +13,7 @@ - Fixed a bug where sending a batched or multi-message payload with `uamqp_transport=True` raised `TypeError: 'BatchMessage' object is not subscriptable` (and a masked `AttributeError` on the list path) when the first message carried a `message_id`, `session_id`, or `partition_key`. The batch envelope properties are now set through the transport-appropriate code path. (regression from [#42598](https://github.com/Azure/azure-sdk-for-python/pull/42598)) - Fixed a bug where the async receiver factory methods on `azure.servicebus.aio.ServiceBusClient` (`get_queue_receiver`, `get_subscription_receiver`) and the async `ServiceBusReceiver` annotated the `auto_lock_renewer` keyword with the synchronous `AutoLockRenewer`, causing static type checkers to reject the documented `azure.servicebus.aio.AutoLockRenewer` usage. The annotation now references the async `AutoLockRenewer`, matching the docstrings and runtime behavior. ([#47948](https://github.com/Azure/azure-sdk-for-python/issues/47948)) - Fixed a bug where closing a `PEEK_LOCK` receiver did not release messages that had been prefetched into the client buffer or were still in flight, so they remained locked at the broker until lock expiry — delaying their redelivery and inflating their delivery count. On close, a non-session `PEEK_LOCK` receiver now drains the link (stopping the broker and flushing in-flight transfers) and releases the buffered messages (`released` disposition), so the broker can redeliver them immediately without incrementing the delivery count. ([#42917](https://github.com/Azure/azure-sdk-for-python/issues/42917)) +- Fixed a bug where the async pure-Python AMQP transport failed to connect with `[Errno 22] Invalid argument` (`amqp:socket-error`) inside containerized/virtualized environments such as Docker Desktop on macOS. The transport no longer reads back and re-applies platform-negotiated TCP options (e.g. `TCP_MAXSEG`) that some platforms reject via `setsockopt`. ([#45394](https://github.com/Azure/azure-sdk-for-python/issues/45394)) ### Other Changes diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_platform.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_platform.py index 9289fb1e34f3..3a836317cc59 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_platform.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_platform.py @@ -31,20 +31,19 @@ def _versionatom(s): return int(match.groups()[0]) if match else 0 -# available socket options for TCP level +# TCP socket options that are safe to set on this platform. +# Used by _transport.py to filter DEFAULT_SOCKET_SETTINGS — any option +# not in this set is skipped during socket configuration. +# +# Not all platforms support all TCP options via setsockopt(). For example, +# Windows only supports TCP_NODELAY, and macOS/BSD do not support +# TCP_USER_TIMEOUT. This set is trimmed per-platform below. KNOWN_TCP_OPTS = { - "TCP_CORK", - "TCP_DEFER_ACCEPT", "TCP_KEEPCNT", "TCP_KEEPIDLE", "TCP_KEEPINTVL", - "TCP_LINGER2", - "TCP_MAXSEG", "TCP_NODELAY", - "TCP_QUICKACK", - "TCP_SYNCNT", "TCP_USER_TIMEOUT", - "TCP_WINDOW_CLAMP", } LINUX_VERSION = None @@ -64,23 +63,13 @@ def _versionatom(s): elif "bsd" in sys.platform: KNOWN_TCP_OPTS.remove("TCP_USER_TIMEOUT") -# According to MSDN Windows platforms support getsockopt(TCP_MAXSSEG) but not -# setsockopt(TCP_MAXSEG) on IPPROTO_TCP sockets. elif sys.platform.startswith("win"): KNOWN_TCP_OPTS = {"TCP_NODELAY"} elif sys.platform.startswith("cygwin"): KNOWN_TCP_OPTS = {"TCP_NODELAY"} -# illumos does not allow to set the TCP_MAXSEG socket option, -# even if the Oracle documentation says otherwise. -elif sys.platform.startswith("sunos"): - KNOWN_TCP_OPTS.remove("TCP_MAXSEG") - -# aix does not allow to set the TCP_MAXSEG -# or the TCP_USER_TIMEOUT socket options. elif sys.platform.startswith("aix"): - KNOWN_TCP_OPTS.remove("TCP_MAXSEG") KNOWN_TCP_OPTS.remove("TCP_USER_TIMEOUT") pack = struct.pack diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_transport.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_transport.py index bfb1f9561066..ac5faa8a89ec 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_transport.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/_transport.py @@ -99,6 +99,9 @@ def set_cloexec(fd, cloexec): # noqa # pylint: disable=inconsistent-return-stat # Match things like: [fe80::1]:5432, from RFC 2732 IPV6_LITERAL = re.compile(r"\[([\.0-9a-f:]+)\](?::(\d+))?") +# TCP socket options applied to every new connection. +# Any option added here must also be present in KNOWN_TCP_OPTS +# (in _platform.py) for the platforms where it should be set. DEFAULT_SOCKET_SETTINGS = { "TCP_NODELAY": 1, "TCP_USER_TIMEOUT": 1000, @@ -213,11 +216,11 @@ def connect(self): @contextmanager def block(self): - bocking_timeout = None + blocking_timeout = None sock = self.sock prev = sock.gettimeout() - if prev != bocking_timeout: - sock.settimeout(bocking_timeout) + if prev != blocking_timeout: + sock.settimeout(blocking_timeout) try: yield self.sock except SSLError as exc: @@ -233,16 +236,16 @@ def block(self): raise socket.timeout() raise finally: - if bocking_timeout != prev: + if blocking_timeout != prev: sock.settimeout(prev) @contextmanager def non_blocking(self): - non_bocking_timeout = 0.0 + non_blocking_timeout = 0.0 sock = self.sock prev = sock.gettimeout() - if prev != non_bocking_timeout: - sock.settimeout(non_bocking_timeout) + if prev != non_blocking_timeout: + sock.settimeout(non_blocking_timeout) try: yield self.sock except SSLError as exc: @@ -258,7 +261,7 @@ def non_blocking(self): raise socket.timeout() raise finally: - if non_bocking_timeout != prev: + if non_blocking_timeout != prev: sock.settimeout(prev) @@ -280,28 +283,25 @@ def _init_socket(self, socket_settings, socket_timeout): self._setup_transport() self.sock.settimeout(socket_timeout) # set socket to blocking with timeout - def _get_tcp_socket_defaults(self, sock): + def _get_tcp_socket_defaults(self): tcp_opts = {} - for opt in KNOWN_TCP_OPTS: + for opt, val in DEFAULT_SOCKET_SETTINGS.items(): + if opt not in KNOWN_TCP_OPTS: + continue enum = None if opt == "TCP_USER_TIMEOUT": try: from socket import TCP_USER_TIMEOUT as enum except ImportError: - # should be in Python 3.6+ on Linux. enum = 18 elif hasattr(socket, opt): enum = getattr(socket, opt) - if enum: - if opt in DEFAULT_SOCKET_SETTINGS: - tcp_opts[enum] = DEFAULT_SOCKET_SETTINGS[opt] - elif hasattr(socket, opt): - tcp_opts[enum] = sock.getsockopt(SOL_TCP, getattr(socket, opt)) + tcp_opts[enum] = val return tcp_opts def _set_socket_options(self, socket_settings): - tcp_opts = self._get_tcp_socket_defaults(self.sock) + tcp_opts = self._get_tcp_socket_defaults() if socket_settings: tcp_opts.update(socket_settings) for opt, val in tcp_opts.items(): diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/aio/_transport_async.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/aio/_transport_async.py index 97a49c214506..6b96e090c26b 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/aio/_transport_async.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/aio/_transport_async.py @@ -32,7 +32,7 @@ # THE POSSIBILITY OF SUCH DAMAGE. # ------------------------------------------------------------------------- -import asyncio # pylint:disable=do-not-import-asyncio +import asyncio # pylint:disable=do-not-import-asyncio import errno import socket import ssl @@ -273,28 +273,25 @@ async def connect(self): self.connected = False raise - def _get_tcp_socket_defaults(self, sock): + def _get_tcp_socket_defaults(self): tcp_opts = {} - for opt in KNOWN_TCP_OPTS: + for opt, val in DEFAULT_SOCKET_SETTINGS.items(): + if opt not in KNOWN_TCP_OPTS: + continue enum = None if opt == "TCP_USER_TIMEOUT": try: from socket import TCP_USER_TIMEOUT as enum except ImportError: - # should be in Python 3.6+ on Linux. enum = 18 elif hasattr(socket, opt): enum = getattr(socket, opt) - if enum: - if opt in DEFAULT_SOCKET_SETTINGS: - tcp_opts[enum] = DEFAULT_SOCKET_SETTINGS[opt] - elif hasattr(socket, opt): - tcp_opts[enum] = sock.getsockopt(SOL_TCP, getattr(socket, opt)) + tcp_opts[enum] = val return tcp_opts def _set_socket_options(self, sock, socket_settings): - tcp_opts = self._get_tcp_socket_defaults(sock) + tcp_opts = self._get_tcp_socket_defaults() if socket_settings: tcp_opts.update(socket_settings) for opt, val in tcp_opts.items(): diff --git a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/utils.py b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/utils.py index eec4e3495cf1..2edb14e27cb3 100644 --- a/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/utils.py +++ b/sdk/servicebus/azure-servicebus/azure/servicebus/_pyamqp/utils.py @@ -23,6 +23,7 @@ # This is the lowest value that can be represented by an AMQP timestamp. CE_ZERO_SECONDS: int = -62_135_596_800 + def utc_from_timestamp(timestamp: float) -> datetime.datetime: """ :param float timestamp: Timestamp in seconds to be converted to datetime. diff --git a/sdk/servicebus/azure-servicebus/tests/unittests/test_transport.py b/sdk/servicebus/azure-servicebus/tests/unittests/test_transport.py new file mode 100644 index 000000000000..ec2fd17a09a2 --- /dev/null +++ b/sdk/servicebus/azure-servicebus/tests/unittests/test_transport.py @@ -0,0 +1,55 @@ +import socket +from unittest.mock import MagicMock, patch + +from azure.servicebus._pyamqp._transport import ( + DEFAULT_SOCKET_SETTINGS, + _AbstractTransport, +) +from azure.servicebus._pyamqp._platform import KNOWN_TCP_OPTS, SOL_TCP +from azure.servicebus._pyamqp.aio._transport_async import AsyncTransport + + +def test_get_defaults_returns_only_configured_settings(): + t = _AbstractTransport.__new__(_AbstractTransport) + result = t._get_tcp_socket_defaults() + for name, expected_val in DEFAULT_SOCKET_SETTINGS.items(): + if name not in KNOWN_TCP_OPTS: + continue + enum = getattr(socket, name, 18 if name == "TCP_USER_TIMEOUT" else None) + if enum is not None: + assert enum in result + assert result[enum] == expected_val + # no extra keys beyond what DEFAULT_SOCKET_SETTINGS defines + assert len(result) <= len(DEFAULT_SOCKET_SETTINGS) + + +def test_get_defaults_skips_platform_unsupported_opts(): + # Simulate a platform (e.g. Windows) that only supports TCP_NODELAY + t = _AbstractTransport.__new__(_AbstractTransport) + with patch( + "azure.servicebus._pyamqp._transport.KNOWN_TCP_OPTS", + {"TCP_NODELAY"}, + ): + result = t._get_tcp_socket_defaults() + assert len(result) == 1 + assert result[socket.TCP_NODELAY] == DEFAULT_SOCKET_SETTINGS["TCP_NODELAY"] + + +def test_sync_set_socket_options_applies_defaults(): + t = _AbstractTransport.__new__(_AbstractTransport) + t.sock = MagicMock(spec=socket.socket) + t._set_socket_options(None) + defaults = t._get_tcp_socket_defaults() + assert t.sock.setsockopt.call_count == len(defaults) + for enum, val in defaults.items(): + t.sock.setsockopt.assert_any_call(SOL_TCP, enum, val) + + +def test_async_set_socket_options_applies_defaults(): + t = AsyncTransport.__new__(AsyncTransport) + mock_sock = MagicMock(spec=socket.socket) + t._set_socket_options(mock_sock, None) + defaults = t._get_tcp_socket_defaults() + assert mock_sock.setsockopt.call_count == len(defaults) + for enum, val in defaults.items(): + mock_sock.setsockopt.assert_any_call(SOL_TCP, enum, val)