From 5595ffb25497c587cdbac5494fb0346546d6c492 Mon Sep 17 00:00:00 2001 From: zhuxiaolong37 Date: Wed, 29 Oct 2025 11:04:58 +0800 Subject: [PATCH] AccessMonitor adds allow_copy field --- .../models/bucket_access_monitor.py | 4 + sample/put_bucket_access_monitor.py | 4 +- .../test_bucket_access_monitor_client.py | 37 ++++++++ .../unit/models/test_bucket_access_monitor.py | 90 +++++++++++++++++++ 4 files changed, 134 insertions(+), 1 deletion(-) diff --git a/alibabacloud_oss_v2/models/bucket_access_monitor.py b/alibabacloud_oss_v2/models/bucket_access_monitor.py index 3feb257..9a769b5 100644 --- a/alibabacloud_oss_v2/models/bucket_access_monitor.py +++ b/alibabacloud_oss_v2/models/bucket_access_monitor.py @@ -20,6 +20,7 @@ class AccessMonitorConfiguration(serde.Model): _attribute_map = { 'status': {'tag': 'xml', 'rename': 'Status', 'type': 'str'}, + 'allow_copy': {'tag': 'xml', 'rename': 'AllowCopy', 'type': 'bool'}, } _xml_map = { @@ -29,14 +30,17 @@ class AccessMonitorConfiguration(serde.Model): def __init__( self, status: Optional[Union[str, AccessMonitorStatusType]] = None, + allow_copy: Optional[bool] = None, **kwargs: Any ) -> None: """ Args: status (str | AccessMonitorStatusType, optional): The access tracking status of the bucket. Valid values:- Enabled: Access tracking is enabled.- Disabled: Access tracking is disabled. + allow_copy (bool, optional): Whether to allow copying. """ super().__init__(**kwargs) self.status = status + self.allow_copy = allow_copy class PutBucketAccessMonitorRequest(serde.RequestModel): diff --git a/sample/put_bucket_access_monitor.py b/sample/put_bucket_access_monitor.py index b17618b..1122098 100644 --- a/sample/put_bucket_access_monitor.py +++ b/sample/put_bucket_access_monitor.py @@ -6,6 +6,7 @@ parser.add_argument('--bucket', help='The name of the bucket.', required=True) parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS') parser.add_argument('--status', help='The access tracking status of the bucket. Valid values:- Enabled: Access tracking is enabled.- Disabled: Access tracking is disabled.', required=True) +parser.add_argument('--allow_copy', help='Whether to allow copying.') def main(): @@ -26,7 +27,8 @@ def main(): # result = client.put_bucket_access_monitor(oss.PutBucketAccessMonitorRequest( # bucket=args.bucket, # access_monitor_configuration=oss.AccessMonitorConfiguration( - # status=oss.AccessMonitorStatusType.ENABLED + # status=oss.AccessMonitorStatusType.ENABLED, + # allow_copy=args.allow_copy # ), # )) diff --git a/tests/integration/test_bucket_access_monitor_client.py b/tests/integration/test_bucket_access_monitor_client.py index 8df3887..798d539 100644 --- a/tests/integration/test_bucket_access_monitor_client.py +++ b/tests/integration/test_bucket_access_monitor_client.py @@ -42,6 +42,43 @@ def test_bucket_access_monitor(self): self.assertEqual(24, len(result.headers.get('x-oss-request-id'))) self.assertEqual(oss.AccessMonitorStatusType.ENABLED, result.access_monitor_configuration.status) + def test_bucket_access_monitor_with_allow_copy(self): + # create bucket + bucket_name = random_bucket_name() + result = self.client.put_bucket(oss.PutBucketRequest( + bucket=bucket_name, + acl='private', + create_bucket_configuration=oss.CreateBucketConfiguration( + storage_class='IA' + ) + )) + self.assertEqual(200, result.status_code) + self.assertEqual('OK', result.status) + self.assertEqual(24, len(result.request_id)) + self.assertEqual(24, len(result.headers.get('x-oss-request-id'))) + + # put bucket access monitor with allow_copy + result = self.client.put_bucket_access_monitor(oss.PutBucketAccessMonitorRequest( + bucket=bucket_name, + access_monitor_configuration=oss.AccessMonitorConfiguration( + status=oss.AccessMonitorStatusType.ENABLED, + allow_copy=True + ), + )) + self.assertEqual(200, result.status_code) + self.assertEqual('OK', result.status) + self.assertEqual(24, len(result.headers.get('x-oss-request-id'))) + + # get bucket access monitor + result = self.client.get_bucket_access_monitor(oss.GetBucketAccessMonitorRequest( + bucket=bucket_name, + )) + self.assertEqual(200, result.status_code) + self.assertEqual('OK', result.status) + self.assertEqual(24, len(result.headers.get('x-oss-request-id'))) + self.assertEqual(oss.AccessMonitorStatusType.ENABLED, result.access_monitor_configuration.status) + self.assertTrue(result.access_monitor_configuration.allow_copy) + def test_bucket_access_monitor_v1(self): # create bucket bucket_name = random_bucket_name() diff --git a/tests/unit/models/test_bucket_access_monitor.py b/tests/unit/models/test_bucket_access_monitor.py index d0e763f..ca7c5c3 100644 --- a/tests/unit/models/test_bucket_access_monitor.py +++ b/tests/unit/models/test_bucket_access_monitor.py @@ -34,6 +34,31 @@ def test_constructor_request(self): self.assertEqual('bucket_name', request.bucket) self.assertEqual('Enabled', request.access_monitor_configuration.status) + def test_constructor_request_with_allow_copy(self): + # Test constructor with allow_copy parameter + request = model.PutBucketAccessMonitorRequest( + bucket='bucket_name', + access_monitor_configuration=model.AccessMonitorConfiguration( + status='Enabled', + allow_copy=True + ), + ) + self.assertEqual('bucket_name', request.bucket) + self.assertEqual('Enabled', request.access_monitor_configuration.status) + self.assertTrue(request.access_monitor_configuration.allow_copy) + + # Test allow_copy with False value + request = model.PutBucketAccessMonitorRequest( + bucket='bucket_name', + access_monitor_configuration=model.AccessMonitorConfiguration( + status='Disabled', + allow_copy=False + ), + ) + self.assertEqual('bucket_name', request.bucket) + self.assertEqual('Disabled', request.access_monitor_configuration.status) + self.assertFalse(request.access_monitor_configuration.allow_copy) + def test_serialize_request(self): request = model.PutBucketAccessMonitorRequest( bucket='bucket_name', @@ -133,6 +158,27 @@ def test_constructor_result(self): ) self.assertEqual('Disabled', result.access_monitor_configuration.status) + def test_constructor_result_with_allow_copy(self): + # Test constructor with allow_copy parameter + result = model.GetBucketAccessMonitorResult( + access_monitor_configuration=model.AccessMonitorConfiguration( + status='Enabled', + allow_copy=True + ), + ) + self.assertEqual('Enabled', result.access_monitor_configuration.status) + self.assertTrue(result.access_monitor_configuration.allow_copy) + + # Test allow_copy with False value + result = model.GetBucketAccessMonitorResult( + access_monitor_configuration=model.AccessMonitorConfiguration( + status='Disabled', + allow_copy=False + ), + ) + self.assertEqual('Disabled', result.access_monitor_configuration.status) + self.assertFalse(result.access_monitor_configuration.allow_copy) + def test_deserialize_result(self): xml_data = r''' @@ -160,3 +206,47 @@ def test_deserialize_result(self): self.assertEqual('OK', result.status) self.assertEqual('Enabled', result.access_monitor_configuration.status) + def test_deserialize_result_with_allow_copy(self): + # Test deserialization with AllowCopy field in XML data + xml_data = r''' + + Enabled + true + + ''' + + result = model.GetBucketAccessMonitorResult() + op_output = OperationOutput( + status='OK', + status_code=200, + http_response=MockHttpResponse( + body=xml_data, + ) + ) + deserializer = [serde.deserialize_output_xmlbody] + serde.deserialize_output(result, op_output, custom_deserializer=deserializer) + self.assertEqual('OK', result.status) + self.assertEqual('Enabled', result.access_monitor_configuration.status) + self.assertTrue(result.access_monitor_configuration.allow_copy) + + # Test allow_copy with False value + xml_data = r''' + + Disabled + false + + ''' + + result = model.GetBucketAccessMonitorResult() + op_output = OperationOutput( + status='OK', + status_code=200, + http_response=MockHttpResponse( + body=xml_data, + ) + ) + deserializer = [serde.deserialize_output_xmlbody] + serde.deserialize_output(result, op_output, custom_deserializer=deserializer) + self.assertEqual('OK', result.status) + self.assertEqual('Disabled', result.access_monitor_configuration.status) + self.assertFalse(result.access_monitor_configuration.allow_copy) \ No newline at end of file