diff --git a/pytest_tests/helpers/ssh_helper.py b/pytest_tests/helpers/ssh_helper.py index 05ea99a..31316c3 100644 --- a/pytest_tests/helpers/ssh_helper.py +++ b/pytest_tests/helpers/ssh_helper.py @@ -65,7 +65,7 @@ class HostClient: TIMEOUT_RESTORE_CONNECTION = 10, 24 - def __init__(self, ip: str, login: str, password: Optional[str], + def __init__(self, ip: str, login: Optional[str] = None, password: Optional[str] = None, private_key_path: Optional[str] = None, init_ssh_client=True) -> None: self.ip = ip self.login = login diff --git a/pytest_tests/testsuites/failovers/test_failover_storage.py b/pytest_tests/testsuites/failovers/test_failover_storage.py index 4a782d5..1620993 100644 --- a/pytest_tests/testsuites/failovers/test_failover_storage.py +++ b/pytest_tests/testsuites/failovers/test_failover_storage.py @@ -4,7 +4,6 @@ from time import sleep import allure import pytest -from common import STORAGE_NODE_PWD, STORAGE_NODE_USER from python_keywords.container import create_container from python_keywords.neofs_verbs import get_object, put_object from python_keywords.utility_keywords import generate_file, get_file_hash @@ -13,13 +12,26 @@ from ssh_helper import HostClient, HostIsNotAvailable from storage_policy import get_nodes_with_object from wellknown_acl import PUBLIC_ACL +SSH_PK_PATH = f'{os.getcwd()}/configuration/id_rsa' logger = logging.getLogger('NeoLogger') stopped_hosts = [] +@pytest.fixture(scope='session') +def free_storage_check(): + if os.getenv('FREE_STORAGE', default='False').lower() not in ('true', '1'): + pytest.skip('Test work only on SberCloud infrastructure') + yield + + @pytest.fixture(scope='session') def sbercloud_client(): - yield SberCloud(f'{os.getcwd()}/configuration/sbercloud.yaml') + with allure.step('Connect to SberCloud'): + try: + yield SberCloud(f'{os.getcwd()}/configuration/sbercloud.yaml') + except Exception: + pytest.fail('SberCloud infrastructure not available') + yield None @pytest.fixture(scope='session', autouse=True) @@ -30,7 +42,8 @@ def return_all_storage_nodes_fixture(sbercloud_client): @allure.title('Hard reboot host via magic SysRq option') def panic_reboot_host(ip: str = None): - ssh = HostClient(ip=ip, login=STORAGE_NODE_USER, password=STORAGE_NODE_PWD) + ssh = HostClient(ip=ip, init_ssh_client=False) + ssh.pk = SSH_PK_PATH ssh.create_connection(attempts=1) ssh.exec('echo 1 > /proc/sys/kernel/sysrq') with pytest.raises(HostIsNotAvailable): @@ -56,7 +69,8 @@ def wait_object_replication(wallet, cid, oid, expected_copies: int) -> [str]: @allure.title('Lost and return nodes') @pytest.mark.parametrize('hard_reboot', [True, False]) -def test_lost_storage_node(prepare_wallet_and_deposit, sbercloud_client: SberCloud, hard_reboot: bool): +def test_lost_storage_node(prepare_wallet_and_deposit, sbercloud_client: SberCloud, + free_storage_check, hard_reboot: bool): wallet = prepare_wallet_and_deposit placement_rule = 'REP 2 IN X CBF 2 SELECT 2 FROM * AS X' source_file_path = generate_file() @@ -85,7 +99,7 @@ def test_lost_storage_node(prepare_wallet_and_deposit, sbercloud_client: SberClo @allure.title('Panic storage node(s)') @pytest.mark.parametrize('sequence', [True, False]) -def test_panic_storage_node(prepare_wallet_and_deposit, sequence: bool): +def test_panic_storage_node(prepare_wallet_and_deposit, free_storage_check, sequence: bool): wallet = prepare_wallet_and_deposit placement_rule = 'REP 2 IN X CBF 2 SELECT 2 FROM * AS X' source_file_path = generate_file()