add funtion of bucket encryption

Author: jayzhenghan

qcloud_cos/cos_client.py

@@ -3425,23 +3425,21 @@ def update_object_meta(self, Bucket, Key, **kwargs):
         )
         return response
 
-    def put_bucket_encryption(self, Bucket, SSEAlgorithm=None, **kwargs):
+    def put_bucket_encryption(self, Bucket, ServerSideEncryptionConfiguration={}, **kwargs):
         """设置执行存储桶下的默认加密配置
 
         :param Bucket(string): 存储桶名称.
         :param SSEAlgorithm(string): 要使用的服务端加密算法
         :param kwargs(dict): 设置下载的headers.
         :return(dict): 设置成功返回的结果.
         """
-        if SSEAlgorithm is None:
-            raise CosClientError('SSEAlgorithm is None and must be set!')
-
-        root = xml.etree.ElementTree.Element("ServerSideEncryptionConfiguration")
-        rule_node = xml.etree.ElementTree.SubElement(root, "Rule")
-        apply_node = xml.etree.ElementTree.SubElement(rule_node, "ApplySideEncryptionConfiguration")
-        xml.etree.ElementTree.SubElement(apply_node, "SSEAlgorithm").text = to_str(SSEAlgorithm)
-        body = xml.etree.ElementTree.tostring(root, encoding='utf-8')
-
+        # 类型为list的标签
+        lst = [
+            '<Rule>',
+            '</Rule>'
+        ]
+        xml_config = format_xml(data=ServerSideEncryptionConfiguration, root='ServerSideEncryptionConfiguration', lst=lst)
+        logger.info('xml_config: {0}'.format(xml_config))
         headers = mapped(kwargs)
         params = {'encryption': ''}
         url = self._conf.uri(bucket=Bucket)
@@ -3453,12 +3451,11 @@ def put_bucket_encryption(self, Bucket, SSEAlgorithm=None, **kwargs):
             url=url,
             bucket=Bucket,
             auth=CosS3Auth(self._conf, params=params),
-            data=to_bytes(body),
+            data=xml_config,
             headers=headers,
             params=params)
 
-        response = dict(**rt.headers)
-        return response
+        return None
 
     def get_bucket_encryption(self, Bucket, **kwargs):
         """获取存储桶下的默认加密配置
@@ -3480,15 +3477,11 @@ def get_bucket_encryption(self, Bucket, **kwargs):
             auth=CosS3Auth(self._conf, params=params),
             headers=headers,
             params=params)
-        data = xml.etree.ElementTree.fromstring(rt.content)
-        sse_algorithm = None
-        rule = data.find('Rule')
-        if rule is not None:
-            apply = rule.find('ApplyServerSideEncryptionByDefault')
-            if apply is not None:
-                sse_algorithm = to_str(apply.find('SSEAlgorithm').text)
 
-        return sse_algorithm
+        data = xml_to_dict(rt.content)
+        format_dict(data, ['Rule'])
+        logger.info('data: {0}'.format(data))
+        return data
 
     def delete_bucket_encryption(self, Bucket, **kwargs):
         """用于删除指定存储桶下的默认加密配置
@@ -3500,7 +3493,7 @@ def delete_bucket_encryption(self, Bucket, **kwargs):
         headers = mapped(kwargs)
         params = {'encryption': ''}
         url = self._conf.uri(bucket=Bucket)
-        logger.info("put bucket encryption, url=:{url} ,headers=:{headers}".format(
+        logger.info("delete bucket encryption, url=:{url} ,headers=:{headers}".format(
             url=url,
             headers=headers))
         rt = self.send_request(
@@ -3511,8 +3504,7 @@ def delete_bucket_encryption(self, Bucket, **kwargs):
             headers=headers,
             params=params)
 
-        response = dict(**rt.headers)
-        return response
+        return None
 
 
 if __name__ == "__main__":

ut/test.py

@@ -1215,14 +1215,24 @@ def test_download_file():
 def test_bucket_encryption():
     """测试存储桶默认加密配置"""
     # 测试设置存储桶的默认加密配置
-    client.put_bucket_encryption(copy_test_bucket, 'AES256')
+    config_dict = {
+        'Rule': [
+            {
+                'ApplySideEncryptionConfiguration': {
+                    'SSEAlgorithm': 'AES256',
+                }
+            },
+        ]
+    }
+    client.put_bucket_encryption(test_bucket, config_dict)
 
     # 测试获取存储桶默认加密配置
-    ret = client.get_bucket_encryption(copy_test_bucket)
-    assert(ret == 'AES256')
+    ret = client.get_bucket_encryption(test_bucket)
+    sse_algorithm = ret['Rule'][0]['ApplyServerSideEncryptionByDefault']['SSEAlgorithm']
+    assert(sse_algorithm == 'AES256')
 
     # 删除存储桶默认加密配置
-    client.delete_bucket_encryption(copy_test_bucket)
+    client.delete_bucket_encryption(test_bucket)
 
 
 if __name__ == "__main__":