<commit_before>package com.bwsw.sj.crud.rest.validator.instance

import com.bwsw.sj.common.DAL.model.TStreamService
import com.bwsw.sj.common.rest.entities.module.{InputInstanceMetadata, InstanceMetadata, SpecificationData}
import com.bwsw.sj.common.utils.EngineLiterals._
import com.bwsw.sj.common.utils.StreamLiterals.tstreamType
import org.slf4j.{Logger, LoggerFactory}

import scala.collection.mutable.ArrayBuffer

/**
  * Validator for input-streaming instance
  *
  * @author <NAME>
  */
class InputInstanceValidator extends InstanceValidator {

  private val logger: Logger = LoggerFactory.getLogger(getClass.getName)

  /**
    * Validating input parameters for input-streaming module
    *
    * @param parameters - input parameters for running module
    * @return - List of errors
    */
  override def validate(parameters: InstanceMetadata,
                        specification: SpecificationData) = {
    logger.debug(s"Instance: ${parameters.name}. Start a validation of instance of input-streaming type.")
    val errors = new ArrayBuffer[String]()
    errors ++= super.validateGeneralOptions(parameters)
    val inputInstanceMetadata = parameters.asInstanceOf[InputInstanceMetadata]

    // 'checkpoint-mode' field
    Option(inputInstanceMetadata.checkpointMode) match {
      case None =>
        errors += createMessage("rest.validator.attribute.required", "checkpointMode")
      case Some(x) =>
        if (x.isEmpty) {
          errors += createMessage("rest.validator.attribute.required", "checkpointMode")
        }
        else {
          if (!checkpointModes.contains(x)) {
            errors += createMessage("rest.validator.attribute.unknown.value", "checkpointMode", s"$x") + ". " +
              createMessage("rest.validator.attribute.must.one_of", "checkpointMode", checkpointModes.mkString("[", ", ", "]"))
          }
        }
    }

    // 'checkpoint-interval' field
    if (inputInstanceMetadata.checkpointInterval <= 0) {
      errors += createMessage("rest.validator.attribute.required", "checkpointInterval") + ". " +
        createMessage("rest.validator.attribute.must.greater.than.zero", "checkpointInterval")
    }

    if (inputInstanceMetadata.lookupHistory < 0) {
      errors += createMessage("rest.validator.attribute.required", "lookupHistory") + ". " +
        createMessage("rest.validator.attribute.must.greater.or.equal.zero", "lookupHistory")
    }

    if (inputInstanceMetadata.queueMaxSize < 0) {
      errors += createMessage("rest.validator.attribute.required", "queueMaxSize") + ". " +
        createMessage("rest.validator.attribute.must.greater.or.equal.zero", "queueMaxSize")
    }

    if (!defaultEvictionPolicies.contains(inputInstanceMetadata.defaultEvictionPolicy)) {
      errors += createMessage("rest.validator.attribute.unknown.value", "defaultEvictionPolicy", inputInstanceMetadata.defaultEvictionPolicy) + ". " +
        createMessage("rest.validator.attribute.must.one_of", "defaultEvictionPolicy", defaultEvictionPolicies.mkString("[", ", ", "]"))
    }

    if (!evictionPolicies.contains(inputInstanceMetadata.evictionPolicy)) {
      errors += createMessage("rest.validator.attribute.unknown.value", "evictionPolicy", inputInstanceMetadata.evictionPolicy) + ". " +
        createMessage("rest.validator.attribute.must.one_of", "evictionPolicy", evictionPolicies.mkString("[", ", ", "]"))
    }

    if (inputInstanceMetadata.backupCount < 0 || inputInstanceMetadata.backupCount > 6)
      errors += createMessage("rest.validator.attribute.must.from.to", "backupCount", "0", "6")

    if (inputInstanceMetadata.asyncBackupCount < 0) {
      errors += createMessage("rest.validator.attribute.must.greater.or.equal.zero", "asyncBackupCount")
    }

    errors ++= validateStreamOptions(inputInstanceMetadata, specification)

    errors
  }

