test_bucket_policy_elements_NotAction[Allow] - occasional test case failures botocore.exceptions.ClientError - check and fix the test

[ypersky1980]

test_bucket_policy_elements_NotAction[Allow] - occasional test case failures botocore.exceptions.ClientError: An error occurred (AccessDenied) when calling the PutObject operation: Access Denied
Relevant links:

https://reportportal-ocs4.apps.ocp-c1.prod.psi.redhat.com/ui/#ocs/launches/795/27895/1372224/1372312/log
https://reportportal-ocs4.apps.ocp-c1.prod.psi.redhat.com/ui/#ocs/launches/795/28201/1389234/1389325/log
https://reportportal-ocs4.apps.ocp-c1.prod.psi.redhat.com/ui/#ocs/launches/795/28309/1392793/1392881/log
https://reportportal-ocs4.apps.ocp-c1.prod.psi.redhat.com/ui/#ocs/launches/795/28524/1400328/1400387/log
https://reportportal-ocs4.apps.ocp-c1.prod.psi.redhat.com/ui/#ocs/launches/795/28641/1405926/1406020/log
https://reportportal-ocs4.apps.ocp-c1.prod.psi.redhat.com/ui/#ocs/launches/795/28815/1413461/1413558/log

The full backtrace is:

self = <mcg.test_bucket_policy.TestS3BucketPolicy object at 0x7fafa8c57be0>
mcg_obj = <ocs_ci.ocs.resources.mcg.MCG object at 0x7fafa9082a30>
bucket_factory = <function bucket_factory_fixture.._create_buckets at 0x7fafb4b163a0>
effect = 'Allow'

@pytest.mark.parametrize(
argnames="effect",
argvalues=[
pytest.param(
["Allow"], marks=[tier1, pytest.mark.polarion_id("OCS-5768")]
),
pytest.param(["Deny"], marks=[tier1, pytest.mark.polarion_id("OCS-5769")]),
],
)
def test_bucket_policy_elements_NotAction(self, mcg_obj, bucket_factory, effect):
"""
Test bucket policy element of NotAction with Effect: Allow/Deny
"""

# Creating obc and obc object to get account details, keys etc
obc_bucket = bucket_factory(amount=2, interface="OC")
obc_obj = OBC(obc_bucket[0].name)
obc_obj1 = OBC(obc_bucket[1].name)

# Set bucket policy for user
bucket_policy_generated = gen_bucket_policy(
    user_list=obc_obj1.obc_account,
    action_property="NotAction",
    actions_list=["DeleteBucket"],
    resources_list=[f'{obc_obj.bucket_name}/{"*"}'],
    effect=effect,
)
if effect == "Allow":
    bucket_policy_generated["Statement"][0]["NotAction"][0] = "s3:ListBucket"
bucket_policy = json.dumps(bucket_policy_generated)

# Add Bucket Policy
logger.info(f"Creating bucket policy on bucket: {obc_obj.bucket_name}")
put_policy = put_bucket_policy(mcg_obj, obc_obj.bucket_name, bucket_policy)
logger.info(f"Put bucket policy response from admin: {put_policy}")

# Get bucket policy on the bucket
logger.info(f"Getting Bucket policy on bucket: {obc_obj.bucket_name}")
get_policy = get_bucket_policy(mcg_obj, obc_obj.bucket_name)
logger.info(f"Got bucket policy: {get_policy['Policy']}")

# Verify DeleteBucket and putObject operation
# in both scenarios: Effect=Allow/Deny
if effect == "Allow":
    # Put Object is allowed
    logger.info("Writing index data to the bucket")

  assert s3_put_object(
        s3_obj=obc_obj1,
        bucketname=obc_obj.bucket_name,
        object_key="index.html",
        data=index,
        content_type="text/html",
    ), "Failed to put object."

tests/functional/object/mcg/test_bucket_policy.py:977:

ocs_ci/ocs/bucket_utils.py:1316: in s3_put_object
return s3_obj.s3_client.put_object(
venv/lib64/python3.9/site-packages/botocore/client.py:514: in _api_call
return self._make_api_call(operation_name, kwargs)

self = <botocore.client.S3 object at 0x7fafb2040b80>
operation_name = 'PutObject'
api_params = {'Body': <_io.BytesIO object at 0x7fafa9f3c310>, 'Bucket': 'oc-bucket-38a7e8763ba44b58ab0b235b098994', 'ContentEncoding': '', 'ContentType': 'text/html', ...}

def _make_api_call(self, operation_name, api_params):
operation_model = self._service_model.operation_model(operation_name)
service_name = self._service_model.service_name
history_recorder.record(
'API_CALL',
{
'service': service_name,
'operation': operation_name,
'params': api_params,
},
)
if operation_model.deprecated:
logger.debug(
'Warning: %s.%s() is deprecated', service_name, operation_name
)
request_context = {
'client_region': self.meta.region_name,
'client_config': self.meta.config,
'has_streaming_input': operation_model.has_streaming_input,
'auth_type': operation_model.auth_type,
}
request_dict = self._convert_to_request_dict(
api_params, operation_model, context=request_context
)
resolve_checksum_context(request_dict, operation_model, api_params)

service_id = self._service_model.service_id.hyphenize()
handler, event_response = self.meta.events.emit_until_response(
    'before-call.{service_id}.{operation_name}'.format(
        service_id=service_id, operation_name=operation_name
    ),
    model=operation_model,
    params=request_dict,
    request_signer=self._request_signer,
    context=request_context,
)

if event_response is not None:
    http, parsed_response = event_response
else:
    apply_request_checksum(request_dict)
    http, parsed_response = self._make_request(
        operation_model, request_dict, request_context
    )

self.meta.events.emit(
    'after-call.{service_id}.{operation_name}'.format(
        service_id=service_id, operation_name=operation_name
    ),
    http_response=http,
    parsed=parsed_response,
    model=operation_model,
    context=request_context,
)

if http.status_code >= 300:
    error_code = parsed_response.get("Error", {}).get("Code")
    error_class = self.exceptions.from_code(error_code)

  raise error_class(parsed_response, operation_name)

E botocore.exceptions.ClientError: An error occurred (AccessDenied) when calling the PutObject operation: Access Denied

venv/lib64/python3.9/site-packages/botocore/client.py:938: ClientError

2025-01-04 03:55:42