<commit_before>package io.hydrosphere.mist.master

import akka.actor.ActorSystem
import akka.stream.scaladsl
import akka.testkit.TestKit
import io.hydrosphere.mist.core.CommonData.{Action, JobParams, RunJobRequest}
import io.hydrosphere.mist.core.MockitoSugar
import io.hydrosphere.mist.master.JobDetails.Status
import io.hydrosphere.mist.master.Messages.StatusMessages._
import io.hydrosphere.mist.master.logging.JobsLogger
import io.hydrosphere.mist.master.store.JobRepository
import mist.api.data._
import org.mockito.Matchers._
import org.mockito.Mockito._
import org.scalatest.concurrent.Eventually
import org.scalatest.prop.TableDrivenPropertyChecks._
import org.scalatest.time.{Millis, Seconds, Span}
import org.scalatest.{FunSpecLike, Matchers}

import scala.concurrent.Future

class StatusServiceSpec extends TestKit(ActorSystem("testFront"))
  with FunSpecLike
  with MockitoSugar
  with Matchers
  with Eventually {

  val params = JobParams("path", "className", Map.empty, Action.Execute)

  it("should register jobs") {
    val store = mock[JobRepository]
    val jobsLogger = mock[JobsLogger]
    when(store.update(any[JobDetails]))
      .thenReturn(Future.successful(()))

    val status = system.actorOf(StatusService.props(store, mock[EventsStreamer], jobsLogger))

    status ! Register(
      RunJobRequest("id", params),
      "endpoint",
      "context",
       JobDetails.Source.Async("Kafka"),
       None,
      "workerId"
    )

    eventually(timeout(Span(3, Seconds))) {
      verify(store).update(any[JobDetails])
    }
  }

  it("should update status in storage and call publisher") {
    val store = mock[JobRepository]
    val jobsLogger = mock[JobsLogger]
    when(store.get(any[String])).thenReturn({
      val jobDetails = JobDetails(
        params = params,
        jobId = "id",
        source = JobDetails.Source.Http,
        endpoint = "endpoint",
        context = "context",
        externalId = None,
        workerId = "workerId"
      )
      Future.successful(Some(jobDetails))
    })
    when(store.update(any[JobDetails])).thenReturn(Future.successful(()))

    val streamer = mock[EventsStreamer]

    val status = system.actorOf(StatusService.props(store, streamer, jobsLogger))

    status ! StartedEvent("id", System.currentTimeMillis())

    eventually(timeout(Span(1, Seconds))) {
      verify(store).update(any[JobDetails])
    }

    eventually(timeout(Span(1, Seconds))) {
      verify(streamer).push(any[StartedEvent])
    }
  }

  describe("event conversion") {

    val baseDetails = JobDetails(
        params = params,
        jobId = "id",
        source = JobDetails.Source.Http,
        endpoint = "endpoint",
        context = "context",
        externalId = None,
        workerId = "workerId"
      )<commit_msg>tests status service - increase eventually timeout<commit_after>package io.hydrosphere.mist.master

import akka.actor.ActorSystem
import akka.stream.scaladsl
import akka.testkit.TestKit
import io.hydrosphere.mist.core.CommonData.{Action, JobParams, RunJobRequest}
import io.hydrosphere.mist.core.MockitoSugar
import io.hydrosphere.mist.master.JobDetails.Status
import io.hydrosphere.mist.master.Messages.StatusMessages._
import io.hydrosphere.mist.master.logging.JobsLogger
import io.hydrosphere.mist.master.store.JobRepository
import mist.api.data._
import org.mockito.Matchers._
import org.mockito.Mockito._
import org.scalatest.concurrent.Eventually
import org.scalatest.prop.TableDrivenPropertyChecks._
import org.scalatest.time.{Millis, Seconds, Span}
import org.scalatest.{FunSpecLike, Matchers}

import scala.concurrent.Future

class StatusServiceSpec extends TestKit(ActorSystem("testFront"))
  with FunSpecLike
  with MockitoSugar
  with Matchers
  with Eventually {

  val params = JobParams("path", "className", Map.empty, Action.Execute)

  it("should register jobs") {
    val store = mock[JobRepository]
    val jobsLogger = mock[JobsLogger]
    when(store.update(any[JobDetails]))
      .thenReturn(Future.successful(()))

    val status = system.actorOf(StatusService.props(store, mock[EventsStreamer], jobsLogger))

    status ! Register(
      RunJobRequest("id", params),
      "endpoint",
      "context",
       JobDetails.Source.Async("Kafka"),
       None,
      "workerId"
    )

    eventually(timeout(Span(3, Seconds))) {
      verify(store).update(any[JobDetails])
    }
  }

  it("should update status in storage and call publisher") {
    val store = mock[JobRepository]
    val jobsLogger = mock[JobsLogger]
    when(store.get(any[String])).thenReturn({
      val jobDetails = JobDetails(
        params = params,
        jobId = "id",
        source = JobDetails.Source.Http,
        endpoint = "endpoint",
        context = "context",
        externalId = None,
        workerId = "workerId"
      )
      Future.successful(Some(jobDetails))
    })
    when(store.update(any[JobDetails])).thenReturn(Future.successful(()))

    val streamer = mock[EventsStreamer]

    val status = system.actorOf(StatusService.props(store, streamer, jobsLogger))

    status ! StartedEvent("id", System.currentTimeMillis())

    eventually(timeout(Span(3, Seconds))) {
      verify(store).update(any[JobDetails])
    }

    eventually(timeout(Span(1, Seconds))) {
      verify(streamer).push(any[StartedEvent])
    }
  }

  describe("event conversion") {

    val baseDetails = JobDetails(
        params = params,
        jobId = "id",
        source = JobDetails.Source.Http,
        endpoint = "endpoint",
        context = "context",
        externalId = None,
        workerId = "workerId"
      )
{'fraction_non_alphanumeric': 0.11734238256831576, 'fraction_numerical': 0.0010716199321307376, 'mean_word_length': 3.612850082372323, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '5454425', 'n_tokens_mistral': 1713, 'n_tokens_neox': 1661, 'n_words': 319}