<commit_before>/**
 * Copyright (C) 2011 LShift 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 net.lshift.diffa.kernel.differencing

import org.easymock.EasyMock._
import org.easymock.{IAnswer, EasyMock}
import org.joda.time.DateTime
import net.lshift.diffa.kernel.util.EasyMockScalaUtils._
import net.lshift.diffa.kernel.participants._
import org.apache.commons.codec.digest.DigestUtils
import net.lshift.diffa.kernel.util.Dates._
import net.lshift.diffa.kernel.util.DateUtils._
import net.lshift.diffa.kernel.events._
import org.junit.Test
import org.junit.Assert._
import collection.mutable.HashMap
import collection.mutable.Map
import net.lshift.diffa.kernel.config.ConfigStore
import scala.collection.JavaConversions._
import net.lshift.diffa.kernel.participants.EasyConstraints._
import net.lshift.diffa.kernel.util.Conversions._

/**
 * Base class for the various policy tests.
 */
abstract class AbstractPolicyTest {
  // The policy instance under test
  protected def policy:VersionPolicy

  // A method for generating a downstream version based on an upstream version
  protected def downstreamVersionFor(v:String):String

  // The various mocks for listeners and participants
  val usMock = createStrictMock("us", classOf[UpstreamParticipant])
  val dsMock = createStrictMock("ds", classOf[DownstreamParticipant])
  val nullListener = new NullDifferencingListener
  
  val store = createStrictMock("versionStore", classOf[VersionCorrelationStore])
  EasyMock.checkOrder(store, false)   // Store doesn't care about order
  val listener = createStrictMock("listener", classOf[DifferencingListener])

  val configStore = createStrictMock("configStore", classOf[ConfigStore])
  val abPair = "A-B"

  val categories = new HashMap[String,String]
  val pair = new net.lshift.diffa.kernel.config.Pair()
  pair.categories = Map("bizDate" -> "date")

  expect(configStore.getPair(abPair)).andReturn(pair).anyTimes
  replay(configStore)

  protected def replayAll = replay(usMock, dsMock, store, listener)
  protected def verifyAll = verify(usMock, dsMock, store, listener, configStore)

  // Make declaring of sequences of specific types clearer
  def DigestsFromParticipant[T](vals:T*) = Seq[T](vals:_*)
  def VersionsFromStore[T](vals:T*) = Seq[T](vals:_*)

  protected val yearly = YearlyCategoryFunction
  protected val monthly = MonthlyCategoryFunction
  protected val daily = DailyCategoryFunction
  protected val individual = IndividualCategoryFunction
  
  def Up(v:VersionID, d:DateTime, s:String) = UpstreamVersion(v, Seq(d.toString()), d, s)
  def Down(v:VersionID, d:DateTime, s1:String, s2:String) = DownstreamVersion(v, Seq(d.toString()), d, s1, s2)
  def UpInt(v:VersionID, i:Int, s:String) = UpstreamVersion(v, Seq(i.toString), new DateTime, s)
  def DownInt(v:VersionID, i:Int, s1:String, s2:String) = DownstreamVersion(v, Seq(i.toString), new DateTime, s1, s2)

  def bizDateMap(d:DateTime) = Map("bizDate" -> d.toString())
  def bizDateSeq(d:DateTime) = Seq(d.toString())

  @Test
  def shouldOnlySyncTopLevelsWhenParticipantsAndStoresMatch {
    // Expect only a top-level sync between the pairs
    expectUpstreamAggregateSync(abPair, List(unconstrainedDate(YearlyCategoryFunction)),
      DigestsFromParticipant(
        AggregateDigest(Seq("2009"), null, DigestUtils.md5Hex("vsn1")),
        AggregateDigest(Seq("2010"), null, DigestUtils.md5Hex("vsn2"))),
      VersionsFromStore(
        Up(VersionID(abPair, "id1"), JUN_6_2009_1, "vsn1"),
        Up(VersionID(abPair, "id2"), JUL_8_2010_1, "vsn2")))
    expectDownstreamAggregateSync(abPair, List(unconstrainedDate(YearlyCategoryFunction)),
      DigestsFromParticipant(
        AggregateDigest(Seq("2009"),  null, DigestUtils.md5Hex(downstreamVersionFor("vsn1"))),
        AggregateDigest(Seq("2010"), null, DigestUtils.md5Hex(downstreamVersionFor("vsn2")))),
      VersionsFromStore(
        DownstreamVersion(VersionID(abPair, "id1"), bizDateSeq(JUN_6_2009_1), JUN_6_2009_1, "vsn1", downstreamVersionFor("vsn1")),
        DownstreamVersion(VersionID(abPair, "id2"), bizDateSeq(JUL_8_2010_1), JUL_8_2010_1, "vsn2", downstreamVersionFor("vsn2"))))

    // We should still see an unmatched version check
    expect(store.unmatchedVersions(EasyMock.eq(abPair), EasyMock.eq(Seq(unconstrainedDate(YearlyCategoryFunction))))).
        andReturn(Seq())
    replayAll

    policy.difference(abPair, usMock, dsMock, nullListener)
    verifyAll
  }