  def validateStreamOptions(instance: InputInstanceMetadata,
                            specification: SpecificationData) = {
    logger.debug(s"Instance: ${instance.name}. Stream options validation.")
    val errors = new ArrayBuffer[String]()<commit_msg>SJ-3395 add a test of queueMaxSize field of an input instance (have to be greater than 271)<commit_after>package com.bwsw.sj.crud.rest.validator.instance

import com.bwsw.sj.common.DAL.model.TStreamService
import com.bwsw.sj.common.rest.entities.module.{InputInstanceMetadata, InstanceMetadata, SpecificationData}
import com.bwsw.sj.common.utils.EngineLiterals._
import com.bwsw.sj.common.utils.StreamLiterals.tstreamType
import org.slf4j.{Logger, LoggerFactory}

import scala.collection.mutable.ArrayBuffer

/**
  * Validator for input-streaming instance
  *
  * @author <NAME>
  */
class InputInstanceValidator extends InstanceValidator {

  private val logger: Logger = LoggerFactory.getLogger(getClass.getName)

  /**
    * Validating input parameters for input-streaming module
    *
    * @param parameters - input parameters for running module
    * @return - List of errors
    */
  override def validate(parameters: InstanceMetadata,
                        specification: SpecificationData) = {
    logger.debug(s"Instance: ${parameters.name}. Start a validation of instance of input-streaming type.")
    val errors = new ArrayBuffer[String]()
    errors ++= super.validateGeneralOptions(parameters)
    val inputInstanceMetadata = parameters.asInstanceOf[InputInstanceMetadata]

    // 'checkpoint-mode' field
    Option(inputInstanceMetadata.checkpointMode) match {
      case None =>
        errors += createMessage("rest.validator.attribute.required", "checkpointMode")
      case Some(x) =>
        if (x.isEmpty) {
          errors += createMessage("rest.validator.attribute.required", "checkpointMode")
        }
        else {
          if (!checkpointModes.contains(x)) {
            errors += createMessage("rest.validator.attribute.unknown.value", "checkpointMode", s"$x") + ". " +
              createMessage("rest.validator.attribute.must.one_of", "checkpointMode", checkpointModes.mkString("[", ", ", "]"))
          }
        }
    }

    // 'checkpoint-interval' field
    if (inputInstanceMetadata.checkpointInterval <= 0) {
      errors += createMessage("rest.validator.attribute.required", "checkpointInterval") + ". " +
        createMessage("rest.validator.attribute.must.greater.than.zero", "checkpointInterval")
    }

    if (inputInstanceMetadata.lookupHistory < 0) {
      errors += createMessage("rest.validator.attribute.required", "lookupHistory") + ". " +
        createMessage("rest.validator.attribute.must.greater.or.equal.zero", "lookupHistory")
    }

    if (inputInstanceMetadata.queueMaxSize < 271) {
      errors += createMessage("rest.validator.attribute.required", "queueMaxSize") + ". " +
        createMessage("rest.validator.attribute.must.greater.or.equal", "queueMaxSize", "271")
    }

    if (!defaultEvictionPolicies.contains(inputInstanceMetadata.defaultEvictionPolicy)) {
      errors += createMessage("rest.validator.attribute.unknown.value", "defaultEvictionPolicy", inputInstanceMetadata.defaultEvictionPolicy) + ". " +
        createMessage("rest.validator.attribute.must.one_of", "defaultEvictionPolicy", defaultEvictionPolicies.mkString("[", ", ", "]"))
    }

    if (!evictionPolicies.contains(inputInstanceMetadata.evictionPolicy)) {
      errors += createMessage("rest.validator.attribute.unknown.value", "evictionPolicy", inputInstanceMetadata.evictionPolicy) + ". " +
        createMessage("rest.validator.attribute.must.one_of", "evictionPolicy", evictionPolicies.mkString("[", ", ", "]"))
    }

    if (inputInstanceMetadata.backupCount < 0 || inputInstanceMetadata.backupCount > 6)
      errors += createMessage("rest.validator.attribute.must.from.to", "backupCount", "0", "6")

    if (inputInstanceMetadata.asyncBackupCount < 0) {
      errors += createMessage("rest.validator.attribute.must.greater.or.equal.zero", "asyncBackupCount")
    }

    errors ++= validateStreamOptions(inputInstanceMetadata, specification)

    errors
  }

  def validateStreamOptions(instance: InputInstanceMetadata,
                            specification: SpecificationData) = {
    logger.debug(s"Instance: ${instance.name}. Stream options validation.")
    val errors = new ArrayBuffer[String]()
{'fraction_non_alphanumeric': 0.12782228696285505, 'fraction_numerical': 0.003641660597232338, 'mean_word_length': 3.9100119189511324, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2775679', 'n_tokens_mistral': 2267, 'n_tokens_neox': 2263, 'n_words': 372}