text:
import datetime
import inspect
import re
from operator import attrgetter
from uuid import UUID as PythonUUID

import pytz
from dateutil import parser

# Sentinel value which means "pick the default value" when encountered.
default_sentinel = object()

class ValidationError(Exception):
    pass

class StopValidation(Exception):
    pass

class Field(object):

    # If specified, the field ensures that the supplied value is an instance
    # of this type. Can be either a single specific type (e.g. int) or a tuple
    # of multiple types.
    base_type = None

    # Value to be used when there was no specific value supplied for this
    # field and the field is not required.
    blank_value = None

    def __init__(
        self,
        required=True,
        default=None,
        field_name=None,
        raw_field_name=None,
        mutable=True,
        read_only=False,
        blank_value=default_sentinel,
    ):
        """
        By default, the field name is derived from the schema model, but in
        certain cases it can be overridden. Specifying field_name overrides
        both the name of the field in the raw (unclean) data, as well as in the
        clean data model. If the raw data has a different field name than the
        clean data, raw_field_name can be overridden.
        """
        self.required = required
        self.default = default
        self.mutable = mutable
        self.field_name = field_name
        self.raw_field_name = raw_field_name or field_name
        self.read_only = read_only
        if blank_value is not default_sentinel:
            self.blank_value = blank_value

    def has_value(self, value):
        return value is not None

    def clean(self, value):
        """Take a dirty value and clean it."""
        if (
            self.base_type is not None
            and value is not None
            and not isinstance(value, self.base_type)
        ):
            if isinstance(self.base_type, tuple):
                allowed_types = [typ.__name__ for typ in self.base_type]
                allowed_types_text = ' or '.join(allowed_types)
            else:
                allowed_types_text = self.base_type.__name__
            err_msg = 'Value must be of %s type.' % allowed_types_text
            raise ValidationError(err_msg)

        if not self.has_value(value):
            if self.default is not None:
                raise StopValidation(self.default)

            if self.required:
                raise ValidationError('This field is required.')
            else:
                raise StopValidation(self.blank_value)

        return value

    def serialize(self, value):
        """
        Takes a cleaned value and serializes it.

        Keep in mind that if this field is not required, the cleaned value
        might be None.
        """
        return value

class String(Field):
    base_type = str
    blank_value = ''
    min_length = None
    max_length = None

    def __init__(self, min_length=None, max_length=None, **kwargs):
        if min_length is not None:
            self.min_length = min_length
        if max_length is not None:
            self.max_length = max_length
        super(String, self).__init__(**kwargs)

    def _check_length(self, value):
        if self.max_length is not None and len(value) > self.max_length:
            err_msg = 'The value must be no longer than %s characters.' % (
                self.max_length
            )
            raise ValidationError(err_msg)

        if self.min_length is not None and len(value) < self.min_length:
            err_msg = 'The value must be at least %s characters long.' % (
                self.min_length
            )
            raise ValidationError(err_msg)

    def clean(self, value):
        value = super(String, self).clean(value)
        self._check_length(value)
        return value

    def has_value(self, value):
        return bool(value)

class TrimmedString(String):
    base_type = str
    blank_value = ''

    def clean(self, value):
        # XXX we skip a level of inheritance so that we can perform length
        # checks *after* trimming.
        value = super(String, self).clean(value)

        if value:
            value = value.strip()
        self._check_length(value)
        return value

    def has_value(self, value):
        return bool(value and value.strip())

class Bool(Field):
    base_type = bool
    blank_value = False

class Regex(String):
    regex = None
    regex_flags = 0
    regex_message = 'Invalid input.'

    def __init__(
        self, regex=None, regex_flags=None, regex_message=None, **kwargs
    ):
        super(Regex, self).__init__(**kwargs)
        if regex is not None:
            self.regex = regex
        if regex_flags is not None:
            self.regex_flags = regex_flags
        if regex_message is not None:
            self.regex_message = regex_message

    def get_regex(self):
        if not getattr(self, '_compiled_regex', None):
            self._compiled_regex = re.compile(self.regex, self.regex_flags)
        return self._compiled_regex

    def clean(self, value):
        value = super(Regex, self).clean(value)

        if not self.get_regex().match(value):
            raise ValidationError(self.regex_message)

        return value

class DateTime(Regex):
    """ISO 8601 from http://www.pelagodesign.com/blog/2009/05/20/iso-8601-date-validation-that-doesnt-suck/"""

    regex = "^([\\+-]?\\d{4}(?!\\d{2}\\b))((-?)((0[1-9]|1[0-2])(\\3([12]\\d|0[1-9]|3[01]))?|W([0-4]\\d|5[0-2])(-?[1-7])?|(00[1-9]|0[1-9]\\d|[12]\\d{2}|3([0-5]\\d|6[1-6])))([T\\s]((([01]\\d|2[0-3])((:?)[0-5]\\d)?|24\\:?00)([\\.,]\\d+(?!:))?)?(\\17[0-5]\\d([\\.,]\\d+)?)?([zZ]|([\\+-])([01]\\d|2[0-3]):?([0-5]\\d)?)?)?)?$"
    regex_message = 'Invalid ISO 8601 datetime.'
    blank_value = None
    _force_datetime: bool

    def __init__(self, *args, force_datetime: bool = False, **kwargs):
        """
        Args:
            force_datetime: If True, always return a datetime object, even if
                the input does not specify a time.
        """
        self.min_date = kwargs.pop('min_date', None)
        self._force_datetime = force_datetime
        super(DateTime, self).__init__(*args, **kwargs)

    def clean(self, value):
        # XXX we're skipping a level of inheritance so that we can reuse
        # the regex match later in this method.
        value = super(Regex, self).clean(value)
        match = self.get_regex().match(value)
        if not match:
            raise ValidationError(self.regex_message)
        try:
            dt = parser.parse(value)
        except ValueError:
            raise ValidationError('Could not parse datetime')
        if self.min_date:
            if dt.tzinfo is not None and self.min_date.tzinfo is None:
                min_date = self.min_date.replace(tzinfo=pytz.utc)
            else:
                min_date = self.min_date
            if dt < min_date:
                err_msg = 'Date cannot be earlier than %s.' % (
                    self.min_date.strftime('%Y-%m-%d')
                )
                raise ValidationError(err_msg)

        if self._force_datetime:  # don't convert to a date
            return dt

        time_group = match.groups()[11]
        if time_group and len(time_group) > 1:
            return dt
        return dt.date()

    def serialize(self, value):
        if value is not None:
            return value.isoformat()

class Email(Regex):
    regex = (
        r'^(?:[^\.@\s]|[^\.@\s]\.(?!\.))*[^.@\s]@'
        r'[^.@\s](?:[^\.@\s]|\.(?!\.))*\.[a-z]{2,63}$'
    )
    regex_flags = re.IGNORECASE
    regex_message = 'Invalid email address.'
    max_length = 254

    def clean(self, value):
        # trim any leading/trailing whitespace before validating the email
        if isinstance(value, str):
            value = value.strip()
        return super(Email, self).clean(value)

class URL(Regex):
    blank_value = None

    def __init__(
        self,
        require_tld=True,
        default_scheme=None,
        allowed_schemes=None,
        disallowed_schemes=None,
        **kwargs
    ):
        def normalize_scheme(sch):
            if sch.endswith('://') or sch.endswith(':'):
                return sch
            return sch + '://'

        # FQDN validation similar to https://github.com/chriso/validator.js/blob/master/src/lib/isFQDN.js

        # ff01-ff5f -> full-width chars, not allowed
        alpha_numeric_and_symbols_ranges = u'0-9a-z\u00a1-\uff00\uff5f-\uffff'

        tld_part = (
            require_tld
            and r'\.[%s-]{2,63}' % alpha_numeric_and_symbols_ranges
            or ''
        )
        scheme_part = '[a-z]+://'
        self.default_scheme = default_scheme
        if self.default_scheme:
            self.default_scheme = normalize_scheme(self.default_scheme)
        self.scheme_regex = re.compile('^' + scheme_part, re.IGNORECASE)
        if default_scheme:
            scheme_part = '(%s)?' % scheme_part
        regex = (
            r'^%s([-%s@:%%_+.~#?&/\\=]{1,256}%s|([0-9]{1,3}\.){3}[0-9]{1,3})(:[0-9]+)?([/?].*)?$'
            % (scheme_part, alpha_numeric_and_symbols_ranges, tld_part)
        )
        super(URL, self).__init__(
            regex=regex,
            regex_flags=re.IGNORECASE | re.UNICODE,
            regex_message='Invalid URL.',
            **kwargs
        )

        def compile_schemes_to_regexes(schemes):
            return [
                re.compile('^' + normalize_scheme(sch) + '.*', re.IGNORECASE)
                for sch in schemes
            ]

        self.allowed_schemes = allowed_schemes or []
        self.allowed_schemes_regexes = compile_schemes_to_regexes(
            self.allowed_schemes
        )

        self.disallowed_schemes = disallowed_schemes or []
        self.disallowed_schemes_regexes = compile_schemes_to_regexes(
            self.disallowed_schemes
        )

    def clean(self, value):
        value = super(URL, self).clean(value)
        if not self.scheme_regex.match(value):
            value = self.default_scheme + value

        if self.allowed_schemes:
            if not any(
                allowed_regex.match(value)
                for allowed_regex in self.allowed_schemes_regexes
            ):
                allowed_schemes_text = ' or '.join(self.allowed_schemes)
                err_msg = (
                    "This URL uses a scheme that's not allowed. You can only "
                    "use %s." % allowed_schemes_text
                )
                raise ValidationError(err_msg)

        if self.disallowed_schemes:
            if any(
                disallowed_regex.match(value)
                for disallowed_regex in self.disallowed_schemes_regexes
            ):
                err_msg = "This URL uses a scheme that's not allowed."
                raise ValidationError(err_msg)

        return value

class RelaxedURL(URL):
    """Like URL but will just ignore values like "http://" and treat them
    as blank.
    """

    def clean(self, value):
        if not self.required and value == self.default_scheme:
            return None
        value = super(RelaxedURL, self).clean(value)
        return value

class Integer(Field):
    base_type = int

    def __init__(self, min_value=None, max_value=None, **kwargs):
        self.max_value = max_value
        self.min_value = min_value
        super(Integer, self).__init__(**kwargs)

    def _check_value(self, value):
        if self.max_value is not None and value > self.max_value:
            err_msg = 'The value must not be larger than %d.' % self.max_value
            raise ValidationError(err_msg)

        if self.min_value is not None and value < self.min_value:
            err_msg = 'The value must be at least %d.' % self.min_value
            raise ValidationError(err_msg)

    def clean(self, value):
        value = super(Integer, self).clean(value)
        self._check_value(value)
        return value

class List(Field):
    base_type = list
    blank_value = []

    def __init__(self, field_instance, max_length=None, **kwargs):
        self.max_length = max_length
        super(List, self).__init__(**kwargs)
        self.field_instance = field_instance

    def has_value(self, value):
        return bool(value)

    def clean(self, value):
        value = super(List, self).clean(value)

        item_cnt = len(value)
        if self.required and not item_cnt:
            raise ValidationError('List must not be empty.')

        if self.max_length and item_cnt > self.max_length:
            raise ValidationError('List is too long.')

        errors = {}
        data = []
        for n, item in enumerate(value):
            try:
                cleaned_data = self.field_instance.clean(item)
            except ValidationError as e:
                errors[n] = e.args and e.args[0]
            else:
                data.append(cleaned_data)

        if errors:
            raise ValidationError({'errors': errors})

        return data

    def serialize(self, value):
        # Serialize all falsy values as an empty list.
        if not value:
            return []
        return [self.field_instance.serialize(item) for item in value]

class Dict(Field):
    base_type = dict

    def has_value(self, value):
        return bool(value)

    def serialize(self, value):
        # Serialize all falsy values as an empty dict.
        return value or {}

class Embedded(Dict):
    def __init__(self, schema_class, **kwargs):
        super(Embedded, self).__init__(**kwargs)
        self.schema_class = schema_class

    def clean(self, value):
        value = super(Embedded, self).clean(value)
        try:
            cleaned_value = self.schema_class(value).full_clean()
        except ValidationError as e:
            raise e
        else:
            return cleaned_value

    def is_valid(self):
        try:
            self.clean()
        except ValidationError:
            return False
        else:
            return True

    def serialize(self, value):
        if value is not None:
            return self.schema_class(data=value).serialize()

