diff --git a/pytest_tests/tests/s3/test_s3_nexus.py b/pytest_tests/tests/s3/test_s3_nexus.py new file mode 100644 index 000000000..7ff1682d4 --- /dev/null +++ b/pytest_tests/tests/s3/test_s3_nexus.py @@ -0,0 +1,127 @@ +import json +import logging +import sys +import time + +import allure +import pytest +import requests +from neofs_testlib.env.env import NeoFSEnv +from neofs_testlib.shell import CommandOptions +from s3.s3_base import TestNeofsS3Base + +logger = logging.getLogger("NeoLogger") + +NEXUS_DOCKER_IMAGE = "sonatype/nexus3" +NEXUS_URL = "http://localhost:8081" +NEXUS_DOCKER_CONTAINER_NAME = "nexus" +NEXUS_DOCKER_VOLUME_NAME = "nexus-data" +ADMIN_USER = "admin" + + +def pytest_generate_tests(metafunc): + if "s3_client" in metafunc.fixturenames: + metafunc.parametrize("s3_client", ["boto3"], indirect=True) + + +def ensure_nexus_running(admin_password: str): + retries = 20 + for attempt in range(retries): + try: + response = requests.get(f"{NEXUS_URL}/service/rest/v1/status", auth=(ADMIN_USER, admin_password)) + if response.status_code == 200: + logger.info("Nexus is up and running.") + return True + logger.info(f"Status Code: {response.status_code}, Response: {response.text}") + except requests.ConnectionError: + pass + logger.info(f"Attempt {attempt + 1}/{retries} failed. Retrying...") + time.sleep(10) + logger.info("Nexus is not running after several attempts.") + return False + + +def configure_blob_store( + admin_password: str, + blob_store_name: str, + s3_bucket_name: str, + s3_endpoint: str, + access_key: str, + secret_key: str, +): + logger.info("Configuring S3 blob store...") + headers = {"Content-Type": "application/json"} + data = { + "name": blob_store_name, + "type": "S3", + "bucketConfiguration": { + "bucket": {"region": "DEFAULT", "name": s3_bucket_name, "prefix": "", "expiration": -1}, + "bucketSecurity": { + "accessKeyId": access_key, + "secretAccessKey": secret_key, + }, + "advancedBucketConnection": {"endpoint": s3_endpoint, "signerType": "AWSS3V4SignerType"}, + }, + } + response = requests.post( + f"{NEXUS_URL}/service/rest/v1/blobstores/s3", + auth=(ADMIN_USER, admin_password), + headers=headers, + data=json.dumps(data), + ) + if response.status_code == 204: + logger.info("S3 blob store configured successfully.") + else: + raise AssertionError( + f"Failed to configure blob store. Status code: {response.status_code}, Response: {response.text}" + ) + + +def get_nexus_admin_password(neofs_env: NeoFSEnv) -> str: + nexus_password = "" + retries = 20 + while retries > 0: + nexus_password_raw = neofs_env.shell.exec( + f"docker exec -it {NEXUS_DOCKER_CONTAINER_NAME} cat /nexus-data/admin.password", + options=CommandOptions(check=False), + ).stdout.strip() + if "No such file or directory" not in nexus_password_raw and nexus_password_raw != "": + nexus_password = nexus_password_raw + break + retries -= 1 + time.sleep(10) + logger.info(f"{nexus_password=}") + assert nexus_password, "Can not retrieve nexus admin password" + return nexus_password + + +class TestS3Nexus(TestNeofsS3Base): + @pytest.fixture(scope="class", autouse=True) + def cleanup_nexus(self): + yield + nexus_logs_file = "nexus_logs.txt" + self.neofs_env.shell.exec(f"docker logs {NEXUS_DOCKER_CONTAINER_NAME} > {nexus_logs_file} 2>&1") + self.neofs_env.shell.exec(f"docker stop {NEXUS_DOCKER_CONTAINER_NAME}") + self.neofs_env.shell.exec(f"docker rm {NEXUS_DOCKER_CONTAINER_NAME}") + self.neofs_env.shell.exec(f"docker volume rm {NEXUS_DOCKER_VOLUME_NAME}") + allure.attach.file(nexus_logs_file, name="nexus logs", extension="txt") + + @allure.title("Test S3: Nexus") + @pytest.mark.skip(reason="Nexus is not supported right now") + @pytest.mark.skipif(sys.platform == "darwin", reason="not supported on macos runners") + def test_s3_nexus_configure_blob_store(self, bucket): + self.neofs_env.shell.exec( + f"docker run -d --network host --name {NEXUS_DOCKER_CONTAINER_NAME} -v {NEXUS_DOCKER_VOLUME_NAME}:/{NEXUS_DOCKER_VOLUME_NAME} " + f'-e INSTALL4J_ADD_VM_PARAMS="-Dcom.amazonaws.sdk.disableCertChecking=true" {NEXUS_DOCKER_IMAGE}' + ) + time.sleep(60) + nexus_admin_password = get_nexus_admin_password(self.neofs_env) + assert ensure_nexus_running(nexus_admin_password), "Nexus is not running" + configure_blob_store( + admin_password=nexus_admin_password, + blob_store_name="s3-blob-store", + s3_bucket_name=bucket, + s3_endpoint=self.neofs_env.s3_gw.address, + access_key=self.access_key_id, + secret_key=self.secret_access_key, + )