lib upgrades: pyproj, lxml, owslib, libpass - removed: wtforms, passlib

GeoHealthCheck/models.py

@@ -30,7 +30,6 @@
 
 import json
 import logging
-from flask_babel import gettext as _
 from datetime import datetime, timedelta, timezone
 from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
 from sqlalchemy import func, and_
@@ -43,7 +42,6 @@
 from factory import Factory
 from init import App
 from resourceauth import ResourceAuth
-from wtforms.validators import Email, ValidationError
 from owslib.util import bind_url
 
 APP = App.get_app()
@@ -246,30 +244,21 @@ def _validate_webhook(value):
     try:
         _parse_webhook_location(value)
     except ValueError as err:
-        raise ValidationError('{}: {}'.format(value, err))
+        raise ValueError('{}: {}'.format(value, err))
     return value
 
 
 def _validate_email(value):
     if not value:
-        raise ValidationError("Email cannot be empty value")
+        raise ValueError("Email cannot be empty value")
     try:
         if not value.strip():
-            raise ValidationError("Email cannot be empty value")
+            raise ValueError("Email cannot be empty value")
     except AttributeError:
-        raise ValidationError("Email cannot be empty value")
+        raise ValueError("Email cannot be empty value")
 
-    v = Email()
-
-    class dummy_value(object):
-        data = value
-
-        @staticmethod
-        def gettext(*args, **kwargs):
-            return _(*args, **kwargs)
-
-    dummy_form = None
-    v(dummy_form, dummy_value())
+    if not util.validate_email(value):
+        raise ValueError("Invalid email address")
 
 
 class Recipient(DB.Model):
@@ -313,7 +302,7 @@ def validate(cls, channel, value):
         for v in validators:
             try:
                 v(value)
-            except (ValidationError, TypeError) as err:
+            except (ValueError, TypeError) as err:
                 raise ValueError("Bad value: {}".format(err), err)
 
     def is_email(self):
@@ -337,7 +326,7 @@ def burry_dead(cls):
     def get_or_create(cls, channel, location):
         try:
             cls.validate(channel, location)
-        except ValidationError as err:
+        except ValueError as err:
             raise ValueError("invalid value {}: {}".format(location, err))
 
         try:

GeoHealthCheck/plugins/probe/ogcfeat.py

@@ -1,5 +1,5 @@
 from owslib.ogcapi.features import Features
-from openapi_spec_validator import openapi_v3_spec_validator
+from openapi_spec_validator import openapi_v30_spec_validator
 
 from GeoHealthCheck.probe import Probe
 from GeoHealthCheck.result import Result, push_result
@@ -339,7 +339,7 @@ def perform_request(self):
         result.start()
         try:
             # Call the openapi-spec-validator and iterate through errors
-            errors_iterator = openapi_v3_spec_validator.iter_errors(api_doc)
+            errors_iterator = openapi_v30_spec_validator.iter_errors(api_doc)
             for error in errors_iterator:
                 # Add each validation error as separate Result object
                 result = push_result(

GeoHealthCheck/util.py

@@ -30,6 +30,7 @@
 import io
 import logging
 import os
+import re
 import smtplib
 import base64
 import requests
@@ -47,6 +48,7 @@
 APP = App.get_app()
 CONFIG = App.get_config()
 LOGGER = logging.getLogger(__name__)
+EMAIL_REGEX = r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,7}"
 
 
 def average(values):
@@ -193,6 +195,12 @@ def send_email(mail_config, fromaddr, toaddr, msg):
     return True
 
 
+def validate_email(email):
+    """convenience function to validate an email address"""
+
+    return re.fullmatch(EMAIL_REGEX, email)
+
+
 def geocode(value, spatial_keyword_type='hostname'):
     """convenience function to geocode a value"""
     lat, lon = 0.0, 0.0