class ReferenceNotFoundError(Exception):
    """Exception to be raised when a referenced object isn't found."""

class EmbeddedReference(Dict):
    """Represents an object which can be referenced by its ID.

    This field allows one to submit a dict of values which will then create
    a new instance of the object, or it will update fields of an existing
    object if a field representing its ID (called `pk_field`) was included
    in the submitted dict.
    """

    def __init__(self, object_class, schema_class, pk_field='id', **kwargs):
        self.object_class = object_class
        self.schema_class = schema_class
        self.pk_field = pk_field
        super(EmbeddedReference, self).__init__(**kwargs)

    def clean(self, value):
        # Clean the dict first.
        value = super(EmbeddedReference, self).clean(value)

        # Then, depending on whether `pk_field` is in the dict of submitted
        # values or not, update an existing object or create a new one.
        if value and self.pk_field in value:
            return self.clean_existing(value)
        return self.clean_new(value)

    def serialize(self, obj):
        if obj is None:
            return

        obj_data = self.get_orig_data_from_existing(obj)
        serialized = self.schema_class(data=obj_data).serialize()
        serialized[self.pk_field] = getattr(obj, self.pk_field)
        return serialized

    def clean_new(self, value):
        """Return a new object instantiated with cleaned data."""
        value = self.schema_class(value).full_clean()
        return self.object_class(**value)

    def clean_existing(self, value):
        """Clean the data and return an existing document with its fields
        updated based on the cleaned values.
        """
        existing_pk = value[self.pk_field]
        try:
            obj = self.fetch_existing(existing_pk)
        except ReferenceNotFoundError:
            raise ValidationError('Object does not exist.')
        orig_data = self.get_orig_data_from_existing(obj)

        # Clean the data (passing the new data dict and the original data to
        # the schema).
        value = self.schema_class(value, orig_data).full_clean()

        # Set cleaned data on the object (except for the pk_field).
        for field_name, field_value in value.items():
            if field_name != self.pk_field:
                setattr(obj, field_name, field_value)

        return obj

    def fetch_existing(self, pk):
        """Fetch an existing object that corresponds to a given ID.

        This needs to be subclassed since, depending on the object class,
        the fetching mechanism might be different. See implementations of
        SQLAEmbeddedReference and MongoEmbeddedReference for a concrete
        example of fetching objects from a relational and non-relational
        database.

        :param str pk: ID of the object that's supposed to exist.
        :returns: an instance of the object class.
        :raises: ReferenceNotFoundError if the object doesn't exist.
        """
        raise NotImplementedError  # should be subclassed

    def get_orig_data_from_existing(self, obj):
        """Return a dictionary of field names and values for a given object.

        The values in the dictionary should be in their "cleaned" state (as
        in, exactly as they were set on the object, without any serialization).

        :param object obj: existing object for which new data is currently
            being cleaned.
        :returns: dict of fields and values that are currently set on the
            object (before the new cleaned data is applied).
        """
        raise NotImplementedError  # should be subclassed

class Reference(Field):
    """Represents an object which can be referenced by its ID.

    This field allows one to submit an ID of an object and get a cleaned
    instance of that object. Equivalently, serialization accepts an object
    and outputs its ID.
    """

    # The ID is assumed to be supplied as a string. However, subclasses can
    # change this field if need be (e.g. it's common for objects persisted in
    # relational databases to use integers as IDs).
    base_type = str

    def __init__(self, object_class, **kwargs):
        self.object_class = object_class
        super(Reference, self).__init__(**kwargs)

    def clean(self, value):
        obj_id = super(Reference, self).clean(value)
        try:
            return self.fetch_object(obj_id)
        except ReferenceNotFoundError:
            raise ValidationError('Object does not exist.')

    def fetch_object(self, ref_id):
        """Fetch an existing object that corresponds to a given ID.

        This needs to be subclassed since, depending on the object class,
        the fetching mechanism might be different. See implementations of
        SQLAReference and MongoReference for a concrete example of fetching
        objects from a relational and non-relational database.

        :param str pk: ID of the referenced object.
        :returns: an instance of the object class.
        :raises: ReferenceNotFoundError if the object doesn't exist.
        """
        raise NotImplementedError  # should be subclassed

class Choices(Field):
    """
    A field that accepts the given choices.
    """

    def __init__(
        self,
        choices,
        case_insensitive=False,
        error_invalid_choice=None,
        **kwargs
    ):
        super(Choices, self).__init__(**kwargs)
        self.choices = choices
        self.case_insensitive = case_insensitive
        self.error_invalid_choice = (
            error_invalid_choice or 'Not a valid choice.'
        )

    def get_choices(self):
        return self.choices

    def format_invalid_choice_msg(self, value):
        return self.error_invalid_choice.format(
            value=value, valid_choices=', '.join(self.get_choices())
        )

    def clean(self, value):
        value = super(Choices, self).clean(value)

        choices = self.get_choices()

        if self.case_insensitive:
            choices = {choice.lower(): choice for choice in choices}

            if not isinstance(value, str):
                raise ValidationError(u'Value needs to be a string.')

            if value.lower() not in choices:
                err_msg = self.format_invalid_choice_msg(value)
                raise ValidationError(err_msg)

            return choices[value.lower()]

        if value not in choices:
            err_msg = self.format_invalid_choice_msg(value)
            raise ValidationError(err_msg)

        return value

class Enum(Choices):
    """Like Choices, but expects a Python 3 Enum."""

    def __init__(self, choices, **kwargs):
        """Initialize the Enum field.

        The `choices` param can be either:
        * an enum.Enum class (in which case all of its values will become
          valid choices),
        * a list containing a subset of the enum's choices (e.g.
          `[SomeEnumCls.OptionA, SomeEnumCls.OptionB]`). You must provide
          more than one choice in this list and *all* of the choices must
          belong to the same enum class.
        """
        is_cls = inspect.isclass(choices)
        if is_cls:
            self.enum_cls = choices
        else:
            assert choices, 'You need to provide at least one enum choice.'
            self.enum_cls = choices[0].__class__
        return super(Enum, self).__init__(choices, **kwargs)

    def get_choices(self):
        return [choice.value for choice in self.choices]

    def clean(self, value):
        value = super(Enum, self).clean(value)
        return self.enum_cls(value)

    def serialize(self, choice):
        if choice is not None:
            return choice.value

class SortedSet(List):
    """Sorted, unique set of values represented as a list."""

    def __init__(self, field_instance, max_length=None, key=None, **kwargs):
        super(SortedSet, self).__init__(field_instance, max_length, **kwargs)
        if isinstance(field_instance, Enum) and key is None:
            key = attrgetter('value')
        self.key = key

    def clean(self, value):
        return list(
            sorted(set(super(SortedSet, self).clean(value)), key=self.key)
        )

class Schema(object):
    """
    Base Schema class. Provides core behavior like fields declaration
    and construction, validation, and data and error proxying.

    There are 3 steps to using a Schema:

    1. Define the Schema, e.g.

        class UserSchema(Schema):
            first_name = String()
            last_name = String()
            email = Email()

    2. Create a Schema instance, passing data into it.

        # Scenario 1: Creation of a new object.
        schema = UserSchema({
            'first_name': 'Donald',
            'last_name': 'Glover',
            'email': 'gambino@example.com'
        })

        # Scenario 2: Update of an existing object.
        schema = UserSchema(
            raw_data={
                'first_name': 'Childish',
                'last_name': 'Gambino'
            },
            data={
                'first_name': 'Donald',
                'last_name': 'Glover',
                'email': 'gambino@example.com'
            }
        )

    3. Clean the Schema (validating the data you passed into it).

        data = schema.full_clean()

    This operation will raise a ValidationError if the data you passed
    into the Schema is invalid.

    To introduce custom validation to the Schema (beyond the basics
    covered by various Field types), override the "clean" method and
    raise a ValidationError with a descriptive message if you encounter
    any invalid data.

    Parameters:
    - raw_data - a dict with the data you want to validate.
    - data - dict with existing data, e.g. based on some object you're
             trying to update.
    """

    @classmethod
    def get_fields(cls):
        """
        Returns a dictionary of fields and field instances for this schema.
        """
        fields = {}
        for field_name in dir(cls):
            if isinstance(getattr(cls, field_name), Field):
                field = getattr(cls, field_name)
                field_name = field.field_name or field_name
                fields[field_name] = field
        return fields

    @classmethod
    def obj_to_dict(cls, obj):
        """
        Takes a model object and converts it into a dictionary suitable for
        passing to the constructor's data attribute.
        """
        data = {}
        for field_name in cls.get_fields():
            try:
                value = getattr(obj, field_name)
            except AttributeError:
                # If the field doesn't exist on the object, fail gracefully
                # and don't include the field in the data dict at all. Fail
                # loudly if the field exists but produces a different error
                # (edge case: accessing an *existing* field could technically
                # produce an unrelated AttributeError).
                continue

            if callable(value):
                value = value()
            data[field_name] = value

        return data

    def __init__(self, raw_data=None, data=None):
        conflicting_fields = set(
            [
                'raw_data',
                'orig_data',
                'data',
                'errors',
                'field_errors',
                'fields',
            ]
        ).intersection(dir(self))
        if conflicting_fields:
            raise Exception(
                'The following field names are reserved and need to be renamed: %s. '
                'Please use the field_name keyword to use them.'
                % list(conflicting_fields)
            )

        self.raw_data = raw_data or {}
        self.orig_data = data or None
        self.data = data and dict(data) or {}
        self.field_errors = {}
        self.errors = []
        self.fields = self.get_fields()

    def clean(self):
        """
        Override to add additional validations.

        Always called at the end of `full_clean()` method, even if validation
        failed for some fields.
        Cleaned input data are available in `self.data` dict.
        Keys that failed validation won't be set. Use conditional dictionary
        access or return early `if self.errors or self.field_errors`.
        """
        pass

    def full_clean(self):
        if not isinstance(self.raw_data, dict):
            raise ValidationError(
                {'errors': ['Invalid request: JSON dictionary expected.']}
            )

        for field_name, field in self.fields.items():
            if field.read_only:
                continue
            raw_field_name = field.raw_field_name or field_name
            try:
                # Validate a field if it's posted in raw_data, or if we don't
                # have a value for it in case it's required.
                if raw_field_name in self.raw_data or not field.has_value(
                    self.data.get(field_name, None)
                ):
                    value = field.clean(self.raw_data.get(raw_field_name))
                    if (
                        not field.mutable
                        and self.orig_data
                        and field_name in self.orig_data
                    ):

                        old_value = self.orig_data[field_name]

                        # compare datetimes properly, regardless of whether they're offset-naive or offset-aware
                        if isinstance(value, datetime.datetime) and isinstance(
                            old_value, datetime.datetime
                        ):
                            value = value.replace(tzinfo=None) + (
                                value.utcoffset()
                                or datetime.timedelta(seconds=0)
                            )
                            old_value = old_value.replace(tzinfo=None) + (
                                old_value.utcoffset()
                                or datetime.timedelta(seconds=0)
                            )

                        if value != old_value:
                            raise ValidationError('Value cannot be changed.')

                    self.data[field_name] = value

            except ValidationError as e:
                self.field_errors[raw_field_name] = e.args and e.args[0]
            except StopValidation as e:
                self.data[field_name] = e.args and e.args[0]

        try:
            self.clean()
        except ValidationError as e:
            self.errors = [e.args and e.args[0]]

        self.raise_on_errors()
        return self.data

    def raise_on_errors(self):
        if self.field_errors or self.errors:
            raise ValidationError(
                {'field-errors': self.field_errors, 'errors': self.errors}
            )

    def external_clean(self, cls, raise_on_errors=True):
        try:
            # Instantiate the external schema with the right raw_data/data.
            external_schema = cls(raw_data=self.raw_data, data=self.data)

            # Make sure its orig_data is the same as this schema's
            external_schema.orig_data = self.orig_data

            # Validate the schema and update self.data with its results.
            self.data.update(external_schema.full_clean())
        except ValidationError as e:
            self.field_errors.update(e.args[0]['field-errors'])
            self.errors += e.args[0]['errors']
            if raise_on_errors:
                self.raise_on_errors()

    def serialize(self):
        data = {}
        for field_name, field in self.fields.items():
            raw_field_name = field.raw_field_name or field_name
            value = self.data[field_name]
            data[raw_field_name] = field.serialize(value)
        return data

