<commit_before>/*
 * Copyright (c) Microsoft. All rights reserved.
 * Licensed under the MIT license. See LICENSE file in the project root for full license information.
 */
package org.apache.spark.streaming.api.csharp

import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.scheduler.{BatchInfo, StreamingListenerBatchCompleted}
import org.apache.spark.streaming.{StreamingContext, Duration, Time}
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.csharp.SparkCLRFunSuite

import scala.reflect.ClassTag

private class MockSparkContext(config: SparkConf) extends SparkContext(config) {
  var numParallelJobs = 0
  override def runJob[T, U: ClassTag](rdd: RDD[T], func: Iterator[T] => U): Array[U] = {
    numParallelJobs = numParallelJobs + 1
    Array.empty
  }
}

private class MockCSharpStateDStream(
    parent: DStream[Array[Byte]],
    reduceFunc: Array[Byte],
    serializationMode: String,
    serializationMode2: String)
  extends CSharpStateDStream(parent, reduceFunc, serializationMode, serializationMode2) {

  override def dependencies: List[DStream[_]] = List.empty
  override def slideDuration: Duration = new Duration(1000)
}

class CSharpDStreamSuite extends SparkCLRFunSuite {

  test("runParallelJob in UpdateStateByKey") {

    val conf = new SparkConf().setAppName("test").setMaster("local")
    val sc = new MockSparkContext(conf)<commit_msg>add conf.set("spark.testing", "true")<commit_after>/*
 * Copyright (c) Microsoft. All rights reserved.
 * Licensed under the MIT license. See LICENSE file in the project root for full license information.
 */
package org.apache.spark.streaming.api.csharp

import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.scheduler.{BatchInfo, StreamingListenerBatchCompleted}
import org.apache.spark.streaming.{StreamingContext, Duration, Time}
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.csharp.SparkCLRFunSuite

import scala.reflect.ClassTag

private class MockSparkContext(config: SparkConf) extends SparkContext(config) {
  var numParallelJobs = 0
  override def runJob[T, U: ClassTag](rdd: RDD[T], func: Iterator[T] => U): Array[U] = {
    numParallelJobs = numParallelJobs + 1
    Array.empty
  }
}

private class MockCSharpStateDStream(
    parent: DStream[Array[Byte]],
    reduceFunc: Array[Byte],
    serializationMode: String,
    serializationMode2: String)
  extends CSharpStateDStream(parent, reduceFunc, serializationMode, serializationMode2) {

  override def dependencies: List[DStream[_]] = List.empty
  override def slideDuration: Duration = new Duration(1000)
}

class CSharpDStreamSuite extends SparkCLRFunSuite {

  test("runParallelJob in UpdateStateByKey") {

    val conf = new SparkConf().setAppName("test").setMaster("local").set("spark.testing", "true")
    val sc = new MockSparkContext(conf)
{'fraction_non_alphanumeric': 0.1036936631650288, 'fraction_numerical': 0.005421890884445951, 'mean_word_length': 5.417391304347826, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '3637411', 'n_tokens_mistral': 943, 'n_tokens_neox': 893, 'n_words': 218}