frostfs-testcases/pytest_tests/testsuites/services/s3_gate/test_s3_policy.py

import os
import time
from datetime import datetime, timedelta
from random import choice
from string import ascii_letters
from typing import Tuple

import allure
import pytest
from file_helper import generate_file, generate_file_with_content
from python_keywords.container import search_container_by_name
from python_keywords.storage_policy import get_simple_object_copies
from s3_helper import (
    assert_object_lock_mode,
    check_objects_in_bucket,
    object_key_from_file_path,
    set_bucket_versioning,
)

from steps import s3_gate_bucket, s3_gate_object
from steps.s3_gate_base import TestS3GateBase


def pytest_generate_tests(metafunc):
    policy = f"{os.getcwd()}/robot/resources/files/policy.json"
    if "s3_client" in metafunc.fixturenames:
        metafunc.parametrize(
            "s3_client",
            [("aws cli", policy), ("boto3", policy)],
            indirect=True,
            ids=["aws cli", "boto3"],
        )


@pytest.mark.s3_gate
class TestS3GatePolicy(TestS3GateBase):
    @allure.title("Test S3: Verify bucket creation with retention policy applied")
    def test_s3_bucket_location(self):
        file_path_1 = generate_file()
        file_name_1 = object_key_from_file_path(file_path_1)
        file_path_2 = generate_file()
        file_name_2 = object_key_from_file_path(file_path_2)

        with allure.step("Create two buckets with different bucket configuration"):
            bucket_1 = s3_gate_bucket.create_bucket_s3(
                self.s3_client, bucket_configuration="complex"
            )
            set_bucket_versioning(self.s3_client, bucket_1, s3_gate_bucket.VersioningStatus.ENABLED)
            bucket_2 = s3_gate_bucket.create_bucket_s3(self.s3_client, bucket_configuration="rep-3")
            set_bucket_versioning(self.s3_client, bucket_2, s3_gate_bucket.VersioningStatus.ENABLED)
            list_buckets = s3_gate_bucket.list_buckets_s3(self.s3_client)
            assert (
                bucket_1 in list_buckets and bucket_2 in list_buckets
            ), f"Expected two buckets {bucket_1, bucket_2}, got {list_buckets}"

            # with allure.step("Check head buckets"):
            head_1 = s3_gate_bucket.head_bucket(self.s3_client, bucket_1)
            head_2 = s3_gate_bucket.head_bucket(self.s3_client, bucket_2)
            assert head_1 == {} or head_1.get("HEAD") == None, "Expected head is empty"
            assert head_2 == {} or head_2.get("HEAD") == None, "Expected head is empty"

        with allure.step("Put objects into buckets"):
            version_id_1 = s3_gate_object.put_object_s3(self.s3_client, bucket_1, file_path_1)
            version_id_2 = s3_gate_object.put_object_s3(self.s3_client, bucket_2, file_path_2)
            check_objects_in_bucket(self.s3_client, bucket_1, [file_name_1])
            check_objects_in_bucket(self.s3_client, bucket_2, [file_name_2])

        with allure.step("Check bucket location"):
            bucket_loc_1 = s3_gate_bucket.get_bucket_location(self.s3_client, bucket_1)
            bucket_loc_2 = s3_gate_bucket.get_bucket_location(self.s3_client, bucket_2)
            assert bucket_loc_1 == "complex"
            assert bucket_loc_2 == "rep-3"

        with allure.step("Check object policy"):
            cid_1 = search_container_by_name(
                self.wallet, bucket_1, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint
            )
            copies_1 = get_simple_object_copies(
                wallet=self.wallet,
                cid=cid_1,
                oid=version_id_1,
                shell=self.shell,
                nodes=self.cluster.storage_nodes,
            )
            assert copies_1 == 1
            cid_2 = search_container_by_name(
                self.wallet, bucket_2, shell=self.shell, endpoint=self.cluster.default_rpc_endpoint
            )
            copies_2 = get_simple_object_copies(
                wallet=self.wallet,
                cid=cid_2,
                oid=version_id_2,
                shell=self.shell,
                nodes=self.cluster.storage_nodes,
            )
            assert copies_2 == 3

    @allure.title("Test S3: bucket policy ")
    def test_s3_bucket_policy(self):
        with allure.step("Create bucket with default policy"):
            bucket = s3_gate_bucket.create_bucket_s3(self.s3_client)
            set_bucket_versioning(self.s3_client, bucket, s3_gate_bucket.VersioningStatus.ENABLED)

        with allure.step("GetBucketPolicy"):
            s3_gate_bucket.get_bucket_policy(self.s3_client, bucket)

        with allure.step("Put new policy"):
            custom_policy = f"file://{os.getcwd()}/robot/resources/files/bucket_policy.json"
            custom_policy = {
                "Version": "2008-10-17",
                "Id": "aaaa-bbbb-cccc-dddd",
                "Statement": [
                    {
                        "Sid": "AddPerm",
                        "Effect": "Allow",
                        "Principal": {"AWS": "*"},
                        "Action": ["s3:GetObject"],
                        "Resource": [f"arn:aws:s3:::{bucket}/*"],
                    }
                ],
            }

            s3_gate_bucket.put_bucket_policy(self.s3_client, bucket, custom_policy)
        with allure.step("GetBucketPolicy"):
            policy_1 = s3_gate_bucket.get_bucket_policy(self.s3_client, bucket)
            print(policy_1)

    @allure.title("Test S3: bucket policy ")
    def test_s3_cors(self):
        with allure.step("Create bucket without cors"):
            bucket = s3_gate_bucket.create_bucket_s3(self.s3_client)
            set_bucket_versioning(self.s3_client, bucket, s3_gate_bucket.VersioningStatus.ENABLED)

        with pytest.raises(Exception):
            bucket_cors = s3_gate_bucket.get_bucket_cors(self.s3_client, bucket)

        with allure.step("Put bucket cors"):
            cors = {
                "CORSRules": [
                    {
                        "AllowedOrigins": ["http://www.example.com"],
                        "AllowedHeaders": ["*"],
                        "AllowedMethods": ["PUT", "POST", "DELETE"],
                        "MaxAgeSeconds": 3000,
                        "ExposeHeaders": ["x-amz-server-side-encryption"],
                    },
                    {
                        "AllowedOrigins": ["*"],
                        "AllowedHeaders": ["Authorization"],
                        "AllowedMethods": ["GET"],
                        "MaxAgeSeconds": 3000,
                    },
                ]
            }
            s3_gate_bucket.put_bucket_cors(self.s3_client, bucket, cors)
            bucket_cors = s3_gate_bucket.get_bucket_cors(self.s3_client, bucket)
            assert bucket_cors == cors.get(
                "CORSRules"
            ), f"Expected corsrules must be {cors.get('CORSRules')}"

        with allure.step("delete bucket cors"):
            s3_gate_bucket.delete_bucket_cors(self.s3_client, bucket)

        with pytest.raises(Exception):
            bucket_cors = s3_gate_bucket.get_bucket_cors(self.s3_client, bucket)