All extractor tests

build.sbt

@@ -18,7 +18,6 @@ lazy val comet = (project in file("."))
       wesoUtils,
       // testing
       scalaTest,
-      scalaTestPlus,
       catsEffectTesting
     ),
     buildInfoSettings,
@@ -51,18 +50,19 @@ lazy val buildInfoSettings = Seq(
 // Core dependencies
 lazy val catsVersion = "3.3.11"
 // FS2 dependencies
-lazy val fs2Version = "3.2.5"
+lazy val fs2Version = "3.2.7"
 lazy val fs2KafkaVersion = "3.0.0-M6"
 // WESO dependencies
 lazy val shaclexVersion = "0.2.2"
 lazy val shexsVersion = "0.2.2"
-lazy val umlShaclZexVersion = "0.0.82"
+lazy val umlShaclexVersion = "0.0.82"
 lazy val wesoUtilsVersion = "0.2.4"
 // Testing dependencies
 lazy val scalaTestVersion = "3.2.11" // Usual testing
-lazy val scalaTestPlusVersion = "3.2.11.0" // Integration of ScalaTest with ScalaCheck
 lazy val catsEffectTestingVersion = "1.4.0" // Integration of ScalaTest with cats-effect
+
 // -------------------------------------------------------------------------- //
+
 // Core dependencies
 lazy val catsEffect = "org.typelevel" %% "cats-effect" % catsVersion
 // FS2 dependencies
@@ -74,5 +74,4 @@ lazy val shaclex = "es.weso" %% "shaclex" % shaclexVersion
 lazy val wesoUtils = "es.weso" %% "utilstest" % wesoUtilsVersion
 // Testing dependencies
 lazy val scalaTest = "org.scalatest" %% "scalatest" % scalaTestVersion % Test
-lazy val scalaTestPlus = "org.scalatestplus" %% "scalacheck-1-15" % scalaTestPlusVersion % Test
-lazy val catsEffectTesting = "org.typelevel" %% "cats-effect-testing-scalatest" % catsEffectTestingVersion % Test
+lazy val catsEffectTesting = "org.typelevel" %% "cats-effect-testing-scalatest" % catsEffectTestingVersion % Test

src/main/scala/Main.scala

@@ -18,9 +18,6 @@ import validation.configuration.ValidatorConfiguration
 
 import cats.effect.*
 import cats.syntax.functor.*
-import es.weso.rdf.PrefixMap
-import es.weso.rdf.jena.RDFAsJenaModel
-import es.weso.schema.{Schema, Schemas, ShapeMapTrigger, ValidationTrigger as ValidationTriggerShaclex}
 import fs2.io.file.*
 import fs2.{Pipe, Stream}
 

src/test/scala/extraction/ExtractorTests.scala

@@ -3,11 +3,13 @@ package extraction
 
 import data.DataFormat
 import data.DataFormat.*
+import exception.stream.timed.StreamTimeoutException
 import implicits.RDFElementImplicits.rdfFromString
 import schema.ShExSchemaFormat
 import schema.ShExSchemaFormat.*
 import stream.extractors.StreamExtractor
 import stream.extractors.file.FileExtractor
+import stream.extractors.kafka.{KafkaExtractor, KafkaExtractorConfiguration}
 import stream.extractors.list.ListExtractor
 import trigger.ShapeMapFormat.*
 import trigger.TriggerModeType.{SHAPEMAP, TARGET_DECLARATIONS}
@@ -25,6 +27,9 @@ import es.weso.schema.Schema
 import org.scalatest.freespec.AsyncFreeSpec
 import org.scalatest.matchers.should.Matchers
 
+import scala.concurrent.duration.*
+import scala.language.postfixOps
+
 /**
  * Test suite checking that the available basic data extractors work
  *
@@ -47,7 +52,8 @@ import org.scalatest.matchers.should.Matchers
  * @note In these tests we take for granted the functioning of the validator
  *       and we are just interested in the extractors
  * @note For obvious reasons, the Kafka Extractor can't be tested in a unit test
- *       and the Kafka extractor is therefore mocked with dummy data
+ *       (expect integration tests) and the Kafka extractor is therefore
+ *       expected to timeout and fail
  */
 //noinspection RedundantDefaultArgument
 class ExtractorTests extends AsyncFreeSpec with AsyncIOSpec with Matchers {
@@ -120,8 +126,40 @@ class ExtractorTests extends AsyncFreeSpec with AsyncIOSpec with Matchers {
   }
 
   "KAFKA extractor" - {
+    // Trivial example with a Kafka extractor
+    // Expected to timeout in the absence of a local Stream
     "works" in {
-      IO(1).asserting(_ shouldBe 1)
+      val validationResult: IO[Either[Throwable, ValidationResult]] =
+        for {
+          // Data, schema, trigger
+          schema <- Samples.SchemaSamples.mkSchemaShExIO()
+          trigger = Samples.TriggerSamples.mkTriggerShex()
+
+          // Kafka extractor
+          extractorConfiguration = KafkaExtractorConfiguration("topic")
+          extractor = KafkaExtractor[Unit, String](
+            extractorConfiguration,
+            TURTLE,
+            itemTimeout = Some(1 second)
+          )
+
+          // Validator init
+          validatorConfiguration = ValidatorConfiguration(schema, trigger)
+          validator = new Validator(validatorConfiguration, extractor)
+
+          results: List[Either[Throwable, ValidationResult]] <- validator.validate
+            .attempt
+            .compile
+            .toList
+        } yield results.head
+
+      // Result is an error and of type StreamTimeoutException
+      validationResult.asserting(results => {
+        results.isLeft shouldBe true
+
+        results.swap.toOption.get.isInstanceOf[StreamTimeoutException]
+          .shouldBe(true)
+      })
     }
   }
 }