DEFAULT_TYPE_FIELD = 'type'

class PolymorphicField(Dict):
    """A field that can be validated with one of multiple schemas,
    depending on type provided in input values.

    Requires a dictionary-shaped value.

    Actual schema is taken from provided mapping and selected using
    a field in the input dictionary, for example:

    PolymorphicField(type_map={'my_type': MySchema})

    {'type': 'my_type', 'more': 'data', ...} will be validated with
    MySchema based on 'my_type' value.

    keep_type_field (default no) allows keeping the type-dispatching field
    in dispatched schemas' data.
    """

    base_type = dict

    def __init__(
        self,
        type_map={},
        type_field=DEFAULT_TYPE_FIELD,
        keep_type_field=False,
        *args,
        **kwargs
    ):
        super(PolymorphicField, self).__init__(*args, **kwargs)
        self.type_map = type_map
        self.type_field = type_field
        self.keep_type_field = keep_type_field

    def clean(self, value):
        clean = super(PolymorphicField, self).clean(value)
        field_type = clean.get(self.type_field)
        if field_type not in self.type_map:
            raise ValidationError(
                '{} must be one of {}'.format(
                    self.type_field, ','.join(self.type_map.keys())
                )
            )

        return self.type_map[field_type].clean(
            {
                k: v
                for k, v in value.items()
                if self.keep_type_field or k != self.type_field
            }
        )

class UUID(String):
    """Schema field for UUIDs.

    It handles deserialization from a string to a Python UUID object
    and serialization from a Python UUID object to a string.
    """

    blank_value = None

    def clean(self, value):
        value = super(UUID, self).clean(value)
        try:
            return PythonUUID(value)
        except ValueError:
            raise ValidationError('Not a UUID.')

    def serialize(self, value):
        return str(value)

class CleanDict(Dict):
    """A dictionary in which both keys and values are validated with separate schema fields."""

    def __init__(self, key_schema, value_schema, max_length=None, **kwargs):
        super(CleanDict, self).__init__(**kwargs)
        self.key_schema = key_schema
        self.value_schema = value_schema
        self.max_length = max_length

    def clean(self, value):
        value = super(CleanDict, self).clean(value)

        if self.max_length and len(value) > self.max_length:
            raise ValidationError('Dict is too long.')

        errors = {}
        data = {}
        for key, value in value.items():
            try:
                cleaned_key = self.key_schema.clean(key)
            except ValidationError as e:
                errors[key] = e.args and e.args[0]
            else:
                try:
                    cleaned_value = self.value_schema.clean(value)
                except ValidationError as e:
                    errors[key] = e.args and e.args[0]
                else:
                    data[cleaned_key] = cleaned_value

        if errors:
            raise ValidationError({'errors': errors})

        return data

    def serialize(self, value):
        # Serialize all falsy values as an empty dict.
        if not value:
            return {}
        return {
            self.key_schema.serialize(key): self.value_schema.serialize(value)
            for key, value in value.items()
        }

class EmbeddedFactory(Embedded):
    """Schema field for arbitrary embedded object.

    It generates the object using provided factory function.
    Kwargs for the factory function is the input dict, validated
    validated with schema_class (from Embedded field).
    """

    def __init__(self, factory, *args, **kwargs):
        self.factory = factory
        super(EmbeddedFactory, self).__init__(*args, **kwargs)

    def clean(self, value):
        """Clean the provided dict of values and then return an
        an object created using a factory.
        """
        value = super(EmbeddedFactory, self).clean(value)
        return self.factory(**value)

class LazyField(Field):
    """A field which is instantiated later.

    Useful for resolving circular dependencies in code.
    """

    def __init__(self, fn, *args, **kwargs):
        self.fn = fn
        self.args = args
        self.kwargs = kwargs
        self.real_field = None

    def get_real_field(self):
        if self.real_field is None:
            self.real_field = self.fn(*self.args, **self.kwargs)
            assert self.real_field
        return self.real_field

    def __getattr__(self, item):
        return getattr(self.get_real_field(), item)

    def has_value(self, value):
        return self.get_real_field().has_value(value)

    def clean(self, value):
        return self.get_real_field().clean(value)

    def serialize(self, value):
        return self.get_real_field().serialize(value)


text:
<?php if (post_password_required()) return; ?>

	<?php if (have_comments()) : ?>

<?php comments_number(); ?>

Leave a comment »

			<?php wp_list_comments(); ?>

	<?php endif; ?>

<?php comment_form(); ?>

redpajama_set_name: RedPajamaGithub


text:
package org.apache.hadoop.fs.http.client;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockStoragePolicySpi;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.fs.ContentSummary;
import org.apache.hadoop.fs.FileChecksum;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileSystemTestHelper;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.fs.contract.ContractTestUtils;
import org.apache.hadoop.fs.http.server.HttpFSServerWebApp;
import org.apache.hadoop.fs.permission.AclEntry;
import org.apache.hadoop.fs.permission.AclStatus;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.AppendTestUtil;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DFSTestUtil;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.protocol.SnapshotDiffReport;
import org.apache.hadoop.hdfs.protocol.SnapshotException;
import org.apache.hadoop.hdfs.protocol.SnapshottableDirectoryStatus;
import org.apache.hadoop.hdfs.web.JsonUtil;
import org.apache.hadoop.hdfs.web.WebHdfsFileSystem;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.test.HFSTestCase;
import org.apache.hadoop.test.HadoopUsersConfTestHelper;
import org.apache.hadoop.test.TestDir;
import org.apache.hadoop.test.TestDirHelper;
import org.apache.hadoop.test.TestHdfs;
import org.apache.hadoop.test.TestHdfsHelper;
import org.apache.hadoop.test.TestJetty;
import org.apache.hadoop.test.TestJettyHelper;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.webapp.WebAppContext;

import com.google.common.collect.Lists;

import java.io.File;
import java.io.FileOutputStream;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.Writer;
import java.net.URI;
import java.net.URL;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;

@RunWith(value = Parameterized.class)
public abstract class BaseTestHttpFSWith extends HFSTestCase {

  protected abstract Path getProxiedFSTestDir();

  protected abstract String getProxiedFSURI();

  protected abstract Configuration getProxiedFSConf();

  protected boolean isLocalFS() {
    return getProxiedFSURI().startsWith("file://");
  }

