// Copyright (C) 2018 GerritForge Ltd
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package com.gerritforge.analytics.gitcommits.engine.events
import com.gerritforge.analytics.SparkTestSupport
import com.gerritforge.analytics.gitcommits.engine.GerritAnalyticsTransformations.{CommitInfo, UserActivitySummary}
import com.gerritforge.analytics.gitcommits.engine.events.GerritEventsTransformations.NotParsableJsonEvent
import org.apache.spark.rdd.RDD
import org.scalatest.{Inside, Matchers, WordSpec}

class GerritEventsTransformationsSpec
    extends WordSpec
    with Matchers
    with SparkTestSupport
    with Inside
    with EventFixture {

  "tryParseGerritTriggeredEvent" should {

    implicit val eventParser: GerritJsonEventParser = EventParser

    "Parse a correctly formed event" in new EventFixture {
      GerritEventsTransformations.tryParseGerritTriggeredEvent(refUpdated.json) shouldBe Right(
        refUpdated.event)
    }

    "Return a description of the failure in with the original event source a Left object if the JSON provided is invalid" in {
      val invalidJson = "invalid json string"
      GerritEventsTransformations.tryParseGerritTriggeredEvent(invalidJson) shouldBe Left(
        NotParsableJsonEvent(invalidJson, "unknown token i - Near: i"))
    }

    "Return a description of the failure with the original event source  in a Left object if the JSON event is not supported" in {
      val unsupportedEvent = """{"type":"ref-updated-UNSUPPORTED","eventCreatedOn":1516531868}"""

      GerritEventsTransformations.tryParseGerritTriggeredEvent(unsupportedEvent) shouldBe Left(
        NotParsableJsonEvent(unsupportedEvent, "Unsupported event type 'ref-updated-UNSUPPORTED'"))
    }
  }

  "PimpedJsonRDD" should {
    "Convert an RDD of JSON events into an RDD of events or unparsed json strings" in {
      val jsonRdd = sc.parallelize(Seq(refUpdated.json, changeMerged.json, "invalid json string"))

      import GerritEventsTransformations._

      jsonRdd.parseEvents(EventParser).collect() should contain only (
        Right(refUpdated.event),
        Right(changeMerged.event),
        Left(NotParsableJsonEvent("invalid json string", "unknown token i - Near: i"))
      )
    }
  }

  "extractUserActivitySummary" should {
    implicit val eventParser: GerritJsonEventParser = EventParser

    "Build one UserActivitySummary object if given a series of non-merge commits" in {
      val events: Seq[ChangeMergedEvent] = Seq(
        aChangeMergedEvent("1",
                           1001l,
                           newRev = "rev1",
                           insertions = 2,
                           deletions = 1,
                           branch = "stable-1.14"),
        aChangeMergedEvent("2",
                           1002l,
                           newRev = "rev2",
                           insertions = 3,
                           deletions = 0,
                           branch = "stable-1.14"),
        aChangeMergedEvent("3",
                           1003l,
                           newRev = "rev3",
                           insertions = 1,
                           deletions = 4,
                           branch = "stable-1.14"),
        aChangeMergedEvent("4",
                           1004l,
                           newRev = "rev4",
                           insertions = 0,
                           deletions = 2,
                           branch = "stable-1.14"),
        aChangeMergedEvent("5",
                           1005l,
                           newRev = "rev5",
                           insertions = 1,
                           deletions = 1,
                           branch = "stable-1.14")
      ).map(_.event)

      val summaries: Iterable[UserActivitySummary] =
        GerritEventsTransformations.extractUserActivitySummary(
          changes = events,
          year = 2018,
          month = 1,
          day = 10,
          hour = 1
        )

      summaries should have size 1

      inside(summaries.head) {
        case UserActivitySummary(year,
                                 month,
                                 day,
                                 hour,
                                 name,
                                 email,
                                 num_commits,
                                 _,
                                 _,
                                 added_lines,
                                 deleted_lines,
                                 commits,
                                 branches,
                                 last_commit_date,
                                 is_merge) =>
          year shouldBe 2018
          month shouldBe 1
          day shouldBe 10
          hour shouldBe 1
          name shouldBe "Administrator"
          email shouldBe "admin@example.com"
          num_commits shouldBe events.size
          last_commit_date shouldBe 1005000l
          is_merge shouldBe false
          added_lines shouldBe 7
          deleted_lines shouldBe 8
          commits should contain only (
            CommitInfo("rev1", 1001000l, false),
            CommitInfo("rev2", 1002000l, false),
            CommitInfo("rev3", 1003000l, false),
            CommitInfo("rev4", 1004000l, false),
            CommitInfo("rev5", 1005000l, false)
          )
          branches should contain only "stable-1.14"
      }
    }

    "Build two UserActivitySummaries object if given a mixed series of merge and non merge commits" in {
      val events: Seq[ChangeMergedEvent] = Seq(
        aChangeMergedEvent("1", 1001l, newRev = "rev1", isMergeCommit = true),
        aChangeMergedEvent("2", 1002l, newRev = "rev2"),
        aChangeMergedEvent("3", 1003l, newRev = "rev3", isMergeCommit = true),
        aChangeMergedEvent("4", 1004l, newRev = "rev4"),
        aChangeMergedEvent("5", 1005l, newRev = "rev5")
      ).map(_.event)

      val summaries: Iterable[UserActivitySummary] =
        GerritEventsTransformations.extractUserActivitySummary(
          changes = events,
          year = 2018,
          month = 1,
          day = 10,
          hour = 1
        )

      summaries should have size 2

      summaries.foreach { summary =>
        inside(summary) {
          case UserActivitySummary(year,
                                   month,
                                   day,
                                   hour,
                                   name,
                                   email,
                                   _,
                                   _,
                                   _,
                                   _,
                                   _,
                                   _,
                                   _,
                                   _,
                                   _) =>
            year shouldBe 2018
            month shouldBe 1
            day shouldBe 10
            hour shouldBe 1
            name shouldBe "Administrator"
            email shouldBe "admin@example.com"
        }
      }

      summaries.foreach { summary =>
        inside(summary) {
          case UserActivitySummary(_,
                                   _,
                                   _,
                                   _,
                                   _,
                                   _,
                                   num_commits,
                                   _,
                                   _,
                                   _,
                                   _,
                                   commits,
                                   _,
                                   last_commit_date,
                                   false) =>
            num_commits shouldBe 3
            last_commit_date shouldBe 1005000l
            commits should contain only (
              CommitInfo("rev2", 1002000l, false),
              CommitInfo("rev4", 1004000l, false),
              CommitInfo("rev5", 1005000l, false)
            )

          case UserActivitySummary(_,
                                   _,
                                   _,
                                   _,
                                   _,
                                   _,
                                   num_commits,
                                   _,
                                   _,
                                   _,
                                   _,
                                   commits,
                                   _,
                                   last_commit_date,
                                   true) =>
            num_commits shouldBe 2
            last_commit_date shouldBe 1003000l
            commits should contain only (
              CommitInfo("rev1", 1001000l, true),
              CommitInfo("rev3", 1003000l, true)
            )
        }
      }

    }
  }

  "Pimped Per Project UserActivitySummary RDD" should {
    "Allow conversion to a DataFrame equivalent to what extracted from the Analytics plugin" in {
      import GerritEventsTransformations._
      import com.gerritforge.analytics.gitcommits.engine.GerritAnalyticsTransformations._
      import spark.implicits._

      val aliasDF = sc
        .parallelize(
          Seq(
            ("stefano_alias", "stefano@galarraga-org.com", "")
          ))
        .toDF("author", "email", "organization")

      val expectedDate = System.currentTimeMillis

      val analyticsJobOutput =
        sc.parallelize(
            Seq(
              "project1" -> UserActivitySummary(
                2018,
                1,
                20,
                10,
                "Stefano",
                "stefano@galarraga-org.com",
                1,
                2,
                1,
                10,
                4,
                Array(CommitInfo("sha1", expectedDate, false)),
                Array("master", "stable-2.14"),
                expectedDate,
                false
              )
            ))
          .asEtlDataFrame(sql)
          .addOrganization()
          .handleAliases(Some(aliasDF))
          .dropCommits

      val expected = sc
        .parallelize(
          Seq(
            ("project1",
             "stefano_alias",
             "stefano@galarraga-org.com",
             2018,
             1,
             20,
             10,
             2,
             1,
             10,
             4,
             1,
             expectedDate,
             false,
             Array("master", "stable-2.14"),
             "galarraga-org")
          ))
        .toDF(
          "project",
          "author",
          "email",
          "year",
          "month",
          "day",
          "hour",
          "num_files",
          "num_distinct_files",
          "added_lines",
          "deleted_lines",
          "num_commits",
          "last_commit_date",
          "is_merge",
          "branches",
          "organization"
        )

      val collected = analyticsJobOutput.collect()

      collected should contain theSameElementsAs expected.collect()
    }
  }

  "removeEventsForCommits" should {
    "remove any event modifying a gerrit repo to go toward the commits to be excluded" in {
      import GerritEventsTransformations._

      val toKeep1 = aRefUpdatedEvent("oldRev", "RevToKeep")
      val toKeep2 = aChangeMergedEvent(changeId = "changeId2", newRev = "RevToKeep2")
      val events: RDD[GerritRefHasNewRevisionEvent] = sc.parallelize(
        Seq(
          aRefUpdatedEvent("oldRev", "RevToExclude1"),
          toKeep1,
          aRefUpdatedEvent("oldRev", "RevToExclude2"),
          aChangeMergedEvent(changeId = "changeId1", newRev = "RevToExclude3"),
          toKeep2
        ).map(_.event))

      events
        .removeEventsForCommits(
          sc.parallelize(Seq("RevToExclude1", "RevToExclude2", "RevToExclude3")))
        .collect() should contain only (toKeep1.event, toKeep2.event)
    }
  }
}

