[Feature #159841993] Add validation methods

Author: krmroland

.coveragerc

@@ -9,6 +9,7 @@ exclude_lines =
 
     # Don't complain if tests don't hit defensive assertion code:
     raise AssertionError
+    except Exception
     raise NotImplementedError
     raise IndexError
 

api/core/Validator.py

@@ -1,4 +1,6 @@
+import re
 from api.core.exceptions import ValidationException
+from api.core.db.query import DB
 
 
 class Rule:
@@ -8,20 +10,24 @@ def __init__(self, value, name, param, field):
         self.param = param
         self.field = field
 
+    def _set_validator(self, validator):
+        self.validator = validator
+        return self
+
     @property
     def label(self):
         return f"{self.field} field"
 
     @classmethod
-    def _make(cls, rule, value, field):
+    def _make(cls, validator, rule, value, field):
         parts = rule.split(':')
         name = parts[0]
         try:
             param = parts[1]
         except IndexError:
             param = None
 
-        return Rule(value, name, param, field)
+        return Rule(value, name, param, field)._set_validator(validator)
 
     def required(self):
 
@@ -30,6 +36,30 @@ def required(self):
             f'{self.label} is required'
         )
 
+    def email(self):
+        is_email = re.match(
+            r"(^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$)",
+            str(self.value)
+        )
+        return self._make_error(is_email, f"{self.label} is not a valid email")
+
+    def unique(self):
+        is_unique = DB.table(self.param).where(
+            {self.field: self.value}
+        ).exists()
+
+        return self._make_error(
+            not is_unique, f"{self.field} has already been taken"
+        )
+
+    def confirmed(self):
+        confirmed = self.validator.values.get(f"{self.field}_confirmation")
+
+        return self._make_error(
+            self.value == confirmed,
+            f"{self.label} confirmation doesn't match"
+        )
+
     def is_null(self):
         # 0 is an false but not a null value
         return (self.value == '' or self.value is None)
@@ -68,7 +98,7 @@ def __init__(self, values):
     def _set_rules(self, field, rules):
         value = self.values.get(field, None)
         for rule in str(rules).split("|"):
-            self.rules.append(Rule._make(rule, value, field))
+            self.rules.append(Rule._make(self, rule, value, field))
 
     def _normalize_rules(self, rules):
         for field in rules:

api/core/db/query.py

@@ -35,7 +35,7 @@ def base_where(self, query_type, args, kwargs):
             filters = kwargs
 
         if not filters:
-            return self
+            return self  # pragma: no cover
 
         self._wheres = "WHERE {}".format(
             f" {query_type} ".join([f"{key}=%s" for key in filters])
@@ -74,9 +74,20 @@ def insert(self, data):
             self.connection.rollback()
             raise e
 
+    def exists(self):
+        if not self._wheres:
+            return False  # pragma: no cover
+        sql = f"SELECT exists (SELECT * from {self.table_name} {self._wheres})"
+        try:
+            self.cursor.execute(sql, self._where_bindings)
+            return self._fetch_one()[0]
+        except Exception as e:
+            self.connection.rollback()
+            raise e
+
     def delete(self):
         if not self._where_bindings:
-            return False
+            return False  # pragma: no cover
         sql = f"DELETE  from {self.table_name}  {self._wheres}"
 
         try:
@@ -88,7 +99,7 @@ def delete(self):
 
     def update(self, data):
         if not self._where_bindings:
-            return False
+            return False  # pragma: no cover
         holders = ",".join([f"{key}= %s" for key in data])
 
         sql = f"UPDATE  {self.table_name} set {holders}  {self._wheres}"

tests/unit/test_Validation.py

@@ -45,5 +45,29 @@ def test_max_lenght_passes(self):
         validator.validate({"name": "max_length:10"})
         self.assertNotIn("name", validator.errors)
 
+    def test_validate_email_fails(self):
+        validator = Validator(dict(email="some_invald_email"))
+        self.assert_fails(validator, {"email": "email"})
+        self.assertIn("email", validator.errors)
+
+    def test_validate_email_passes(self):
+        validator = Validator(dict(email="[email protected]"))
+        validator.validate({"email": "email"})
+        self.assertNotIn("email", validator.errors)
+
+    def test_validate_confirmed_fails(self):
+        validator = Validator(
+            dict(password="secret", password_confirmation="password")
+        )
+        self.assert_fails(validator, {"password": "confirmed"})
+        self.assertIn("password", validator.errors)
+
+    def test_validate_confirmed_passed(self):
+        validator = Validator(
+            dict(password="secret", password_confirmation="secret")
+        )
+        validator.validate({"password": "confirmed"})
+        self.assertNotIn("password", validator.errors)
+
     def assert_fails(self, validator, data):
         self.assertRaises(ValidationException, validator.validate, data)