  private void createHttpFSServer() throws Exception {
    File homeDir = TestDirHelper.getTestDir();
    assertTrue(new File(homeDir, "conf").mkdir());
    assertTrue(new File(homeDir, "log").mkdir());
    assertTrue(new File(homeDir, "temp").mkdir());
    HttpFSServerWebApp.setHomeDirForCurrentThread(homeDir.getAbsolutePath());

    File secretFile = new File(new File(homeDir, "conf"), "secret");
    Writer w = new FileWriter(secretFile);
    w.write("secret");
    w.close();

    //FileSystem being served by HttpFS
    String fsDefaultName = getProxiedFSURI();
    Configuration conf = new Configuration(false);
    conf.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY, fsDefaultName);
    conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_ACLS_ENABLED_KEY, true);
    conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_XATTRS_ENABLED_KEY, true);
    // For BaseTestHttpFSWith#testFileAclsCustomizedUserAndGroupNames
    conf.set(HdfsClientConfigKeys.DFS_WEBHDFS_USER_PATTERN_KEY,
        "^[A-Za-z0-9_][A-Za-z0-9._-]*[$]?$");
    conf.set(HdfsClientConfigKeys.DFS_WEBHDFS_ACL_PERMISSION_PATTERN_KEY,
        "^(default:)?(user|group|mask|other):" +
            "[[0-9A-Za-z_][@A-Za-z0-9._-]]*:([rwx-]{3})?(,(default:)?" +
            "(user|group|mask|other):[[0-9A-Za-z_][@A-Za-z0-9._-]]*:" +
            "([rwx-]{3})?)*$");
    File hdfsSite = new File(new File(homeDir, "conf"), "hdfs-site.xml");
    OutputStream os = new FileOutputStream(hdfsSite);
    conf.writeXml(os);
    os.close();

    //HTTPFS configuration
    conf = new Configuration(false);
    conf.set("httpfs.proxyuser." + HadoopUsersConfTestHelper.getHadoopProxyUser() + ".groups",
             HadoopUsersConfTestHelper.getHadoopProxyUserGroups());
    conf.set("httpfs.proxyuser." + HadoopUsersConfTestHelper.getHadoopProxyUser() + ".hosts",
             HadoopUsersConfTestHelper.getHadoopProxyUserHosts());
    conf.set("httpfs.authentication.signature.secret.file", secretFile.getAbsolutePath());
    File httpfsSite = new File(new File(homeDir, "conf"), "httpfs-site.xml");
    os = new FileOutputStream(httpfsSite);
    conf.writeXml(os);
    os.close();

    ClassLoader cl = Thread.currentThread().getContextClassLoader();
    URL url = cl.getResource("webapp");
    WebAppContext context = new WebAppContext(url.getPath(), "/webhdfs");
    Server server = TestJettyHelper.getJettyServer();
    server.setHandler(context);
    server.start();
  }

  protected Class getFileSystemClass() {
    return HttpFSFileSystem.class;
  }

  protected String getScheme() {
    return "webhdfs";
  }

  protected FileSystem getHttpFSFileSystem(Configuration conf) throws
      Exception {
    conf.set("fs.webhdfs.impl", getFileSystemClass().getName());
    URI uri = new URI(getScheme() + "://" +
                      TestJettyHelper.getJettyURL().toURI().getAuthority());
    return FileSystem.get(uri, conf);
  }

  protected FileSystem getHttpFSFileSystem() throws Exception {
    Configuration conf = new Configuration();
    return getHttpFSFileSystem(conf);
  }

  protected void testGet() throws Exception {
    FileSystem fs = getHttpFSFileSystem();
    Assert.assertNotNull(fs);
    URI uri = new URI(getScheme() + "://" +
                      TestJettyHelper.getJettyURL().toURI().getAuthority());
    assertEquals(fs.getUri(), uri);
    fs.close();
  }

  private void testOpen() throws Exception {
    FileSystem fs = FileSystem.get(getProxiedFSConf());
    Path path = new Path(getProxiedFSTestDir(), "foo.txt");
    OutputStream os = fs.create(path);
    os.write(1);
    os.close();
    fs.close();
    fs = getHttpFSFileSystem();
    InputStream is = fs.open(new Path(path.toUri().getPath()));
    assertEquals(is.read(), 1);
    is.close();
    fs.close();
  }

  private void testCreate(Path path, boolean override) throws Exception {
    FileSystem fs = getHttpFSFileSystem();
    FsPermission permission = new FsPermission(FsAction.READ_WRITE, FsAction.NONE, FsAction.NONE);
    OutputStream os = fs.create(new Path(path.toUri().getPath()), permission, override, 1024,
                                (short) 2, 100 * 1024 * 1024, null);
    os.write(1);
    os.close();
    fs.close();

    fs = FileSystem.get(getProxiedFSConf());
    FileStatus status = fs.getFileStatus(path);
    if (!isLocalFS()) {
      assertEquals(status.getReplication(), 2);
      assertEquals(status.getBlockSize(), 100 * 1024 * 1024);
    }
    assertEquals(status.getPermission(), permission);
    InputStream is = fs.open(path);
    assertEquals(is.read(), 1);
    is.close();
    fs.close();
  }

  private void testCreate() throws Exception {
    Path path = new Path(getProxiedFSTestDir(), "foo.txt");
    FileSystem fs = FileSystem.get(getProxiedFSConf());
    fs.delete(path, true);
    testCreate(path, false);
    testCreate(path, true);
    try {
      testCreate(path, false);
      Assert.fail("the create should have failed because the file exists " +
                  "and override is FALSE");
    } catch (IOException ex) {
      System.out.println("#");
    } catch (Exception ex) {
      Assert.fail(ex.toString());
    }
  }

  private void testAppend() throws Exception {
    if (!isLocalFS()) {
      FileSystem fs = FileSystem.get(getProxiedFSConf());
      fs.mkdirs(getProxiedFSTestDir());
      Path path = new Path(getProxiedFSTestDir(), "foo.txt");
      OutputStream os = fs.create(path);
      os.write(1);
      os.close();
      fs.close();
      fs = getHttpFSFileSystem();
      os = fs.append(new Path(path.toUri().getPath()));
      os.write(2);
      os.close();
      fs.close();
      fs = FileSystem.get(getProxiedFSConf());
      InputStream is = fs.open(path);
      assertEquals(is.read(), 1);
      assertEquals(is.read(), 2);
      assertEquals(is.read(), -1);
      is.close();
      fs.close();
    }
  }

  private void testTruncate() throws Exception {
    if (!isLocalFS()) {
      final short repl = 3;
      final int blockSize = 1024;
      final int numOfBlocks = 2;
      FileSystem fs = FileSystem.get(getProxiedFSConf());
      fs.mkdirs(getProxiedFSTestDir());
      Path file = new Path(getProxiedFSTestDir(), "foo.txt");
      final byte[] data = FileSystemTestHelper.getFileData(
          numOfBlocks, blockSize);
      FileSystemTestHelper.createFile(fs, file, data, blockSize, repl);

      final int newLength = blockSize;

      boolean isReady = fs.truncate(file, newLength);
      assertTrue("Recovery is not expected.", isReady);

      FileStatus fileStatus = fs.getFileStatus(file);
      assertEquals(fileStatus.getLen(), newLength);
      AppendTestUtil.checkFullFile(fs, file, newLength, data, file.toString());

      fs.close();
    }
  }

  private void testConcat() throws Exception {
    Configuration config = getProxiedFSConf();
    config.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, 1024);
    if (!isLocalFS()) {
      FileSystem fs = FileSystem.get(config);
      fs.mkdirs(getProxiedFSTestDir());
      Path path1 = new Path("/test/foo.txt");
      Path path2 = new Path("/test/bar.txt");
      Path path3 = new Path("/test/derp.txt");
      DFSTestUtil.createFile(fs, path1, 1024, (short) 3, 0);
      DFSTestUtil.createFile(fs, path2, 1024, (short) 3, 0);
      DFSTestUtil.createFile(fs, path3, 1024, (short) 3, 0);
      fs.close();
      fs = getHttpFSFileSystem();
      fs.concat(path1, new Path[]{path2, path3});
      fs.close();
      fs = FileSystem.get(config);
      assertTrue(fs.exists(path1));
      assertFalse(fs.exists(path2));
      assertFalse(fs.exists(path3));
      fs.close();
    }
  }

  private void testRename() throws Exception {
    FileSystem fs = FileSystem.get(getProxiedFSConf());
    Path path = new Path(getProxiedFSTestDir(), "foo");
    fs.mkdirs(path);
    fs.close();
    fs = getHttpFSFileSystem();
    Path oldPath = new Path(path.toUri().getPath());
    Path newPath = new Path(path.getParent(), "bar");
    fs.rename(oldPath, newPath);
    fs.close();
    fs = FileSystem.get(getProxiedFSConf());
    assertFalse(fs.exists(oldPath));
    assertTrue(fs.exists(newPath));
    fs.close();
  }

  private void testDelete() throws Exception {
    Path foo = new Path(getProxiedFSTestDir(), "foo");
    Path bar = new Path(getProxiedFSTestDir(), "bar");
    Path foe = new Path(getProxiedFSTestDir(), "foe");
    FileSystem fs = FileSystem.get(getProxiedFSConf());
    fs.mkdirs(foo);
    fs.mkdirs(new Path(bar, "a"));
    fs.mkdirs(foe);

    FileSystem hoopFs = getHttpFSFileSystem();
    assertTrue(hoopFs.delete(new Path(foo.toUri().getPath()), false));
    assertFalse(fs.exists(foo));
    try {
      hoopFs.delete(new Path(bar.toUri().getPath()), false);
      Assert.fail();
    } catch (IOException ex) {
    } catch (Exception ex) {
      Assert.fail();
    }
    assertTrue(fs.exists(bar));
    assertTrue(hoopFs.delete(new Path(bar.toUri().getPath()), true));
    assertFalse(fs.exists(bar));

    assertTrue(fs.exists(foe));
    assertTrue(hoopFs.delete(foe, true));
    assertFalse(fs.exists(foe));

    hoopFs.close();
    fs.close();
  }

  private void testListStatus() throws Exception {
    FileSystem fs = FileSystem.get(getProxiedFSConf());
    Path path = new Path(getProxiedFSTestDir(), "foo.txt");
    OutputStream os = fs.create(path);
    os.write(1);
    os.close();
    FileStatus status1 = fs.getFileStatus(path);
    fs.close();

    fs = getHttpFSFileSystem();
    FileStatus status2 = fs.getFileStatus(new Path(path.toUri().getPath()));
    fs.close();

    assertEquals(status2.getPermission(), status1.getPermission());
    assertEquals(status2.getPath().toUri().getPath(),
        status1.getPath().toUri().getPath());
    assertEquals(status2.getReplication(), status1.getReplication());
    assertEquals(status2.getBlockSize(), status1.getBlockSize());
    assertEquals(status2.getAccessTime(), status1.getAccessTime());
    assertEquals(status2.getModificationTime(), status1.getModificationTime());
    assertEquals(status2.getOwner(), status1.getOwner());
    assertEquals(status2.getGroup(), status1.getGroup());
    assertEquals(status2.getLen(), status1.getLen());

    FileStatus[] stati = fs.listStatus(path.getParent());
    assertEquals(1, stati.length);
    assertEquals(stati[0].getPath().getName(), path.getName());

    // The full path should be the path to the file. See HDFS-12139
    FileStatus[] statl = fs.listStatus(path);
    Assert.assertEquals(1, statl.length);
    Assert.assertEquals(status2.getPath(), statl[0].getPath());
    Assert.assertEquals(statl[0].getPath().getName(), path.getName());
    Assert.assertEquals(stati[0].getPath(), statl[0].getPath());
  }

  private void testFileStatusAttr() throws Exception {
    if (!this.isLocalFS()) {
      // Create a directory
      Path path = new Path("/tmp/tmp-snap-test");
      DistributedFileSystem distributedFs = (DistributedFileSystem) FileSystem
          .get(path.toUri(), this.getProxiedFSConf());
      distributedFs.mkdirs(path);
      // Get the FileSystem instance that's being tested
      FileSystem fs = this.getHttpFSFileSystem();
      // Check FileStatus
      assertFalse("Snapshot should be disallowed by default",
          fs.getFileStatus(path).isSnapshotEnabled());
      // Allow snapshot
      distributedFs.allowSnapshot(path);
      // Check FileStatus
      assertTrue("Snapshot enabled bit is not set in FileStatus",
          fs.getFileStatus(path).isSnapshotEnabled());
      // Disallow snapshot
      distributedFs.disallowSnapshot(path);
      // Check FileStatus
      assertFalse("Snapshot enabled bit is not cleared in FileStatus",
          fs.getFileStatus(path).isSnapshotEnabled());
      // Cleanup
      fs.delete(path, true);
      fs.close();
      distributedFs.close();
    }
  }

  private static void assertSameListing(FileSystem expected, FileSystem
      actual, Path p) throws IOException {
    // Consume all the entries from both iterators
    RemoteIterator<FileStatus> exIt = expected.listStatusIterator(p);
    List<FileStatus> exStatuses = new ArrayList<>();
    while (exIt.hasNext()) {
      exStatuses.add(exIt.next());
    }
    RemoteIterator<FileStatus> acIt = actual.listStatusIterator(p);
    List<FileStatus> acStatuses = new ArrayList<>();
    while (acIt.hasNext()) {
      acStatuses.add(acIt.next());
    }
    assertEquals(exStatuses.size(), acStatuses.size());
    for (int i = 0; i < exStatuses.size(); i++) {
      FileStatus expectedStatus = exStatuses.get(i);
      FileStatus actualStatus = acStatuses.get(i);
      // Path URIs are fully qualified, so compare just the path component
      assertEquals(expectedStatus.getPath().toUri().getPath(),
          actualStatus.getPath().toUri().getPath());
    }
  }

  private void testListStatusBatch() throws Exception {
    // LocalFileSystem writes checksum files next to the data files, which
    // show up when listing via LFS. This makes the listings not compare
    // properly.
    Assume.assumeFalse(isLocalFS());

    FileSystem proxyFs = FileSystem.get(getProxiedFSConf());
    Configuration conf = new Configuration();
    conf.setInt(DFSConfigKeys.DFS_LIST_LIMIT, 2);
    FileSystem httpFs = getHttpFSFileSystem(conf);

    // Test an empty directory
    Path dir = new Path(getProxiedFSTestDir(), "dir");
    proxyFs.mkdirs(dir);
    assertSameListing(proxyFs, httpFs, dir);
    // Create and test in a loop
    for (int i = 0; i < 10; i++) {
      proxyFs.create(new Path(dir, "file" + i)).close();
      assertSameListing(proxyFs, httpFs, dir);
    }

    // Test for HDFS-12139
    Path dir1 = new Path(getProxiedFSTestDir(), "dir1");
    proxyFs.mkdirs(dir1);
    Path file1 = new Path(dir1, "file1");
    proxyFs.create(file1).close();

    RemoteIterator<FileStatus> si = proxyFs.listStatusIterator(dir1);
    FileStatus statusl = si.next();
    FileStatus status = proxyFs.getFileStatus(file1);
    Assert.assertEquals(file1.getName(), statusl.getPath().getName());
    Assert.assertEquals(status.getPath(), statusl.getPath());

    si = proxyFs.listStatusIterator(file1);
    statusl = si.next();
    Assert.assertEquals(file1.getName(), statusl.getPath().getName());
    Assert.assertEquals(status.getPath(), statusl.getPath());
  }

  private void testWorkingdirectory() throws Exception {
    FileSystem fs = FileSystem.get(getProxiedFSConf());
    Path workingDir = fs.getWorkingDirectory();
    fs.close();

    fs = getHttpFSFileSystem();
    if (isLocalFS()) {
      fs.setWorkingDirectory(workingDir);
    }
    Path httpFSWorkingDir = fs.getWorkingDirectory();
    fs.close();
    assertEquals(httpFSWorkingDir.toUri().getPath(),
                        workingDir.toUri().getPath());

    fs = getHttpFSFileSystem();
    fs.setWorkingDirectory(new Path("/tmp"));
    workingDir = fs.getWorkingDirectory();
    fs.close();
    assertEquals(workingDir.toUri().getPath(),
        new Path("/tmp").toUri().getPath());
  }

  private void testTrashRoot() throws Exception {
    if (!isLocalFS()) {
      FileSystem fs = FileSystem.get(getProxiedFSConf());

      final Path rootDir = new Path("/");
      final Path fooPath = new Path(getProxiedFSTestDir(), "foo.txt");
      OutputStream os = fs.create(fooPath);
      os.write(1);
      os.close();

      Path trashPath = fs.getTrashRoot(rootDir);
      Path fooTrashPath = fs.getTrashRoot(fooPath);
      fs.close();

      fs = getHttpFSFileSystem();
      Path httpFSTrashPath = fs.getTrashRoot(rootDir);
      Path httpFSFooTrashPath = fs.getTrashRoot(fooPath);
      fs.close();

      assertEquals(trashPath.toUri().getPath(),
          httpFSTrashPath.toUri().getPath());
      assertEquals(fooTrashPath.toUri().getPath(),
          httpFSFooTrashPath.toUri().getPath());
      // trash path is related to USER, not path
      assertEquals(trashPath.toUri().getPath(),
          fooTrashPath.toUri().getPath());
    }
  }

  private void testMkdirs() throws Exception {
    Path path = new Path(getProxiedFSTestDir(), "foo");
    FileSystem fs = getHttpFSFileSystem();
    fs.mkdirs(path);
    fs.close();
    fs = FileSystem.get(getProxiedFSConf());
    assertTrue(fs.exists(path));
    fs.close();
  }

  private void testSetTimes() throws Exception {
    if (!isLocalFS()) {
      FileSystem fs = FileSystem.get(getProxiedFSConf());
      Path path = new Path(getProxiedFSTestDir(), "foo.txt");
      OutputStream os = fs.create(path);
      os.write(1);
      os.close();
      FileStatus status1 = fs.getFileStatus(path);
      fs.close();
      long at = status1.getAccessTime();
      long mt = status1.getModificationTime();

      fs = getHttpFSFileSystem();
      fs.setTimes(path, mt - 10, at - 20);
      fs.close();

      fs = FileSystem.get(getProxiedFSConf());
      status1 = fs.getFileStatus(path);
      fs.close();
      long atNew = status1.getAccessTime();
      long mtNew = status1.getModificationTime();
      assertEquals(mtNew, mt - 10);
      assertEquals(atNew, at - 20);
    }
  }

  protected void testSetPermission() throws Exception {
    FileSystem fs = FileSystem.get(getProxiedFSConf());
    Path path = new Path(getProxiedFSTestDir(), "foodir");
    fs.mkdirs(path);

    fs = getHttpFSFileSystem();
    FsPermission permission1 = new FsPermission(FsAction.READ_WRITE, FsAction.NONE, FsAction.NONE);
    fs.setPermission(path, permission1);
    fs.close();

    fs = FileSystem.get(getProxiedFSConf());
    FileStatus status1 = fs.getFileStatus(path);
    fs.close();
    FsPermission permission2 = status1.getPermission();
    assertEquals(permission2, permission1);

    //sticky bit
    fs = getHttpFSFileSystem();
    permission1 = new FsPermission(FsAction.READ_WRITE, FsAction.NONE, FsAction.NONE, true);
    fs.setPermission(path, permission1);
    fs.close();

    fs = FileSystem.get(getProxiedFSConf());
    status1 = fs.getFileStatus(path);
    fs.close();
    permission2 = status1.getPermission();
    assertTrue(permission2.getStickyBit());
    assertEquals(permission2, permission1);
  }

  private void testSetOwner() throws Exception {
    if (!isLocalFS()) {
      FileSystem fs = FileSystem.get(getProxiedFSConf());
      fs.mkdirs(getProxiedFSTestDir());
      Path path = new Path(getProxiedFSTestDir(), "foo.txt");
      OutputStream os = fs.create(path);
      os.write(1);
      os.close();
      fs.close();

      fs = getHttpFSFileSystem();
      String user = HadoopUsersConfTestHelper.getHadoopUsers()[1];
      String group = HadoopUsersConfTestHelper.getHadoopUserGroups(user)[0];
      fs.setOwner(path, user, group);
      fs.close();

      fs = FileSystem.get(getProxiedFSConf());
      FileStatus status1 = fs.getFileStatus(path);
      fs.close();
      assertEquals(status1.getOwner(), user);
      assertEquals(status1.getGroup(), group);
    }
  }

  private void testSetReplication() throws Exception {
    FileSystem fs = FileSystem.get(getProxiedFSConf());
    Path path = new Path(getProxiedFSTestDir(), "foo.txt");
    OutputStream os = fs.create(path);
    os.write(1);
    os.close();
    fs.setReplication(path, (short) 2);
    fs.close();

    fs = getHttpFSFileSystem();
    fs.setReplication(path, (short) 1);
    fs.close();

    fs = FileSystem.get(getProxiedFSConf());
    FileStatus status1 = fs.getFileStatus(path);
    fs.close();
    assertEquals(status1.getReplication(), (short) 1);
  }

  private void testChecksum() throws Exception {
    if (!isLocalFS()) {
      FileSystem fs = FileSystem.get(getProxiedFSConf());
      fs.mkdirs(getProxiedFSTestDir());
      Path path = new Path(getProxiedFSTestDir(), "foo.txt");
      OutputStream os = fs.create(path);
      os.write(1);
      os.close();
      FileChecksum hdfsChecksum = fs.getFileChecksum(path);
      fs.close();
      fs = getHttpFSFileSystem();
      FileChecksum httpChecksum = fs.getFileChecksum(path);
      fs.close();
      assertEquals(httpChecksum.getAlgorithmName(),
          hdfsChecksum.getAlgorithmName());
      assertEquals(httpChecksum.getLength(), hdfsChecksum.getLength());
      assertArrayEquals(httpChecksum.getBytes(), hdfsChecksum.getBytes());
    }
  }

  private void testContentSummary() throws Exception {
    FileSystem fs = FileSystem.get(getProxiedFSConf());
    Path path = new Path(getProxiedFSTestDir(), "foo.txt");
    OutputStream os = fs.create(path);
    os.write(1);
    os.close();
    ContentSummary hdfsContentSummary = fs.getContentSummary(path);
    fs.close();
    fs = getHttpFSFileSystem();
    ContentSummary httpContentSummary = fs.getContentSummary(path);
    fs.close();
    assertEquals(httpContentSummary.getDirectoryCount(),
        hdfsContentSummary.getDirectoryCount());
    assertEquals(httpContentSummary.getFileCount(),
        hdfsContentSummary.getFileCount());
    assertEquals(httpContentSummary.getLength(),
        hdfsContentSummary.getLength());
    assertEquals(httpContentSummary.getQuota(), hdfsContentSummary.getQuota());
    assertEquals(httpContentSummary.getSpaceConsumed(),
        hdfsContentSummary.getSpaceConsumed());
    assertEquals(httpContentSummary.getSpaceQuota(),
        hdfsContentSummary.getSpaceQuota());
  }

  /** Set xattr */
  private void testSetXAttr() throws Exception {
    if (!isLocalFS()) {
      FileSystem fs = FileSystem.get(getProxiedFSConf());
      fs.mkdirs(getProxiedFSTestDir());
      Path path = new Path(getProxiedFSTestDir(), "foo.txt");
      OutputStream os = fs.create(path);
      os.write(1);
      os.close();
      fs.close();

      final String name1 = "user.a1";
      final byte[] value1 = new byte[]{0x31, 0x32, 0x33};
      final String name2 = "user.a2";
      final byte[] value2 = new byte[]{0x41, 0x42, 0x43};
      final String name3 = "user.a3";
      final byte[] value3 = null;
      final String name4 = "trusted.a1";
      final byte[] value4 = new byte[]{0x31, 0x32, 0x33};
      final String name5 = "a1";
      fs = getHttpFSFileSystem();
      fs.setXAttr(path, name1, value1);
      fs.setXAttr(path, name2, value2);
      fs.setXAttr(path, name3, value3);
      fs.setXAttr(path, name4, value4);
      try {
        fs.setXAttr(path, name5, value1);
        Assert.fail("Set xAttr with incorrect name format should fail.");
      } catch (IOException e) {
      } catch (IllegalArgumentException e) {
      }
      fs.close();

      fs = FileSystem.get(getProxiedFSConf());
      Map<String, byte[]> xAttrs = fs.getXAttrs(path);
      fs.close();
      assertEquals(4, xAttrs.size());
      assertArrayEquals(value1, xAttrs.get(name1));
      assertArrayEquals(value2, xAttrs.get(name2));
      assertArrayEquals(new byte[0], xAttrs.get(name3));
      assertArrayEquals(value4, xAttrs.get(name4));
    }
  }

  /** Get xattrs */
  private void testGetXAttrs() throws Exception {
    if (!isLocalFS()) {
      FileSystem fs = FileSystem.get(getProxiedFSConf());
      fs.mkdirs(getProxiedFSTestDir());
      Path path = new Path(getProxiedFSTestDir(), "foo.txt");
      OutputStream os = fs.create(path);
      os.write(1);
      os.close();
      fs.close();

      final String name1 = "user.a1";
      final byte[] value1 = new byte[]{0x31, 0x32, 0x33};
      final String name2 = "user.a2";
      final byte[] value2 = new byte[]{0x41, 0x42, 0x43};
      final String name3 = "user.a3";
      final byte[] value3 = null;
      final String name4 = "trusted.a1";
      final byte[] value4 = new byte[]{0x31, 0x32, 0x33};
      fs = FileSystem.get(getProxiedFSConf());
      fs.setXAttr(path, name1, value1);
      fs.setXAttr(path, name2, value2);
      fs.setXAttr(path, name3, value3);
      fs.setXAttr(path, name4, value4);
      fs.close();

      // Get xattrs with names parameter
      fs = getHttpFSFileSystem();
      List<String> names = Lists.newArrayList();
      names.add(name1);
      names.add(name2);
      names.add(name3);
      names.add(name4);
      Map<String, byte[]> xAttrs = fs.getXAttrs(path, names);
      fs.close();
      assertEquals(4, xAttrs.size());
      assertArrayEquals(value1, xAttrs.get(name1));
      assertArrayEquals(value2, xAttrs.get(name2));
      assertArrayEquals(new byte[0], xAttrs.get(name3));
      assertArrayEquals(value4, xAttrs.get(name4));

      // Get specific xattr
      fs = getHttpFSFileSystem();
      byte[] value = fs.getXAttr(path, name1);
      assertArrayEquals(value1, value);
      final String name5 = "a1";
      try {
        value = fs.getXAttr(path, name5);
        Assert.fail("Get xAttr with incorrect name format should fail.");
      } catch (IOException e) {
      } catch (IllegalArgumentException e) {
      }
      fs.close();

      // Get all xattrs
      fs = getHttpFSFileSystem();
      xAttrs = fs.getXAttrs(path);
      fs.close();
      assertEquals(4, xAttrs.size());
      assertArrayEquals(value1, xAttrs.get(name1));
      assertArrayEquals(value2, xAttrs.get(name2));
      assertArrayEquals(new byte[0], xAttrs.get(name3));
      assertArrayEquals(value4, xAttrs.get(name4));
    }
  }

  /** Remove xattr */
  private void testRemoveXAttr() throws Exception {
    if (!isLocalFS()) {
      FileSystem fs = FileSystem.get(getProxiedFSConf());
      fs.mkdirs(getProxiedFSTestDir());
      Path path = new Path(getProxiedFSTestDir(), "foo.txt");
      OutputStream os = fs.create(path);
      os.write(1);
      os.close();
      fs.close();

      final String name1 = "user.a1";
      final byte[] value1 = new byte[]{0x31, 0x32, 0x33};
      final String name2 = "user.a2";
      final byte[] value2 = new byte[]{0x41, 0x42, 0x43};
      final String name3 = "user.a3";
      final byte[] value3 = null;
      final String name4 = "trusted.a1";
      final byte[] value4 = new byte[]{0x31, 0x32, 0x33};
      final String name5 = "a1";
      fs = FileSystem.get(getProxiedFSConf());
      fs.setXAttr(path, name1, value1);
      fs.setXAttr(path, name2, value2);
      fs.setXAttr(path, name3, value3);
      fs.setXAttr(path, name4, value4);
      fs.close();

      fs = getHttpFSFileSystem();
      fs.removeXAttr(path, name1);
      fs.removeXAttr(path, name3);
      fs.removeXAttr(path, name4);
      try {
        fs.removeXAttr(path, name5);
        Assert.fail("Remove xAttr with incorrect name format should fail.");
      } catch (IOException e) {
      } catch (IllegalArgumentException e) {
      }

      fs = FileSystem.get(getProxiedFSConf());
      Map<String, byte[]> xAttrs = fs.getXAttrs(path);
      fs.close();
      assertEquals(1, xAttrs.size());
      assertArrayEquals(value2, xAttrs.get(name2));
    }
  }

  /** List xattrs */
  private void testListXAttrs() throws Exception {
    if (!isLocalFS()) {
      FileSystem fs = FileSystem.get(getProxiedFSConf());
      fs.mkdirs(getProxiedFSTestDir());
      Path path = new Path(getProxiedFSTestDir(), "foo.txt");
      OutputStream os = fs.create(path);
      os.write(1);
      os.close();
      fs.close();

      final String name1 = "user.a1";
      final byte[] value1 = new byte[]{0x31, 0x32, 0x33};
      final String name2 = "user.a2";
      final byte[] value2 = new byte[]{0x41, 0x42, 0x43};
      final String name3 = "user.a3";
      final byte[] value3 = null;
      final String name4 = "trusted.a1";
      final byte[] value4 = new byte[]{0x31, 0x32, 0x33};
      fs = FileSystem.get(getProxiedFSConf());
      fs.setXAttr(path, name1, value1);
      fs.setXAttr(path, name2, value2);
      fs.setXAttr(path, name3, value3);
      fs.setXAttr(path, name4, value4);
      fs.close();

      fs = getHttpFSFileSystem();
      List<String> names = fs.listXAttrs(path);
      assertEquals(4, names.size());
      assertTrue(names.contains(name1));
      assertTrue(names.contains(name2));
      assertTrue(names.contains(name3));
      assertTrue(names.contains(name4));
    }
  }

  /**
   * Runs assertions testing that two AclStatus objects contain the same info
   * @param a First AclStatus
   * @param b Second AclStatus
   * @throws Exception
   */
  private void assertSameAcls(AclStatus a, AclStatus b) throws Exception {
    assertTrue(a.getOwner().equals(b.getOwner()));
    assertTrue(a.getGroup().equals(b.getGroup()));
    assertTrue(a.isStickyBit() == b.isStickyBit());
    assertTrue(a.getEntries().size() == b.getEntries().size());
    for (AclEntry e : a.getEntries()) {
      assertTrue(b.getEntries().contains(e));
    }
    for (AclEntry e : b.getEntries()) {
      assertTrue(a.getEntries().contains(e));
    }
  }

  private static void assertSameAcls(FileSystem expected, FileSystem actual,
      Path path) throws IOException {
    FileStatus expectedFileStatus = expected.getFileStatus(path);
    FileStatus actualFileStatus = actual.getFileStatus(path);
    assertEquals(actualFileStatus.hasAcl(), expectedFileStatus.hasAcl());
    // backwards compat
    assertEquals(actualFileStatus.getPermission().getAclBit(),
        expectedFileStatus.getPermission().getAclBit());
  }

  /**
   * Simple ACL tests on a file:  Set an acl, add an acl, remove one acl,
   * and remove all acls.
   * @throws Exception
   */
  private void testFileAcls() throws Exception {
    if ( isLocalFS() ) {
      return;
    }

    final String aclUser1 = "user:foo:rw-";
    final String rmAclUser1 = "user:foo:";
    final String aclUser2 = "user:bar:r--";
    final String aclGroup1 = "group::r--";
    final String aclSet = "user::rwx," + aclUser1 + ","
            + aclGroup1 + ",other::---";

    FileSystem proxyFs = FileSystem.get(getProxiedFSConf());
    FileSystem httpfs = getHttpFSFileSystem();

    Path path = new Path(getProxiedFSTestDir(), "testAclStatus.txt");
    OutputStream os = proxyFs.create(path);
    os.write(1);
    os.close();

    AclStatus proxyAclStat = proxyFs.getAclStatus(path);
    AclStatus httpfsAclStat = httpfs.getAclStatus(path);
    assertSameAcls(httpfsAclStat, proxyAclStat);
    assertSameAcls(httpfs, proxyFs, path);

    httpfs.setAcl(path, AclEntry.parseAclSpec(aclSet,true));
    proxyAclStat = proxyFs.getAclStatus(path);
    httpfsAclStat = httpfs.getAclStatus(path);
    assertSameAcls(httpfsAclStat, proxyAclStat);
    assertSameAcls(httpfs, proxyFs, path);

    httpfs.modifyAclEntries(path, AclEntry.parseAclSpec(aclUser2, true));
    proxyAclStat = proxyFs.getAclStatus(path);
    httpfsAclStat = httpfs.getAclStatus(path);
    assertSameAcls(httpfsAclStat, proxyAclStat);
    assertSameAcls(httpfs, proxyFs, path);

    httpfs.removeAclEntries(path, AclEntry.parseAclSpec(rmAclUser1, false));
    proxyAclStat = proxyFs.getAclStatus(path);
    httpfsAclStat = httpfs.getAclStatus(path);
    assertSameAcls(httpfsAclStat, proxyAclStat);
    assertSameAcls(httpfs, proxyFs, path);

    httpfs.removeAcl(path);
    proxyAclStat = proxyFs.getAclStatus(path);
    httpfsAclStat = httpfs.getAclStatus(path);
    assertSameAcls(httpfsAclStat, proxyAclStat);
    assertSameAcls(httpfs, proxyFs, path);
  }

  /**
   * Simple acl tests on a directory: set a default acl, remove default acls.
   * @throws Exception
   */
  private void testDirAcls() throws Exception {
    if ( isLocalFS() ) {
      return;
    }

    final String defUser1 = "default:user:glarch:r-x";

    FileSystem proxyFs = FileSystem.get(getProxiedFSConf());
    FileSystem httpfs = getHttpFSFileSystem();

    Path dir = getProxiedFSTestDir();

    /* ACL Status on a directory */
    AclStatus proxyAclStat = proxyFs.getAclStatus(dir);
    AclStatus httpfsAclStat = httpfs.getAclStatus(dir);
    assertSameAcls(httpfsAclStat, proxyAclStat);
    assertSameAcls(httpfs, proxyFs, dir);

    /* Set a default ACL on the directory */
    httpfs.setAcl(dir, (AclEntry.parseAclSpec(defUser1,true)));
    proxyAclStat = proxyFs.getAclStatus(dir);
    httpfsAclStat = httpfs.getAclStatus(dir);
    assertSameAcls(httpfsAclStat, proxyAclStat);
    assertSameAcls(httpfs, proxyFs, dir);

    /* Remove the default ACL */
    httpfs.removeDefaultAcl(dir);
    proxyAclStat = proxyFs.getAclStatus(dir);
    httpfsAclStat = httpfs.getAclStatus(dir);
    assertSameAcls(httpfsAclStat, proxyAclStat);
    assertSameAcls(httpfs, proxyFs, dir);
  }

  private void testEncryption() throws Exception {
    if (isLocalFS()) {
      return;
    }
    FileSystem proxyFs = FileSystem.get(getProxiedFSConf());
    FileSystem httpFs = getHttpFSFileSystem();
    FileStatus proxyStatus = proxyFs.getFileStatus(TestHdfsHelper
        .ENCRYPTED_FILE);
    assertTrue(proxyStatus.isEncrypted());
    FileStatus httpStatus = httpFs.getFileStatus(TestHdfsHelper
        .ENCRYPTED_FILE);
    assertTrue(httpStatus.isEncrypted());
    proxyStatus = proxyFs.getFileStatus(new Path("/"));
    httpStatus = httpFs.getFileStatus(new Path("/"));
    assertFalse(proxyStatus.isEncrypted());
    assertFalse(httpStatus.isEncrypted());
  }

  private void testErasureCoding() throws Exception {
    Assume.assumeFalse("Assume its not a local FS!", isLocalFS());
    FileSystem proxyFs = FileSystem.get(getProxiedFSConf());
    FileSystem httpFS = getHttpFSFileSystem();
    Path filePath = new Path(getProxiedFSTestDir(), "foo.txt");
    proxyFs.create(filePath).close();

    ContractTestUtils.assertNotErasureCoded(httpFS, getProxiedFSTestDir());
    ContractTestUtils.assertNotErasureCoded(httpFS, filePath);
    ContractTestUtils.assertErasureCoded(httpFS,
        TestHdfsHelper.ERASURE_CODING_DIR);
    ContractTestUtils.assertErasureCoded(httpFS,
        TestHdfsHelper.ERASURE_CODING_FILE);

    proxyFs.close();
    httpFS.close();
  }

  private void testStoragePolicy() throws Exception {
    Assume.assumeFalse("Assume its not a local FS", isLocalFS());
    FileSystem fs = FileSystem.get(getProxiedFSConf());
    fs.mkdirs(getProxiedFSTestDir());
    Path path = new Path(getProxiedFSTestDir(), "policy.txt");
    FileSystem httpfs = getHttpFSFileSystem();
    // test getAllStoragePolicies
    Assert.assertArrayEquals(
        "Policy array returned from the DFS and HttpFS should be equals",
        fs.getAllStoragePolicies().toArray(), httpfs.getAllStoragePolicies().toArray());

    // test get/set/unset policies
    DFSTestUtil.createFile(fs, path, 0, (short) 1, 0L);
    // get defaultPolicy
   BlockStoragePolicySpi defaultdfsPolicy = fs.getStoragePolicy(path);
    // set policy through webhdfs
    httpfs.setStoragePolicy(path, HdfsConstants.COLD_STORAGE_POLICY_NAME);
    // get policy from dfs
    BlockStoragePolicySpi dfsPolicy = fs.getStoragePolicy(path);
    // get policy from webhdfs
    BlockStoragePolicySpi httpFsPolicy = httpfs.getStoragePolicy(path);
    Assert
       .assertEquals(
            "Storage policy returned from the get API should"
            + " be same as set policy",
            HdfsConstants.COLD_STORAGE_POLICY_NAME.toString(),
            httpFsPolicy.getName());
    Assert.assertEquals(
        "Storage policy returned from the DFS and HttpFS should be equals",
        httpFsPolicy, dfsPolicy);
    // unset policy
    httpfs.unsetStoragePolicy(path);
    Assert
       .assertEquals(
            "After unset storage policy, the get API shoudld"
            + " return the default policy",
            defaultdfsPolicy, httpfs.getStoragePolicy(path));
    fs.close();
  }

  protected enum Operation {
    GET, OPEN, CREATE, APPEND, TRUNCATE, CONCAT, RENAME, DELETE, LIST_STATUS,
    WORKING_DIRECTORY, MKDIRS, SET_TIMES, SET_PERMISSION, SET_OWNER,
    SET_REPLICATION, CHECKSUM, CONTENT_SUMMARY, FILEACLS, DIRACLS, SET_XATTR,
    GET_XATTRS, REMOVE_XATTR, LIST_XATTRS, ENCRYPTION, LIST_STATUS_BATCH,
    GETTRASHROOT, STORAGEPOLICY, ERASURE_CODING,
    CREATE_SNAPSHOT, RENAME_SNAPSHOT, DELETE_SNAPSHOT,
    ALLOW_SNAPSHOT, DISALLOW_SNAPSHOT, DISALLOW_SNAPSHOT_EXCEPTION,
    FILE_STATUS_ATTR, GET_SNAPSHOT_DIFF, GET_SNAPSHOTTABLE_DIRECTORY_LIST
  }

  private void operation(Operation op) throws Exception {
    switch (op) {
    case GET:
      testGet();
      break;
    case OPEN:
      testOpen();
      break;
    case CREATE:
      testCreate();
      break;
    case APPEND:
      testAppend();
      break;
    case TRUNCATE:
      testTruncate();
      break;
    case CONCAT:
      testConcat();
      break;
    case RENAME:
      testRename();
      break;
    case DELETE:
      testDelete();
      break;
    case LIST_STATUS:
      testListStatus();
      break;
    case WORKING_DIRECTORY:
      testWorkingdirectory();
      break;
    case MKDIRS:
      testMkdirs();
      break;
    case SET_TIMES:
      testSetTimes();
      break;
    case SET_PERMISSION:
      testSetPermission();
      break;
    case SET_OWNER:
      testSetOwner();
      break;
    case SET_REPLICATION:
      testSetReplication();
      break;
    case CHECKSUM:
      testChecksum();
      break;
    case CONTENT_SUMMARY:
      testContentSummary();
      break;
    case FILEACLS:
      testFileAclsCustomizedUserAndGroupNames();
      testFileAcls();
      break;
    case DIRACLS:
      testDirAcls();
      break;
    case SET_XATTR:
      testSetXAttr();
      break;
    case REMOVE_XATTR:
      testRemoveXAttr();
      break;
    case GET_XATTRS:
      testGetXAttrs();
      break;
    case LIST_XATTRS:
      testListXAttrs();
      break;
    case ENCRYPTION:
      testEncryption();
      break;
    case LIST_STATUS_BATCH:
      testListStatusBatch();
      break;
    case GETTRASHROOT:
      testTrashRoot();
      break;
    case STORAGEPOLICY:
      testStoragePolicy();
      break;
    case ERASURE_CODING:
      testErasureCoding();
      break;
    case CREATE_SNAPSHOT:
      testCreateSnapshot();
      break;
    case RENAME_SNAPSHOT:
      testRenameSnapshot();
      break;
    case DELETE_SNAPSHOT:
      testDeleteSnapshot();
      break;
    case ALLOW_SNAPSHOT:
      testAllowSnapshot();
      break;
    case DISALLOW_SNAPSHOT:
      testDisallowSnapshot();
      break;
    case DISALLOW_SNAPSHOT_EXCEPTION:
      testDisallowSnapshotException();
      break;
    case FILE_STATUS_ATTR:
      testFileStatusAttr();
      break;
    case GET_SNAPSHOT_DIFF:
      testGetSnapshotDiff();
      testGetSnapshotDiffIllegalParam();
      break;
    case GET_SNAPSHOTTABLE_DIRECTORY_LIST:
      testGetSnapshottableDirListing();
      break;
    }
  }

  @Parameterized.Parameters
  public static Collection operations() {
    Object[][] ops = new Object[Operation.values().length][];
    for (int i = 0; i < Operation.values().length; i++) {
      ops[i] = new Object[]{Operation.values()[i]};
    }
    //To test one or a subset of operations do:
    //return Arrays.asList(new Object[][]{ new Object[]{Operation.APPEND}});
    return Arrays.asList(ops);
  }

  private Operation operation;

  public BaseTestHttpFSWith(Operation operation) {
    this.operation = operation;
  }

  @Test
  @TestDir
  @TestJetty
  @TestHdfs
  public void testOperation() throws Exception {
    createHttpFSServer();
    operation(operation);
  }

  @Test
  @TestDir
  @TestJetty
  @TestHdfs
  public void testOperationDoAs() throws Exception {
    createHttpFSServer();
    UserGroupInformation ugi = UserGroupInformation.createProxyUser(HadoopUsersConfTestHelper.getHadoopUsers()[0],
                                                                    UserGroupInformation.getCurrentUser());
    ugi.doAs(new PrivilegedExceptionAction<Void>() {
      @Override
      public Void run() throws Exception {
        operation(operation);
        return null;
      }
    });
  }

  private void testCreateSnapshot(String snapshotName) throws Exception {
    if (!this.isLocalFS()) {
      Path snapshottablePath = new Path("/tmp/tmp-snap-test");
      createSnapshotTestsPreconditions(snapshottablePath);
      //Now get the FileSystem instance that's being tested
      FileSystem fs = this.getHttpFSFileSystem();
      if (snapshotName == null) {
        fs.createSnapshot(snapshottablePath);
      } else {
        fs.createSnapshot(snapshottablePath, snapshotName);
      }
      Path snapshotsDir = new Path("/tmp/tmp-snap-test/.snapshot");
      FileStatus[] snapshotItems = fs.listStatus(snapshotsDir);
      assertTrue("Should have exactly one snapshot.",
          snapshotItems.length == 1);
      String resultingSnapName = snapshotItems[0].getPath().getName();
      if (snapshotName == null) {
        assertTrue("Snapshot auto generated name not matching pattern",
            Pattern.matches("(s)(\\d{8})(-)(\\d{6})(\\.)(\\d{3})",
                resultingSnapName));
      } else {
        assertTrue("Snapshot name is not same as passed name.",
            snapshotName.equals(resultingSnapName));
      }
      cleanSnapshotTests(snapshottablePath, resultingSnapName);
    }
  }

  private void testCreateSnapshot() throws Exception {
    testCreateSnapshot(null);
    testCreateSnapshot("snap-with-name");
  }

  private void createSnapshotTestsPreconditions(Path snapshottablePath,
      Boolean allowSnapshot) throws Exception {
    //Needed to get a DistributedFileSystem instance, in order to
    //call allowSnapshot on the newly created directory
    DistributedFileSystem distributedFs = (DistributedFileSystem)
        FileSystem.get(snapshottablePath.toUri(), this.getProxiedFSConf());
    distributedFs.mkdirs(snapshottablePath);
    if (allowSnapshot) {
      distributedFs.allowSnapshot(snapshottablePath);
    }
    Path subdirPath = new Path("/tmp/tmp-snap-test/subdir");
    distributedFs.mkdirs(subdirPath);
  }

  private void createSnapshotTestsPreconditions(Path snapshottablePath)
      throws Exception {
    // Allow snapshot by default for snapshot test
    createSnapshotTestsPreconditions(snapshottablePath, true);
  }

  private void cleanSnapshotTests(Path snapshottablePath,
                                  String resultingSnapName) throws Exception {
    DistributedFileSystem distributedFs = (DistributedFileSystem)
        FileSystem.get(snapshottablePath.toUri(), this.getProxiedFSConf());
    distributedFs.deleteSnapshot(snapshottablePath, resultingSnapName);
    distributedFs.delete(snapshottablePath, true);
  }

  private void testRenameSnapshot() throws Exception {
    if (!this.isLocalFS()) {
      Path snapshottablePath = new Path("/tmp/tmp-snap-test");
      createSnapshotTestsPreconditions(snapshottablePath);
      //Now get the FileSystem instance that's being tested
      FileSystem fs = this.getHttpFSFileSystem();
      fs.createSnapshot(snapshottablePath, "snap-to-rename");
      fs.renameSnapshot(snapshottablePath, "snap-to-rename",
          "snap-new-name");
      Path snapshotsDir = new Path("/tmp/tmp-snap-test/.snapshot");
      FileStatus[] snapshotItems = fs.listStatus(snapshotsDir);
      assertTrue("Should have exactly one snapshot.",
          snapshotItems.length == 1);
      String resultingSnapName = snapshotItems[0].getPath().getName();
      assertTrue("Snapshot name is not same as passed name.",
          "snap-new-name".equals(resultingSnapName));
      cleanSnapshotTests(snapshottablePath, resultingSnapName);
    }
  }

  private void testDeleteSnapshot() throws Exception {
    if (!this.isLocalFS()) {
      Path snapshottablePath = new Path("/tmp/tmp-snap-test");
      createSnapshotTestsPreconditions(snapshottablePath);
      //Now get the FileSystem instance that's being tested
      FileSystem fs = this.getHttpFSFileSystem();
      fs.createSnapshot(snapshottablePath, "snap-to-delete");
      Path snapshotsDir = new Path("/tmp/tmp-snap-test/.snapshot");
      FileStatus[] snapshotItems = fs.listStatus(snapshotsDir);
      assertTrue("Should have exactly one snapshot.",
          snapshotItems.length == 1);
      fs.deleteSnapshot(snapshottablePath, "snap-to-delete");
      snapshotItems = fs.listStatus(snapshotsDir);
      assertTrue("There should be no snapshot anymore.",
          snapshotItems.length == 0);
      fs.delete(snapshottablePath, true);
    }
  }

  private void testAllowSnapshot() throws Exception {
    if (!this.isLocalFS()) {
      // Create a directory with snapshot disallowed
      Path path = new Path("/tmp/tmp-snap-test");
      createSnapshotTestsPreconditions(path, false);
      // Get the FileSystem instance that's being tested
      FileSystem fs = this.getHttpFSFileSystem();
      // Check FileStatus
      assertFalse("Snapshot should be disallowed by default",
          fs.getFileStatus(path).isSnapshotEnabled());
      // Allow snapshot
      if (fs instanceof HttpFSFileSystem) {
        HttpFSFileSystem httpFS = (HttpFSFileSystem) fs;
        httpFS.allowSnapshot(path);
      } else if (fs instanceof WebHdfsFileSystem) {
        WebHdfsFileSystem webHdfsFileSystem = (WebHdfsFileSystem) fs;
        webHdfsFileSystem.allowSnapshot(path);
      } else {
        Assert.fail(fs.getClass().getSimpleName() +
            " doesn't support allowSnapshot");
      }
      // Check FileStatus
      assertTrue("allowSnapshot failed",
          fs.getFileStatus(path).isSnapshotEnabled());
      // Cleanup
      fs.delete(path, true);
    }
  }

  private void testDisallowSnapshot() throws Exception {
    if (!this.isLocalFS()) {
      // Create a directory with snapshot allowed
      Path path = new Path("/tmp/tmp-snap-test");
      createSnapshotTestsPreconditions(path);
      // Get the FileSystem instance that's being tested
      FileSystem fs = this.getHttpFSFileSystem();
      // Check FileStatus
      assertTrue("Snapshot should be allowed by DFS",
          fs.getFileStatus(path).isSnapshotEnabled());
      // Disallow snapshot
      if (fs instanceof HttpFSFileSystem) {
        HttpFSFileSystem httpFS = (HttpFSFileSystem) fs;
        httpFS.disallowSnapshot(path);
      } else if (fs instanceof WebHdfsFileSystem) {
        WebHdfsFileSystem webHdfsFileSystem = (WebHdfsFileSystem) fs;
        webHdfsFileSystem.disallowSnapshot(path);
      } else {
        Assert.fail(fs.getClass().getSimpleName() +
            " doesn't support disallowSnapshot");
      }
      // Check FileStatus
      assertFalse("disallowSnapshot failed",
          fs.getFileStatus(path).isSnapshotEnabled());
      // Cleanup
      fs.delete(path, true);
    }
  }

  private void testDisallowSnapshotException() throws Exception {
    if (!this.isLocalFS()) {
      // Create a directory with snapshot allowed
      Path path = new Path("/tmp/tmp-snap-test");
      createSnapshotTestsPreconditions(path);
      // Get the FileSystem instance that's being tested
      FileSystem fs = this.getHttpFSFileSystem();
      // Check FileStatus
      assertTrue("Snapshot should be allowed by DFS",
          fs.getFileStatus(path).isSnapshotEnabled());
      // Create some snapshots
      fs.createSnapshot(path, "snap-01");
      fs.createSnapshot(path, "snap-02");
      // Disallow snapshot
      boolean disallowSuccess = false;
      if (fs instanceof HttpFSFileSystem) {
        HttpFSFileSystem httpFS = (HttpFSFileSystem) fs;
        try {
          httpFS.disallowSnapshot(path);
          disallowSuccess = true;
        } catch (SnapshotException e) {
          // Expect SnapshotException
        }
      } else if (fs instanceof WebHdfsFileSystem) {
        WebHdfsFileSystem webHdfsFileSystem = (WebHdfsFileSystem) fs;
        try {
          webHdfsFileSystem.disallowSnapshot(path);
          disallowSuccess = true;
        } catch (SnapshotException e) {
          // Expect SnapshotException
        }
      } else {
        Assert.fail(fs.getClass().getSimpleName() +
            " doesn't support disallowSnapshot");
      }
      if (disallowSuccess) {
        Assert.fail("disallowSnapshot doesn't throw SnapshotException when "
            + "disallowing snapshot on a directory with at least one snapshot");
      }
      // Check FileStatus, should still be enabled since
      // disallow snapshot should fail
      assertTrue("disallowSnapshot should not have succeeded",
          fs.getFileStatus(path).isSnapshotEnabled());
      // Cleanup
      fs.deleteSnapshot(path, "snap-02");
      fs.deleteSnapshot(path, "snap-01");
      fs.delete(path, true);
    }
  }

  private void testGetSnapshotDiff() throws Exception {
    if (!this.isLocalFS()) {
      // Create a directory with snapshot allowed
      Path path = new Path("/tmp/tmp-snap-test");
      createSnapshotTestsPreconditions(path);
      // Get the FileSystem instance that's being tested
      FileSystem fs = this.getHttpFSFileSystem();
      // Check FileStatus
      Assert.assertTrue(fs.getFileStatus(path).isSnapshotEnabled());
      // Create a file and take a snapshot
      Path file1 = new Path(path, "file1");
      testCreate(file1, false);
      fs.createSnapshot(path, "snap1");
      // Create another file and take a snapshot
      Path file2 = new Path(path, "file2");
      testCreate(file2, false);
      fs.createSnapshot(path, "snap2");
      // Get snapshot diff
      SnapshotDiffReport diffReport = null;
      if (fs instanceof HttpFSFileSystem) {
        HttpFSFileSystem httpFS = (HttpFSFileSystem) fs;
        diffReport = httpFS.getSnapshotDiffReport(path, "snap1", "snap2");
      } else if (fs instanceof WebHdfsFileSystem) {
        WebHdfsFileSystem webHdfsFileSystem = (WebHdfsFileSystem) fs;
        diffReport = webHdfsFileSystem.getSnapshotDiffReport(path,
            "snap1", "snap2");
      } else {
        Assert.fail(fs.getClass().getSimpleName() +
            " doesn't support getSnapshotDiff");
      }
      // Verify result with DFS
      DistributedFileSystem dfs = (DistributedFileSystem)
          FileSystem.get(path.toUri(), this.getProxiedFSConf());
      SnapshotDiffReport dfsDiffReport =
          dfs.getSnapshotDiffReport(path, "snap1", "snap2");
      Assert.assertEquals(diffReport.toString(), dfsDiffReport.toString());
      // Cleanup
      fs.deleteSnapshot(path, "snap2");
      fs.deleteSnapshot(path, "snap1");
      fs.delete(path, true);
    }
  }

  private void testGetSnapshotDiffIllegalParamCase(FileSystem fs, Path path,
      String oldsnapshotname, String snapshotname) throws IOException {
    try {
      if (fs instanceof HttpFSFileSystem) {
        HttpFSFileSystem httpFS = (HttpFSFileSystem) fs;
        httpFS.getSnapshotDiffReport(path, oldsnapshotname, snapshotname);
      } else if (fs instanceof WebHdfsFileSystem) {
        WebHdfsFileSystem webHdfsFileSystem = (WebHdfsFileSystem) fs;
        webHdfsFileSystem.getSnapshotDiffReport(path, oldsnapshotname,
            snapshotname);
      } else {
        Assert.fail(fs.getClass().getSimpleName() +
            " doesn't support getSnapshotDiff");
      }
    } catch (SnapshotException|IllegalArgumentException|RemoteException e) {
      // Expect SnapshotException, IllegalArgumentException
      // or RemoteException(IllegalArgumentException)
      if (e instanceof RemoteException) {
        // Check RemoteException class name, should be IllegalArgumentException
        Assert.assertEquals(((RemoteException) e).getClassName()
            .compareTo(java.lang.IllegalArgumentException.class.getName()), 0);
      }
      return;
    }
    Assert.fail("getSnapshotDiff illegal param didn't throw Exception");
  }

  private void testGetSnapshotDiffIllegalParam() throws Exception {
    if (!this.isLocalFS()) {
      // Create a directory with snapshot allowed
      Path path = new Path("/tmp/tmp-snap-test");
      createSnapshotTestsPreconditions(path);
      // Get the FileSystem instance that's being tested
      FileSystem fs = this.getHttpFSFileSystem();
      // Check FileStatus
      assertTrue("Snapshot should be allowed by DFS",
          fs.getFileStatus(path).isSnapshotEnabled());
      Assert.assertTrue(fs.getFileStatus(path).isSnapshotEnabled());
      // Get snapshot diff
      testGetSnapshotDiffIllegalParamCase(fs, path, "", "");
      testGetSnapshotDiffIllegalParamCase(fs, path, "snap1", "");
      testGetSnapshotDiffIllegalParamCase(fs, path, "", "snap2");
      testGetSnapshotDiffIllegalParamCase(fs, path, "snap1", "snap2");
      // Cleanup
      fs.delete(path, true);
    }
  }

  private void verifyGetSnapshottableDirListing(
      FileSystem fs, DistributedFileSystem dfs) throws Exception {
    // Get snapshottable directory list
    SnapshottableDirectoryStatus[] sds = null;
    if (fs instanceof HttpFSFileSystem) {
      HttpFSFileSystem httpFS = (HttpFSFileSystem) fs;
      sds = httpFS.getSnapshottableDirectoryList();
    } else if (fs instanceof WebHdfsFileSystem) {
      WebHdfsFileSystem webHdfsFileSystem = (WebHdfsFileSystem) fs;
      sds = webHdfsFileSystem.getSnapshottableDirectoryList();
    } else {
      Assert.fail(fs.getClass().getSimpleName() +
          " doesn't support getSnapshottableDirListing");
    }
    // Verify result with DFS
    SnapshottableDirectoryStatus[] dfssds = dfs.getSnapshottableDirListing();
    Assert.assertEquals(JsonUtil.toJsonString(sds),
        JsonUtil.toJsonString(dfssds));
  }

  private void testGetSnapshottableDirListing() throws Exception {
    if (!this.isLocalFS()) {
      FileSystem fs = this.getHttpFSFileSystem();
      // Create directories with snapshot allowed
      Path path1 = new Path("/tmp/tmp-snap-dirlist-test-1");
      DistributedFileSystem dfs = (DistributedFileSystem)
          FileSystem.get(path1.toUri(), this.getProxiedFSConf());
      // Verify response when there is no snapshottable directory
      verifyGetSnapshottableDirListing(fs, dfs);
      createSnapshotTestsPreconditions(path1);
      Assert.assertTrue(fs.getFileStatus(path1).isSnapshotEnabled());
      // Verify response when there is one snapshottable directory
      verifyGetSnapshottableDirListing(fs, dfs);
      Path path2 = new Path("/tmp/tmp-snap-dirlist-test-2");
      createSnapshotTestsPreconditions(path2);
      Assert.assertTrue(fs.getFileStatus(path2).isSnapshotEnabled());
      // Verify response when there are two snapshottable directories
      verifyGetSnapshottableDirListing(fs, dfs);

      // Clean up and verify
      fs.delete(path2, true);
      verifyGetSnapshottableDirListing(fs, dfs);
      fs.delete(path1, true);
      verifyGetSnapshottableDirListing(fs, dfs);
    }
  }

  private void testFileAclsCustomizedUserAndGroupNames() throws Exception {
    if (isLocalFS()) {
      return;
    }

    // Get appropriate conf from the cluster
    MiniDFSCluster miniDFSCluster = ((TestHdfsHelper) hdfsTestHelper)
        .getMiniDFSCluster();
    Configuration conf = miniDFSCluster.getConfiguration(0);
    // If we call getHttpFSFileSystem() without conf from the mini cluster,
    // WebHDFS will be initialized with the default ACL string, causing the
    // setAcl() later to fail. This is only an issue in the unit test.
    FileSystem httpfs = getHttpFSFileSystem(conf);
    if (!(httpfs instanceof WebHdfsFileSystem)
        && !(httpfs instanceof HttpFSFileSystem)) {
      Assert.fail(httpfs.getClass().getSimpleName() +
          " doesn't support custom user and group name pattern. "
          + "Only WebHdfsFileSystem and HttpFSFileSystem support it.");
    }
    final String aclUser = "user:123:rwx";
    final String aclGroup = "group:foo@bar:r--";
    final String aclSet = "user::rwx," + aclUser + ",group::r--," +
        aclGroup + ",other::r--";
    final String dir = "/aclFileTestCustom";
    // Create test file
    FileSystem proxyFs = FileSystem.get(conf);
    proxyFs.mkdirs(new Path(dir));
    Path path = new Path(dir, "/testACL");
    OutputStream os = proxyFs.create(path);
    os.write(1);
    os.close();
    // Set ACL
    httpfs.setAcl(path, AclEntry.parseAclSpec(aclSet, true));
    // Verify getAclStatus responses are the same
    AclStatus proxyAclStat = proxyFs.getAclStatus(path);
    AclStatus httpfsAclStat = httpfs.getAclStatus(path);
    assertSameAcls(httpfsAclStat, proxyAclStat);
    assertSameAcls(httpfs, proxyFs, path);
    // Verify that custom user and group are set.
    List<String> strEntries = new ArrayList<>();
    for (AclEntry aclEntry : httpfsAclStat.getEntries()) {
      strEntries.add(aclEntry.toStringStable());
    }
    Assert.assertTrue(strEntries.contains(aclUser));
    Assert.assertTrue(strEntries.contains(aclGroup));
    // Clean up
    proxyFs.delete(new Path(dir), true);
  }
}


