Add put/get bucket logging

qcloud_cos/cos_client.py

@@ -1021,7 +1021,7 @@ def get_bucket_lifecycle(self, Bucket, **kwargs):
         """
         headers = mapped(kwargs)
         url = self._conf.uri(bucket=Bucket, path="?lifecycle")
-        logger.info("get bucket cors, url=:{url} ,headers=:{headers}".format(
+        logger.info("get bucket lifecycle, url=:{url} ,headers=:{headers}".format(
             url=url,
             headers=headers))
         rt = self.send_request(
@@ -1185,6 +1185,51 @@ def delete_bucket_replication(self, Bucket, **kwargs):
             headers=headers)
         return None
 
+    def put_bucket_logging(self, Bucket, BucketLoggingStatus={}, **kwargs):
+        """设置bucket logging
+
+        :param Bucket(string): 存储桶名称.
+        :param BucketLoggingStatus(dict): 设置Bucket的日志配置.
+        :param kwargs(dict): 设置请求headers.
+        :return: None.
+        """
+        xml_config = format_xml(data=BucketLoggingStatus, root='BucketLoggingStatus')
+        headers = mapped(kwargs)
+        headers['Content-MD5'] = get_md5(xml_config)
+        headers['Content-Type'] = 'application/xml'
+        url = self._conf.uri(bucket=Bucket, path="?logging")
+        logger.info("put bucket logging, url=:{url} ,headers=:{headers}".format(
+            url=url,
+            headers=headers))
+        logging_rt = self.send_request(
+            method='PUT',
+            url=url,
+            data=xml_config,
+            auth=CosS3Auth(self._conf._secret_id, self._conf._secret_key),
+            headers=headers)
+        grant_rt = self.put_bucket_acl(Bucket=Bucket, GrantFullControl=LOGGING_UIN)
+        return None
+
+    def get_bucket_logging(self, Bucket, **kwargs):
+        """获取bucket logging
+
+        :param Bucket(string): 存储桶名称.
+        :param kwargs(dict): 设置请求headers.
+        :return(dict): Bucket对应的logging配置.
+        """
+        headers = mapped(kwargs)
+        url = self._conf.uri(bucket=Bucket, path="?logging")
+        logger.info("get bucket logging, url=:{url} ,headers=:{headers}".format(
+            url=url,
+            headers=headers))
+        rt = self.send_request(
+            method='GET',
+            url=url,
+            auth=CosS3Auth(self._conf._secret_id, self._conf._secret_key),
+            headers=headers)
+        data = xml_to_dict(rt.text)
+        return data
+
     # service interface begin
     def list_buckets(self, **kwargs):
         """列出所有bucket

qcloud_cos/cos_comm.py

@@ -15,6 +15,7 @@
 from cos_exception import CosServiceError
 
 SINGLE_UPLOAD_LENGTH = 5*1024*1024*1024  # 单次上传文件最大为5G
+LOGGING_UIN = 'id="qcs::cam::uin/100001001014:uin/100001001014"'
 # kwargs中params到http headers的映射
 maplist = {
             'ContentLength': 'Content-Length',
@@ -100,6 +101,7 @@ def xml_to_dict(data, origin_str="", replace_str=""):
     xmldict = Xml2Dict(root)
     xmlstr = str(xmldict)
     xmlstr = xmlstr.replace("{http://www.qcloud.com/document/product/436/7751}", "")
+    xmlstr = xmlstr.replace("{http://doc.s3.amazonaws.com/2006-03-01}", "")
     xmlstr = xmlstr.replace("{http://www.w3.org/2001/XMLSchema-instance}", "")
     if origin_str:
         xmlstr = xmlstr.replace(origin_str, replace_str)

ut/test.py

@@ -311,7 +311,6 @@ def test_delete_multiple_objects():
         Bucket=test_bucket,
         Delete=objects
     )
-    assert response
 
 
 def test_create_head_delete_bucket():
@@ -611,6 +610,7 @@ def test_use_get_auth():
     response = requests.get('http://test01-1252448703.cos.ap-beijing-1.myqcloud.com/test.txt?acl&unsed=123', headers={'Authorization': auth})
     assert response.status_code == 200
 
+
 def test_upload_with_server_side_encryption():
     """上传带上加密头部,下载时验证有该头部"""
     response = client.put_object(
@@ -621,6 +621,34 @@ def test_upload_with_server_side_encryption():
     )
     assert response['x-cos-server-side-encryption'] == 'AES256'
 
+
+def test_put_get_bucket_logging():
+    """测试bucket的logging服务"""
+    logging_bucket = 'logging-beijing-1252448703'
+    logging_config = {
+        'LoggingEnabled': {
+            'TargetBucket': logging_bucket,
+            'TargetPrefix': 'test'
+        }
+    }
+    beijing_conf = CosConfig(
+        Region="ap-beijing",
+        Secret_id=SECRET_ID,
+        Secret_key=SECRET_KEY
+    )
+    logging_client = CosS3Client(beijing_conf)
+    response = logging_client.put_bucket_logging(
+        Bucket=logging_bucket,
+        BucketLoggingStatus=logging_config
+    )
+    time.sleep(4)
+    response = logging_client.get_bucket_logging(
+        Bucket=logging_bucket
+    )
+    assert response['LoggingEnabled']['TargetBucket'] == logging_bucket
+    assert response['LoggingEnabled']['TargetPrefix'] == 'test'
+
+
 if __name__ == "__main__":
     setUp()
     test_upload_with_server_side_encryption()
@@ -634,4 +662,5 @@ def test_upload_with_server_side_encryption():
     test_copy_10G_file_in_same_region()
     test_list_objects()
     test_use_get_auth()
+    test_put_get_bucket_logging()
     tearDown()