AccessMonitor adds allow_copy field

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):

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
     #         ),
     # ))
 

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()

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'''
         <AccessMonitorConfiguration>
@@ -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'''
+        <AccessMonitorConfiguration>
+          <Status>Enabled</Status>
+          <AllowCopy>true</AllowCopy>
+        </AccessMonitorConfiguration>
+        '''
+
+        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'''
+        <AccessMonitorConfiguration>
+          <Status>Disabled</Status>
+          <AllowCopy>false</AllowCopy>
+        </AccessMonitorConfiguration>
+        '''
+
+        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)