trait EventFixture {

  // Forcing early type failures
  case class JsonEvent[T <: GerritJsonEvent](json: String) {
    val event: T = EventParser.fromJson(json).get.asInstanceOf[T]
  }

  val refUpdated: JsonEvent[RefUpdatedEvent] = aRefUpdatedEvent(
    oldRev = "863b64002f2a9922deba69407804a44703c996e0",
    newRev = "d3131be8d7c920badd28b70d8c039682568c8de5")

  val changeMerged: JsonEvent[ChangeMergedEvent] = aChangeMergedEvent(
    "I5e6b5a3bbe8a29fb0393e4a28da536e0a198b755")

  def aRefUpdatedEvent(oldRev: String, newRev: String, createdOn: Long = 1000l) =
    JsonEvent[RefUpdatedEvent](
      s"""{"submitter":{"name":"Administrator","email":"admin@example.com","username":"admin"},
       | "refUpdate":{"oldRev":"$oldRev",
       | "newRev":"$newRev",
       | "refName": "refs/heads/master","project":"subcut"},
       |"type":"ref-updated","eventCreatedOn":$createdOn}""".stripMargin)

  def aChangeMergedEvent(changeId: String,
                         createdOnInSecs: Long = 1000l,
                         newRev: String = "863b64002f2a9922deba69407804a44703c996e0",
                         isMergeCommit: Boolean = false,
                         insertions: Integer = 0,
                         deletions: Integer = 0,
                         branch: String = "master") =
    JsonEvent[ChangeMergedEvent](
      s"""{
       |"submitter":{"name":"Administrator","email":"admin@example.com","username":"admin"},
       |"newRev":"$newRev",
       |"patchSet":{
       | "number":1,
       | "revision":"$newRev",
       | "parents": ${if (isMergeCommit)
           """["4a4e59272f1f88824d805c0f4233c1ee7331e986", "4a4e59272f1f88824d805c0f4233c1ee7331e987"]"""
         else """["4a4e59272f1f88824d805c0f4233c1ee7331e986"]"""},
       | "ref":"refs/changes/01/1/1",
       | "uploader":{"name":"Administrator","email":"admin@example.com","username":"admin"},
       | "createdOn":1516530259,
       | "author":{"name":"Stefano Galarraga","email":"galarragas@gmail.com","username":""},
       | "isDraft":false,
       | "kind":"REWORK",
       | "sizeInsertions":$insertions,
       | "sizeDeletions":$deletions
       |},
       |"change":{
       | "project":"subcut","branch":"$branch","topic":"TestEvents","id":"$changeId","number":1,"subject":"Generating some changes to test events",
       | "owner":{"name":"Administrator","email":"admin@example.com","username":"admin"},
       | "url":"http://842860da5b33:8080/1","commitMessage":"Generating some changes to test events Change-Id: $changeId",
       | "createdOn":1516530259,"status":"MERGED"
       |},
       |"project":{"name":"subcut"},
       |"refName":"refs/heads/$branch",
       |"changeKey":{"id":"$changeId"},
       |"type":"change-merged",
       |"eventCreatedOn": $createdOnInSecs
       |}""".stripMargin)

}