Make integration tests faster

tests/docker/.env

@@ -1,4 +1,3 @@
-CLUSTER_SIZE=1
 DELTA_PLUGINS=/opt/docker/plugins/
 DELTA_EXTERNAL_CONF=/config/delta-postgres.conf
 KAMON_ENABLED=false

tests/docker/config/delta-postgres.conf

@@ -58,10 +58,16 @@ app {
   }
 
   projections {
+    cluster {
+      # Number of nodes in the cluster
+      size = ${?CLUSTER_SIZE}
+      # Index of the current Delta instance in the cluster
+      node-index = ${?NODE_INDEX}
+    }
     batch {
       max-elements = 30
       # the maximum batching duration.
-      max-interval = 1 second
+      max-interval = 500 millis
     }
   }
 
@@ -73,7 +79,7 @@ app {
     }
 
     query {
-      refresh-strategy = 250 millis
+      refresh-strategy = 50 millis
     }
   }
 
@@ -116,7 +122,7 @@ plugins {
 
     batch {
       max-elements = 5
-      max-interval = 1 second
+      max-interval = 500 millis
     }
   }
 
@@ -137,10 +143,11 @@ plugins {
       username = "elastic"
       password = "password"
     }
+    sync-indexing-refresh = "true"
 
     batch {
       max-elements = 5
-      max-interval = 1 second
+      max-interval = 500 millis
     }
   }
 

tests/docker/config/delta-startup.sh

@@ -0,0 +1,8 @@
+#!/usr/bin/env bash
+
+ELASTICSEARCH_INDEX_TEMPLATE_URL="http://elastic:password@elasticsearch:9200/_index_template/test_template"
+
+# Create index template to have a quicker refresh and no replica
+curl -f --silent -H "Content-Type: application/json" $ELASTICSEARCH_INDEX_TEMPLATE_URL --upload-file /config/elasticsearch/template.json &&
+ln -sf /opt/docker/plugins/disabled/project-deletion.jar /opt/docker/plugins/project-deletion.jar &&
+/opt/docker/bin/delta-app -Xmx4G

tests/src/test/resources/elasticsearch/template.json → tests/docker/config/elasticsearch/template.json