  @Test
  def shouldOnlySyncTopLevelsWhenParticipantsAndStoresMatchForIntegerCategories {
    pair.categories = Map("someInt" -> "int")
    // Expect only a top-level sync between the pairs
    expectUpstreamAggregateSync(abPair, List(unconstrainedInt(ThousandsCategoryFunction)),
        DigestsFromParticipant(
            AggregateDigest(attributes=Seq("1000"), lastUpdated=null, digest=DigestUtils.md5Hex("vsn1")),
            AggregateDigest(Seq("2000"), null, DigestUtils.md5Hex("vsn2"))),
        VersionsFromStore(
            UpInt(VersionID(abPair, "id1"), 1234, "vsn1"),
            UpInt(VersionID(abPair, "id2"), 2345, "vsn2")))
    expectDownstreamAggregateSync(abPair, List(unconstrainedInt(ThousandsCategoryFunction)),
        DigestsFromParticipant(
            AggregateDigest(Seq("1000"), null, DigestUtils.md5Hex(downstreamVersionFor("vsn1"))),
            AggregateDigest(Seq("2000"), null, DigestUtils.md5Hex(downstreamVersionFor("vsn2")))),
        VersionsFromStore(
            DownstreamVersion(VersionID(abPair, "id1"), Seq("1234"), JUN_6_2009_1, "vsn1", downstreamVersionFor("vsn1")),
            DownstreamVersion(VersionID(abPair, "id2"), Seq("2345"), JUL_8_2010_1, "vsn2", downstreamVersionFor("vsn2"))))

    // We should still see an unmatched version check
    expect(store.unmatchedVersions(EasyMock.eq(abPair), EasyMock.eq(Seq(unconstrainedInt(ThousandsCategoryFunction))))).
           andReturn(Seq())
    replayAll

    policy.difference(abPair, usMock, dsMock, nullListener)
    verifyAll
  }