text:
package main

import (
	"log"
	"net/http"
	"os"

	"git.andrewcsellers.com/acsellers/web2015/store"
	"github.com/acsellers/dr/migrate"
	"github.com/acsellers/platform/router"
)

var Conn *store.Conn

func init() {
	// setup database
	var err error
	Conn, err = store.Open("postgres", "user=demo password=demo dbname=suggest")
	if err != nil {
		log.Fatal("Open Conn:", err)
	}
	m := migrate.Database{
		DB:         Conn.DB,
		Schema:     store.Schema,
		Translator: store.NewAppConfig("postgres"),
		Log:        log.New(os.Stdout, "Migration: ", 0),
	}
	err = m.Migrate()
	if err != nil {
		log.Fatal("Migrate Conn:", err)
	}
}

func main() {
	r := router.NewRouter()
	r.Many(NewSuggestCtrl())
	log.Fatal(http.ListenAndServe(":8080", r))
}


text:
require 'spec_helper'

describe CalcWorkingHours do
  it 'should have a version number' do
    CalcWorkingHours::VERSION.should_not be_nil
  end
end


text:
package org.hisp.dhis.importexport.dhis14.object;

import java.util.Date;

public class Dhis14CalculatedDataElement
{
    private int dataElementId;

    private String dataElementName;

