Add validation for s3

Author: ajaniszewska-dev

storage_backend_s3/components/s3_adapter.py

@@ -46,6 +46,7 @@ def _get_bucket(self):
         s3 = boto3.resource("s3", **params)
         bucket_name = self.collection.aws_bucket
         bucket = self._get_or_create_bucket(s3, bucket_name, **params)
+        self.collection.validated = True
         return bucket
 
     def _get_or_create_bucket(self, s3, bucket_name, **params):
@@ -60,9 +61,7 @@ def _get_or_create_bucket(self, s3, bucket_name, **params):
         return bucket
 
     def _check_bucket_exists(self, s3, bucket_name, force=False):
-        if self.env.context.get("force_s3_check"):
-            force = True
-        if force is False and self.collection.s3_bucket_exists:
+        if force is False and self.collection.validated:
             return True
         # if test is ok, set the flag
         try:
@@ -142,3 +141,8 @@ def list(self, relative_path):
     def delete(self, relative_path):
         s3object = self._get_object(relative_path)
         s3object.delete()
+
+    def validate_config(self,**params):
+        s3 = boto3.resource("s3", **params)
+        bucket_name = self.collection.aws_bucket
+        self.validated = self._check_bucket_exists(s3, bucket_name)

storage_backend_s3/models/storage_backend.py

@@ -6,7 +6,7 @@
 
 import logging
 
-from odoo import fields, models
+from odoo import _, fields, models
 
 _logger = logging.getLogger(__name__)
 
@@ -43,7 +43,7 @@ class StorageBackend(models.Model):
         "eg: Exoscale",
     )
     aws_bucket = fields.Char(string="Bucket")
-    s3_bucket_exists = fields.Boolean(string="Bucket exists")
+    validated = fields.Boolean(string="Bucket exists", compute="_compute_s3_validated")
     aws_access_key_id = fields.Char(string="Access Key ID")
     aws_secret_access_key = fields.Char(string="Secret Access Key")
     aws_region = fields.Selection(selection="_selection_aws_region", string="Region")
@@ -62,6 +62,12 @@ class StorageBackend(models.Model):
         ]
     )
 
+    def _compute_s3_validated(self):
+        for rec in self:
+            if rec.backend_type == "amazon_s3":
+                adapter = self._get_adapter()
+                rec.validated = hasattr(adapter, "validate_config")
+
     @property
     def _server_env_fields(self):
         env_fields = super()._server_env_fields
@@ -86,6 +92,26 @@ def _selection_aws_region(self):
             + [("other", "Empty or Other (Manually specify below)")]
         )
 
-    def action_ensure_bucket_exists(self):
+    def action_test_config(self):
+        if not self.validated:
+            raise AttributeError("Validation not supported!")
         adapter = self._get_adapter()
-        adapter._get_bucket()
+        try:
+            adapter.validate_config()
+            title = _("Connection Test Succeeded!")
+            message = _("Everything seems properly set up!")
+            msg_type = "success"
+        except Exception as err:
+            title = _("Connection Test Failed!")
+            message = str(err)
+            msg_type = "danger"
+        return {
+            "type": "ir.actions.client",
+            "tag": "display_notification",
+            "params": {
+                "title": title,
+                "message": message,
+                "type": msg_type,
+                "sticky": False,
+            },
+        }

storage_backend_s3/views/backend_storage_view.xml

@@ -33,7 +33,7 @@
                             name="aws_bucket"
                             attrs="{'required': [('backend_type', '=', 'amazon_s3')]}"
                         />
-                        <field name="s3_bucket_exists" />
+                        <field name="validated" />
                         <field name="aws_cache_control" />
                         <field name="aws_file_acl" />
                     </group>