@@ -1,5 +1,5 @@
 {
-  "index_patterns" : ["*"],
+  "index_patterns" : ["delta_*"],
   "priority" : 1,
   "template": {
     "settings" : {

tests/docker/delta-blazegraph-compose.yml

@@ -32,12 +32,7 @@ services:
       - SPARQL_TARGET=blazegraph
       - SPARQL_ENDPOINT=http://blazegraph:9999/blazegraph
     image: bluebrain/nexus-delta:latest
-    entrypoint:
-      - '/bin/bash'
-      - '-c'
-      - |
-        ln -sf /opt/docker/plugins/disabled/project-deletion.jar /opt/docker/plugins/project-deletion.jar &&
-        /opt/docker/bin/delta-app -Xmx4G -Dapp.projections.cluster.size=${CLUSTER_SIZE}
+    entrypoint: ["/config/delta-startup.sh"]
     ports:
       - 8080:8080
     volumes:

tests/docker/delta-rdf4j-compose.yml

@@ -33,12 +33,7 @@ services:
       - SPARQL_TARGET=rdf4j
       - SPARQL_ENDPOINT=http://rdf4j:8080/rdf4j-server
     image: bluebrain/nexus-delta:latest
-    entrypoint:
-      - '/bin/bash'
-      - '-c'
-      - |
-        ln -sf /opt/docker/plugins/disabled/project-deletion.jar /opt/docker/plugins/project-deletion.jar &&
-        /opt/docker/bin/delta-app -Xmx4G -Dapp.projections.cluster.size=${CLUSTER_SIZE}
+    entrypoint: ["/config/delta-startup.sh"]
     ports:
       - 8080:8080
     volumes:

tests/docker/elasticsearch.yml

@@ -2,7 +2,7 @@ services:
   elasticsearch:
     image: docker.elastic.co/elasticsearch/elasticsearch:8.17.1
     environment:
-      ES_JAVA_OPTS: "-Xmx2G"
+      ES_JAVA_OPTS: "-Xmx4G"
       discovery.type: "single-node"
       bootstrap.memory_lock: "true"
       xpack.security.enabled: "true"
@@ -15,7 +15,3 @@ services:
       retries: 60
     ports:
       - 9200:9200
-    deploy:
-      resources:
-        limits:
-          memory: 4G

tests/docker/rdf4j.yml

@@ -2,7 +2,7 @@ services:
   rdf4j:
     image: eclipse/rdf4j-workbench:5.1.0
     environment:
-      JAVA_OPTS: "-Xms1g -Xmx16g"
+      JAVA_OPTS: "-Xms1g -Xmx4g"
     ports:
       - 7070:8080
     healthcheck:

tests/src/test/scala/ch/epfl/bluebrain/nexus/tests/BaseIntegrationSpec.scala

@@ -85,7 +85,7 @@ trait BaseIntegrationSpec
   val elasticsearchViewsDsl = new ElasticSearchViewsDsl(deltaClient)
   val storagesDsl           = new StoragesDsl(deltaClient)
 
-  implicit override def patienceConfig: PatienceConfig = PatienceConfig(config.patience, 300.millis)
+  implicit override def patienceConfig: PatienceConfig = PatienceConfig(config.patience, 100.millis)
 
   def eventually(io: IO[Assertion])(implicit pos: Position): Assertion =
     eventually { io.unsafeRunSync() }
@@ -95,7 +95,6 @@ trait BaseIntegrationSpec
   override def beforeAll(): Unit = {
     super.beforeAll()
     val setup = for {
-      _ <- elasticsearchDsl.createTemplate()
       _ <- initRealm(
              Realm.internal,
              Identity.Anonymous,

tests/src/test/scala/ch/epfl/bluebrain/nexus/tests/ElasticsearchDsl.scala

@@ -23,28 +23,12 @@ class ElasticsearchDsl(implicit
     with CirceUnmarshalling
     with Matchers {
 
-  private val loader = ClasspathResourceLoader()
-
   private val logger = Logger[this.type]
 
   private val elasticUrl    = s"http://${sys.props.getOrElse("elasticsearch-url", "localhost:9200")}"
   private val elasticClient = HttpClient(elasticUrl)
   private val credentials   = BasicHttpCredentials("elastic", "password")
 
-  def createTemplate(): IO[StatusCode] = {
-    for {
-      json   <- loader.jsonContentOf("elasticsearch/template.json")
-      _      <- logger.info("Creating template for Elasticsearch indices")
-      result <- elasticClient(
-                  HttpRequest(
-                    method = PUT,
-                    uri = s"$elasticUrl/_index_template/test_template",
-                    entity = HttpEntity(ContentTypes.`application/json`, json.noSpaces)
-                  ).addCredentials(credentials)
-                ).map(_.status)
-    } yield result
-  }
-
   def includes(indices: String*): IO[Assertion] =
     allIndices.map { all =>
       all should contain allElementsOf (indices)

tests/src/test/scala/ch/epfl/bluebrain/nexus/tests/kg/AggregationsSpec.scala

@@ -1,6 +1,7 @@
 package ch.epfl.bluebrain.nexus.tests.kg
 
 import akka.http.scaladsl.model.StatusCodes
+import cats.syntax.all._
 import ch.epfl.bluebrain.nexus.tests.Identity.Anonymous
 import ch.epfl.bluebrain.nexus.tests.Identity.aggregations.{Charlie, Rose}
 import ch.epfl.bluebrain.nexus.tests.admin.ProjectPayload
@@ -37,13 +38,12 @@ final class AggregationsSpec extends BaseIntegrationSpec {
       _ <- aclDsl.addPermission(s"/$ref12", Rose, Views.Query)
     } yield ()
 
-    val resourcePayload = SimpleResource.sourcePayload(5).accepted
-    val postResources   = for {
-      // Creation
-      _ <- deltaClient.put[Json](s"/resources/$ref11/_/r11_1", resourcePayload, Charlie)(expectCreated)
-      _ <- deltaClient.put[Json](s"/resources/$ref11/_/r11_2", resourcePayload, Charlie)(expectCreated)
-      _ <- deltaClient.put[Json](s"/resources/$ref12/_/r12_1", resourcePayload, Charlie)(expectCreated)
-      _ <- deltaClient.put[Json](s"/resources/$ref21/_/r21_1", resourcePayload, Charlie)(expectCreated)
+    val postResources = for {
+      resourcePayload <- SimpleResource.sourcePayload(5)
+      resources        = List(ref11 -> "r11_1", ref11 -> "r11_2", ref12 -> "r12_1", ref21 -> "r21_1")
+      _               <- resources.parTraverse { case (proj, id) =>
+                           deltaClient.put[Json](s"/resources/$proj/_/$id", resourcePayload, Charlie)(expectCreated)
+                         }
     } yield ()
 
     (setup >> postResources).accepted

tests/src/test/scala/ch/epfl/bluebrain/nexus/tests/kg/EventsSpec.scala

@@ -69,7 +69,7 @@ class EventsSpec extends BaseIntegrationSpec {
     }
 
     "fetch resource events filtered by project" in eventually {
-      deltaClient.sseEvents(s"/resources/$id/events", BugsBunny, initialEventId, take = 12L) { sses =>
+      deltaClient.sseEvents(s"/resources/$id/events", BugsBunny, initialEventId, take = 6L) { sses =>
         sses.size shouldEqual 6
         sses.flatMap(_._1) should contain theSameElementsInOrderAs List(
           "ResourceCreated",
@@ -85,7 +85,7 @@ class EventsSpec extends BaseIntegrationSpec {
     }
 
     "fetch resource events filtered by organization 1" in {
-      deltaClient.sseEvents(s"/resources/$orgId/events", BugsBunny, initialEventId, take = 12L) { sses =>
+      deltaClient.sseEvents(s"/resources/$orgId/events", BugsBunny, initialEventId, take = 6L) { sses =>
         sses.size shouldEqual 6
         sses.flatMap(_._1) should contain theSameElementsInOrderAs List(
           "ResourceCreated",
@@ -101,7 +101,7 @@ class EventsSpec extends BaseIntegrationSpec {
     }
 
     "fetch resource events filtered by organization 2" in {
-      deltaClient.sseEvents(s"/resources/$orgId2/events", BugsBunny, initialEventId, take = 7L) { sses =>
+      deltaClient.sseEvents(s"/resources/$orgId2/events", BugsBunny, initialEventId, take = 1L) { sses =>
         sses.size shouldEqual 1
         sses.flatMap(_._1) should contain theSameElementsInOrderAs List("ResourceCreated")
         val json = Json.arr(sses.flatMap(_._2.map(events.filterFields)): _*)

tests/src/test/scala/ch/epfl/bluebrain/nexus/tests/kg/DefaultIndexSpec.scala → tests/src/test/scala/ch/epfl/bluebrain/nexus/tests/kg/MainIndexSpec.scala

@@ -3,6 +3,7 @@ package ch.epfl.bluebrain.nexus.tests.kg
 import akka.http.scaladsl.model.StatusCodes
 import ch.epfl.bluebrain.nexus.delta.kernel.utils.UrlUtils
 import ch.epfl.bluebrain.nexus.tests.BaseIntegrationSpec
+import cats.syntax.all._
 import ch.epfl.bluebrain.nexus.tests.Identity.aggregations.Charlie
 import ch.epfl.bluebrain.nexus.tests.Identity.listings.{Alice, Bob}
 import ch.epfl.bluebrain.nexus.tests.Optics.{filterNestedKeys, hitProjects}
@@ -11,7 +12,7 @@ import ch.epfl.bluebrain.nexus.tests.iam.types.Permission.Organizations
 import ch.epfl.bluebrain.nexus.tests.resources.SimpleResource
 import io.circe.Json
 
-class DefaultIndexSpec extends BaseIntegrationSpec {
+class MainIndexSpec extends BaseIntegrationSpec {
 
   private val org1   = genId()
   private val proj11 = genId()
@@ -29,10 +30,10 @@ class DefaultIndexSpec extends BaseIntegrationSpec {
       _               <- adminDsl.createProject(org1, proj11, ProjectPayload.generate(proj11), Bob)
       _               <- adminDsl.createProject(org1, proj12, ProjectPayload.generate(proj12), Bob)
       resourcePayload <- SimpleResource.sourcePayload(5)
-      _               <- deltaClient.put[Json](s"/resources/$ref11/_/r11_1", resourcePayload, Bob)(expectCreated)
-      _               <- deltaClient.put[Json](s"/resources/$ref11/_/r11_2", resourcePayload, Bob)(expectCreated)
-      _               <- deltaClient.put[Json](s"/resources/$ref12/_/r12_1", resourcePayload, Bob)(expectCreated)
-      _               <- deltaClient.put[Json](s"/resources/$ref12/_/r12_2", resourcePayload, Bob)(expectCreated)
+      resources        = List(ref11 -> "r11_1", ref11 -> "r11_2", ref12 -> "r12_1", ref12 -> "r12_2")
+      _               <- resources.parTraverse { case (proj, id) =>
+                           deltaClient.put[Json](s"/resources/$proj/_/$id", resourcePayload, Bob)(expectCreated)
+                         }
     } yield ()
     setup.accepted
   }

tests/src/test/scala/ch/epfl/bluebrain/nexus/tests/kg/ResourcesSpec.scala

@@ -647,24 +647,16 @@ class ResourcesSpec extends BaseIntegrationSpec {
 
   "check consistency of responses" in {
     (2 to 100).toList.traverse { resourceId =>
-      val payload =
-        SimpleResource.sourcePayload(s"https://dev.nexus.test.com/simplified-resource/$resourceId", 3).accepted
       for {
-        _ <-
-          deltaClient
-            .put[Json](
-              s"/resources/$project1/test-schema/test-resource:$resourceId?indexing=sync",
-              payload,
-              Rick
-            ) {
-              expectCreated
-            }
-        _ <- deltaClient.get[Json](s"/resources/$project1/test-schema", Rick) { (json, response) =>
-               response.status shouldEqual StatusCodes.OK
-               val received = json.asObject.value("_total").value.asNumber.value.toInt.value
-               val expected = resourceId
-               received shouldEqual expected
-             }
+        payload       <- SimpleResource.sourcePayload(s"https://dev.nexus.test.com/simplified-resource/$resourceId", 3)
+        createEndpoint = s"/resources/$project1/test-schema/test-resource:$resourceId?indexing=sync"
+        _             <- deltaClient.put[Json](createEndpoint, payload, Rick) { expectCreated }
+        _             <- deltaClient.get[Json](s"/resources/$project1/test-schema", Rick) { (json, response) =>
+                           response.status shouldEqual StatusCodes.OK
+                           val received = json.asObject.value("_total").value.asNumber.value.toInt.value
+                           val expected = resourceId
+                           received shouldEqual expected
+                         }
       } yield succeed
     }
   }

tests/src/test/scala/ch/epfl/bluebrain/nexus/tests/kg/SearchAccessSpec.scala

@@ -90,11 +90,7 @@ class SearchAccessSpec extends BaseIntegrationSpec {
   /** Post resource across all projects defined in the test suite */
   private def postResource(resourcePath: String) =
     projects.parTraverse { project =>
-      for {
-        _ <- deltaClient.post[Json](s"/resources/$project/_/", jsonContentOf(resourcePath), Rick) { (_, response) =>
-               response.status shouldEqual StatusCodes.Created
-             }
-      } yield succeed
+      deltaClient.post[Json](s"/resources/$project/_/", jsonContentOf(resourcePath), Rick) { expectCreated }
     }
 
   /** Lens to get all project labels in a given json */

tests/src/test/scala/ch/epfl/bluebrain/nexus/tests/kg/SearchConfigIndexingSpec.scala

@@ -1174,7 +1174,7 @@ class SearchConfigIndexingSpec extends BaseIntegrationSpec {
       val prefix = "SSCx-HexO1"
       val search = s"${prefix}-Release-TC__ConnRewire"
 
-      (prefix.length - 1 to search.length).toList.traverse { index =>
+      (prefix.length - 1 to search.length).toList.parTraverse { index =>
         val s              = search.subSequence(0, index).toString
         val query          = fullTextQuery(s)
         val queryLowercase = fullTextQuery(s.toLowerCase)
@@ -1188,7 +1188,7 @@ class SearchConfigIndexingSpec extends BaseIntegrationSpec {
       val prefix = "1818-"
       val search = s"${prefix}ReticularNeuron2_shrink"
 
-      (prefix.length - 1 to search.length).toList.traverse { index =>
+      (prefix.length - 1 to search.length).toList.parTraverse { index =>
         val s              = search.subSequence(0, index).toString
         val query          = fullTextQuery(s)
         val queryLowercase = fullTextQuery(s.toLowerCase)