    private String dataElementShort;

    private String dataElementDescription;

    private String type;

    private String aggregationOperator;

    private Integer sortOrder;

    private Date lastUpdated;

    private String uid;

    private Integer calculated;

    public int getDataElementId()
    {
        return dataElementId;
    }

    public void setDataElementId( int dataElementId )
    {
        this.dataElementId = dataElementId;
    }

    public String getDataElementName()
    {
        return dataElementName;
    }

    public void setDataElementName( String dataElementName )
    {
        this.dataElementName = dataElementName;
    }

    public String getDataElementShort()
    {
        return dataElementShort;
    }

    public void setDataElementShort( String dataElementShort )
    {
        this.dataElementShort = dataElementShort;
    }

    public String getDataElementDescription()
    {
        return dataElementDescription;
    }

    public void setDataElementDescription( String dataElementDescription )
    {
        this.dataElementDescription = dataElementDescription;
    }

    public String getType()
    {
        return type;
    }

    public void setType( String type )
    {
        this.type = type;
    }

    public String getAggregationOperator()
    {
        return aggregationOperator;
    }

    public void setAggregationOperator( String aggregationOperator )
    {
        this.aggregationOperator = aggregationOperator;
    }

    public Integer getSortOrder()
    {
        return sortOrder;
    }

