"""Module for validation functions."""
__copyright__ = "Copyright (c) 2020 Jina AI Limited. All rights reserved."
__license__ = "Apache-2.0"
import os
import re
import tempfile
import unicodedata
from typing import Optional
from pkg_resources import resource_stream
from ..jaml import JAML
image_tag_regex = r'^hub.[a-zA-Z_$][a-zA-Z_\s\-\.$0-9]*$'
required = {'name', 'description'}
sver_regex = (
r'^(=|>=|<=|=>|=<|>|<|!=|~|~>|\^)?(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)'
r'\.(?P<patch>0|[1-9]\d*)(?:-(?P<prerelease>(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)'
r'(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+(?P<buildmetadata>[0-9a-zA-Z-]+'
r'(?:\.[0-9a-zA-Z-]+)*))?$'
)
name_regex = r'^[a-zA-Z_$][a-zA-Z_\s\-$0-9]{2,30}$'
excepts_regex = r'\b(error|failed|FAILURES)\b'
[docs]def check_registry(registry, name, repo_prefix) -> None:
"""
Check registry image is valid within this registry.
:param registry: the registry URL
:param name: the name of the image
:param repo_prefix: the prefix of the repo
"""
if registry == 'https://index.docker.io/v1/' and not name.startswith(repo_prefix):
raise ValueError(
f'default registry only support image named with "{repo_prefix}", yours: {name}'
)
[docs]def check_name(name) -> None:
"""
Check the name is valid against the regex rule.
:param name: the name
"""
if not re.match(name_regex, name):
raise ValueError(
f'{name} is not a valid name, it should match with {name_regex}'
)
[docs]def check_version(version) -> None:
"""
Check the version against the regex.
:param version: the version
"""
if not re.match(sver_regex, version):
raise ValueError(
f'{version} is not a valid semantic version number, see http://semver.org/'
)
[docs]def check_image_name(image_name) -> None:
"""
Check the image name against the image tag regex.
:param image_name: the name of the image
"""
if not re.match(image_tag_regex, image_name):
raise ValueError(
f'{image_name} is not a valid image name for a Jina Hub image, it should match with {image_tag_regex}'
)
[docs]def check_license(lic) -> str:
"""
Check the license is a valid OS supported license.
:param lic: the license
:return: the full name of the license
"""
with resource_stream(
'jina', '/'.join(('resources', 'hub-builder', 'osi-approved.yml'))
) as fp:
approved = JAML.load(fp)
if lic not in approved:
raise ValueError(f'license {lic} is not an OSI-approved license {approved}')
return approved[lic]
[docs]def check_image_type(image_type) -> None:
"""
Check the image type is valid.
:param image_type: the type of image
"""
allowed = {'pod', 'flow', 'app'}
if image_type not in allowed:
raise ValueError(
f'type {image_type} is not allowed, should be one of {allowed}'
)
[docs]def remove_control_characters(s) -> str:
"""
Remove control characters.
:param s: the string to check
:return: the cleaned string
"""
return ''.join(ch for ch in s if unicodedata.category(ch)[0] != 'C')
[docs]def safe_url_name(url) -> str:
"""
Clean the url.
:param url: the url input
:return: the sanitized url
"""
return url.lower().replace('_', '__').replace(' ', '_')
[docs]def get_exist_path(directory, filename) -> Optional[str]:
"""
Check if path exists within the directory.
:param directory: the directory within which to check
:param filename: the filename which we check
:return: if exists, the full path
"""
r = os.path.join(directory, filename)
if os.path.exists(r):
return r
[docs]def get_summary_path(image_name: str) -> str:
"""
Get full path to summary.
:param image_name: the name of the image
:return: the full path to the summary JSON file of the image
"""
return os.path.join(
tempfile.gettempdir(), image_name.replace('/', '_') + '_summary.json'
)
[docs]def is_error_message(s) -> bool:
"""
Check if the string matches an exception regex.
:param s: the string to check
:return: whether or not it matches
"""
return re.search(excepts_regex, s, re.IGNORECASE | re.UNICODE) is not None