  @Test
  def shouldUpdateUpstreamVersionsWhenStoreIsOutOfDateWithUpstreamParticipant {
    val timestamp = new DateTime()
    // Expect only a top-level sync between the pairs
    expectUpstreamAggregateSync(abPair, List(unconstrainedDate(YearlyCategoryFunction)),<commit_msg>Factor out commonalities in shouldOnlySyncTopLevels... tests [#149]<commit_after>/**
 * Copyright (C) 2011 LShift 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 net.lshift.diffa.kernel.differencing

import scala.collection.mutable.{HashMap, Map}
import scala.collection.JavaConversions._

import org.easymock.EasyMock._
import org.easymock.{IAnswer, EasyMock}
import org.joda.time.DateTime
import org.apache.commons.codec.digest.DigestUtils
import org.junit.Test

import net.lshift.diffa.kernel.util.EasyMockScalaUtils._
import net.lshift.diffa.kernel.participants._
import net.lshift.diffa.kernel.util.Dates._
import net.lshift.diffa.kernel.util.DateUtils._
import net.lshift.diffa.kernel.events._
import net.lshift.diffa.kernel.config.Pair
import net.lshift.diffa.kernel.config.ConfigStore
import net.lshift.diffa.kernel.participants.EasyConstraints._
import net.lshift.diffa.kernel.util.Conversions._

/**
 * Base class for the various policy tests.
 */
abstract class AbstractPolicyTest {
  // The policy instance under test
  protected def policy:VersionPolicy

  // A method for generating a downstream version based on an upstream version
  protected def downstreamVersionFor(v:String):String

  // The various mocks for listeners and participants
  val usMock = createStrictMock("us", classOf[UpstreamParticipant])
  val dsMock = createStrictMock("ds", classOf[DownstreamParticipant])
  val nullListener = new NullDifferencingListener
  
  val store = createStrictMock("versionStore", classOf[VersionCorrelationStore])
  EasyMock.checkOrder(store, false)   // Store doesn't care about order
  val listener = createStrictMock("listener", classOf[DifferencingListener])

  val configStore = createStrictMock("configStore", classOf[ConfigStore])
  val abPair = "A-B"

  val categories = new HashMap[String,String]
  val pair = new Pair(categories=Map("bizDate" -> "date"))

  expect(configStore.getPair(abPair)).andReturn(pair).anyTimes
  replay(configStore)

  protected def replayAll = replay(usMock, dsMock, store, listener)
  protected def verifyAll = verify(usMock, dsMock, store, listener, configStore)

  // Make declaring of sequences of specific types clearer
  def DigestsFromParticipant[T](vals:T*) = Seq[T](vals:_*)
  def VersionsFromStore[T](vals:T*) = Seq[T](vals:_*)

  protected val yearly = YearlyCategoryFunction
  protected val monthly = MonthlyCategoryFunction
  protected val daily = DailyCategoryFunction
  protected val individual = IndividualCategoryFunction
  
  def Up(v:VersionID, d:DateTime, s:String) = UpstreamVersion(v, Seq(d.toString()), d, s)
  def Down(v:VersionID, d:DateTime, s1:String, s2:String) = DownstreamVersion(v, Seq(d.toString()), d, s1, s2)
  def UpInt(v:VersionID, i:Int, s:String) = UpstreamVersion(v, Seq(i.toString), new DateTime, s)
  def DownInt(v:VersionID, i:Int, s1:String, s2:String) = DownstreamVersion(v, Seq(i.toString), new DateTime, s1, s2)

  def bizDateMap(d:DateTime) = Map("bizDate" -> d.toString())
  def bizDateSeq(d:DateTime) = Seq(d.toString())

  protected def shouldOnlySyncTopLevelsWhenParticipantsAndStoresMatch(
      categories: Map[String, String],
      constraints: Seq[UnboundedRangeQueryConstraint],
      attributes: (Seq[String], Seq[String]),
      upstreamVersions: (UpstreamVersion, UpstreamVersion),
      downstreamVersions: (DownstreamVersion, DownstreamVersion)) {
    pair.categories = categories
    val (attribute1, attribute2) = attributes
    val (upstreamVersion1, upstreamVersion2) = upstreamVersions
    val (downstreamVersion1, downstreamVersion2) = downstreamVersions
    // Expect only a top-level sync between the pairs
    expectUpstreamAggregateSync(abPair, constraints,
      DigestsFromParticipant(
        AggregateDigest(attribute1, null, DigestUtils.md5Hex("vsn1")),
        AggregateDigest(attribute2, null, DigestUtils.md5Hex("vsn2"))),
      VersionsFromStore(upstreamVersion1, upstreamVersion2))
    expectDownstreamAggregateSync(abPair, constraints,
      DigestsFromParticipant(
        AggregateDigest(attribute1,  null, DigestUtils.md5Hex(downstreamVersionFor("vsn1"))),
        AggregateDigest(attribute2, null, DigestUtils.md5Hex(downstreamVersionFor("vsn2")))),
      VersionsFromStore(downstreamVersion1, downstreamVersion2))

    // We should still see an unmatched version check
    expect(store.unmatchedVersions(EasyMock.eq(abPair), EasyMock.eq(constraints))).andReturn(Seq())
    replayAll

    policy.difference(abPair, usMock, dsMock, nullListener)
    verifyAll
  }

  @Test
  def shouldOnlySyncTopLevelsWhenParticipantsAndStoresMatchForDateCategories =
    shouldOnlySyncTopLevelsWhenParticipantsAndStoresMatch(
      categories = Map("bizDate" -> "date"),
      constraints = List(unconstrainedDate(YearlyCategoryFunction)),
      attributes = (Seq("2009"), Seq("2010")),
      upstreamVersions = (Up(VersionID(abPair, "id1"), JUN_6_2009_1, "vsn1"),
                          Up(VersionID(abPair, "id2"), JUL_8_2010_1, "vsn2")),
      downstreamVersions = (
        DownstreamVersion(VersionID(abPair, "id1"), bizDateSeq(JUN_6_2009_1), JUN_6_2009_1, "vsn1", downstreamVersionFor("vsn1")),
        DownstreamVersion(VersionID(abPair, "id2"), bizDateSeq(JUL_8_2010_1), JUL_8_2010_1, "vsn2", downstreamVersionFor("vsn2")))
    )

  @Test
  def shouldOnlySyncTopLevelsWhenParticipantsAndStoresMatchForIntegerCategories =
    shouldOnlySyncTopLevelsWhenParticipantsAndStoresMatch(
      categories = Map("someInt" -> "int"),
      constraints = List(unconstrainedInt(ThousandsCategoryFunction)),
      attributes = (Seq("1000"), Seq("2000")),
      upstreamVersions = (UpInt(VersionID(abPair, "id1"), 1234, "vsn1"), UpInt(VersionID(abPair, "id2"), 2345, "vsn2")),
      downstreamVersions = (
        DownstreamVersion(VersionID(abPair, "id1"), Seq("1234"), JUN_6_2009_1, "vsn1", downstreamVersionFor("vsn1")),
        DownstreamVersion(VersionID(abPair, "id2"), Seq("2345"), JUL_8_2010_1, "vsn2", downstreamVersionFor("vsn2")))
    )

  @Test
  def shouldUpdateUpstreamVersionsWhenStoreIsOutOfDateWithUpstreamParticipant {
    val timestamp = new DateTime()
    // Expect only a top-level sync between the pairs
    expectUpstreamAggregateSync(abPair, List(unconstrainedDate(YearlyCategoryFunction)),
{'fraction_non_alphanumeric': 0.11059362755969543, 'fraction_numerical': 0.0213646780513048, 'mean_word_length': 5.121266968325791, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 10, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '833844', 'n_tokens_mistral': 4382, 'n_tokens_neox': 4110, 'n_words': 916}