    public void setSortOrder( Integer sortOrder )
    {
        this.sortOrder = sortOrder;
    }

    public Date getLastUpdated()
    {
        return lastUpdated;
    }

    public void setLastUpdated( Date lastUpdated )
    {
        this.lastUpdated = lastUpdated;
    }

    public String getUid()
    {
        return uid;
    }

    public void setUid( String uid )
    {
        this.uid = uid;
    }

    public Integer getCalculated()
    {
        return calculated;
    }

    public void setCalculated( Integer calculated )
    {
        this.calculated = calculated;
    }
}


text:
package com.pleuvoir.config;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.pleuvoir.time.Clock;

/**
 * 配置加载器的抽象实现
 * @author pleuvoir
 *
 */
public abstract class AbstractConfigLoader implements ConfigLoader {

	protected static final Logger logger  = LoggerFactory.getLogger(AbstractConfigLoader.class);

	protected String location;

	@Override
	public void setLocation(String location) {
		if (StringUtils.isBlank(location))
			throw new IllegalArgumentException("文件位置不能为空");
		this.location = location;
	}

	@Override
	public void beforeInit() {
		this.setLocation(location);
		logger.info("文件初始化前,设置文件位置,{}",location);
	}

	@Override
	public void afterInit() {
		logger.info("文件初始化完成,后置处理。。");
	}

	@Override
	public void load() {
		long startTime = Clock.startTime();
		beforeInit();
		init();
		afterInit();
		logger.info("文件加载完成,文件位置:{},耗时:{}", this.location, Clock.passed(startTime));
	}

}