| # Licensed to the Apache Software Foundation (ASF) under one | |
| # or more contributor license agreements. See the NOTICE file | |
| # distributed with this work for additional information | |
| # regarding copyright ownership. The ASF licenses this file | |
| # to you under the Apache License, Version 2.0 (the | |
| # "License"); you may not use this file except in compliance | |
| # with the License. You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, | |
| # software distributed under the License is distributed on an | |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | |
| # KIND, either express or implied. See the License for the | |
| # specific language governing permissions and limitations | |
| # under the License. | |
| # | |
| import importlib | |
| from types import SimpleNamespace | |
| import pytest | |
| from thrift.transport import TTransport | |
| from iotdb.thrift.common.ttypes import TSStatus | |
| from iotdb.utils.exception import StatementExecutionException | |
| class FakeSocket: | |
| def setTimeout(self, timeout): | |
| self.timeout = timeout | |
| class FakeTransport: | |
| def __init__(self, socket): | |
| self.socket = socket | |
| self.opened = False | |
| def isOpen(self): | |
| return self.opened | |
| def open(self): | |
| self.opened = True | |
| def close(self): | |
| self.opened = False | |
| @pytest.mark.parametrize( | |
| "status_code,status_message", | |
| [ | |
| (801, "Authentication failed."), | |
| (822, "Account is blocked due to consecutive failed logins."), | |
| ], | |
| ) | |
| def test_session_does_not_retry_iotdb_status_error( | |
| monkeypatch, status_code, status_message | |
| ): | |
| session_module = importlib.import_module("iotdb.Session") | |
| class RejectingClient: | |
| open_session_calls = 0 | |
| def __init__(self, protocol): | |
| pass | |
| def openSession(self, request): | |
| RejectingClient.open_session_calls += 1 | |
| return SimpleNamespace( | |
| status=TSStatus(code=status_code, message=status_message) | |
| ) | |
| monkeypatch.setattr( | |
| session_module.TSocket, "TSocket", lambda host, port: FakeSocket() | |
| ) | |
| monkeypatch.setattr(session_module.TTransport, "TFramedTransport", FakeTransport) | |
| monkeypatch.setattr(session_module, "Client", RejectingClient) | |
| session = session_module.Session.init_from_node_urls( | |
| ["127.0.0.1:6667"], user="test", password="wrong" | |
| ) | |
| with pytest.raises(StatementExecutionException) as exc_info: | |
| session.open() | |
| assert str(exc_info.value) == f"{status_code}: {status_message}" | |
| assert RejectingClient.open_session_calls == 1 | |
| def test_session_retries_transport_error(monkeypatch): | |
| session_module = importlib.import_module("iotdb.Session") | |
| class FailingClient: | |
| open_session_calls = 0 | |
| def __init__(self, protocol): | |
| pass | |
| def openSession(self, request): | |
| FailingClient.open_session_calls += 1 | |
| raise TTransport.TTransportException(message="Network is unavailable.") | |
| monkeypatch.setattr( | |
| session_module.TSocket, "TSocket", lambda host, port: FakeSocket() | |
| ) | |
| monkeypatch.setattr(session_module.TTransport, "TFramedTransport", FakeTransport) | |
| monkeypatch.setattr(session_module, "Client", FailingClient) | |
| session = session_module.Session.init_from_node_urls( | |
| ["127.0.0.1:6667"], user="test", password="wrong" | |
| ) | |
| with pytest.raises(session_module.IoTDBConnectionException): | |
| session.open() | |
| assert FailingClient.open_session_calls == 1 + session.RETRY_NUM |