Test: Add Influx db coverage (#4284)

Co-authored-by: osulzhenko <osulzhenko@magnite.com>

Author: marki1an

pom.xml

@@ -345,11 +345,22 @@
             <artifactId>postgresql</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>influxdb</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.mock-server</groupId>
             <artifactId>mockserver-client-java</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.influxdb</groupId>
+            <artifactId>influxdb-java</artifactId>
+            <version>2.23</version> <!-- works well with iZettle 1.1.1 and later -->
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

src/test/groovy/org/prebid/server/functional/model/response/influx/InfluxResponse.groovy

@@ -0,0 +1,6 @@
+package org.prebid.server.functional.model.response.influx
+
+class InfluxResponse {
+
+    List<InfluxResult> results
+}

src/test/groovy/org/prebid/server/functional/model/response/influx/InfluxResult.groovy

@@ -0,0 +1,10 @@
+package org.prebid.server.functional.model.response.influx
+
+import com.fasterxml.jackson.annotation.JsonProperty
+
+class InfluxResult {
+
+    @JsonProperty("statement_id")
+    Integer statementId
+    List<Series> series
+}

src/test/groovy/org/prebid/server/functional/model/response/influx/Series.groovy

@@ -0,0 +1,9 @@
+package org.prebid.server.functional.model.response.influx
+
+class Series {
+
+    String name
+    Tags tags
+    List<String> columns
+    List<List<String>> values
+}

src/test/groovy/org/prebid/server/functional/model/response/influx/Tags.groovy

@@ -0,0 +1,6 @@
+package org.prebid.server.functional.model.response.influx
+
+class Tags {
+
+    String measurement
+}

src/test/groovy/org/prebid/server/functional/service/PrebidServerService.groovy

@@ -25,6 +25,7 @@ import org.prebid.server.functional.model.response.cookiesync.CookieSyncResponse
 import org.prebid.server.functional.model.response.cookiesync.RawCookieSyncResponse
 import org.prebid.server.functional.model.response.currencyrates.CurrencyRatesResponse
 import org.prebid.server.functional.model.response.getuids.GetuidResponse
+import org.prebid.server.functional.model.response.influx.InfluxResponse
 import org.prebid.server.functional.model.response.infobidders.BidderInfoResponse
 import org.prebid.server.functional.model.response.setuid.SetuidResponse
 import org.prebid.server.functional.model.response.status.StatusResponse
@@ -42,6 +43,8 @@ import java.time.format.DateTimeFormatter
 
 import static io.restassured.RestAssured.given
 import static java.time.ZoneOffset.UTC
+import static org.prebid.server.functional.testcontainers.Dependencies.influxdbContainer
+
 
 class PrebidServerService implements ObjectMapperWrapper {
 
@@ -59,10 +62,12 @@ class PrebidServerService implements ObjectMapperWrapper {
     static final String HTTP_INTERACTION_ENDPOINT = "/logging/httpinteraction"
     static final String COLLECTED_METRICS_ENDPOINT = "/collected-metrics"
     static final String PROMETHEUS_METRICS_ENDPOINT = "/metrics"
+    static final String INFLUX_DB_ENDPOINT = "/query"
     static final String UIDS_COOKIE_NAME = "uids"
 
     private final PrebidServerContainer pbsContainer
     private final RequestSpecification requestSpecification
+    private final RequestSpecification influxRequestSpecification
     private final RequestSpecification adminRequestSpecification
     private final RequestSpecification prometheusRequestSpecification
 
@@ -75,6 +80,8 @@ class PrebidServerService implements ObjectMapperWrapper {
         this.pbsContainer = pbsContainer
         requestSpecification = new RequestSpecBuilder().setBaseUri(pbsContainer.rootUri)
                 .build()
+        influxRequestSpecification = new RequestSpecBuilder().setBaseUri(pbsContainer.influxUri)
+                .build()
         adminRequestSpecification = buildAndGetRequestSpecification(pbsContainer.adminRootUri, authenticationScheme)
         prometheusRequestSpecification = buildAndGetRequestSpecification(pbsContainer.prometheusRootUri, authenticationScheme)
     }
@@ -290,6 +297,16 @@ class PrebidServerService implements ObjectMapperWrapper {
         decode(response.asString(), new TypeReference<Map<String, Number>>() {})
     }
 
+    Map<String, Number> sendInfluxMetricsRequest() {
+        def response = given(influxRequestSpecification)
+                .queryParams(["db": influxdbContainer.getDatabase(),
+                              "q" : "SELECT COUNT(count) FROM /.*/  WHERE count >= 1 GROUP BY \"measurement\""])
+                .get(INFLUX_DB_ENDPOINT)
+
+        checkResponseStatusCode(response)
+        collectInToMap(decode(response.getBody().asString(), InfluxResponse))
+    }
+
     String sendPrometheusMetricsRequest() {
         def response = given(prometheusRequestSpecification).get(PROMETHEUS_METRICS_ENDPOINT)
 
@@ -431,6 +448,15 @@ class PrebidServerService implements ObjectMapperWrapper {
         }
     }
 
+    Boolean isContainMetricByValue(String value) {
+        try {
+            PBSUtils.waitUntil({ sendInfluxMetricsRequest()[value] != null })
+            true
+        } catch (IllegalStateException ignored) {
+            false
+        }
+    }
+
     private String getPbsLogsByValue(String value) {
         pbsContainer.logs.split("\n").find { it.contains(value) }
     }
@@ -454,4 +480,10 @@ class PrebidServerService implements ObjectMapperWrapper {
                 .setAuth(authScheme)
                 .build()
     }
+
+    private static Map<String, Number> collectInToMap(InfluxResponse responseBody) {
+        responseBody?.results?.first()?.series?.collectEntries {
+            [(it.name): it.values?.first()?.getAt(1) as Integer]
+        } ?: [:]
+    }
 }

src/test/groovy/org/prebid/server/functional/testcontainers/Dependencies.groovy

@@ -2,6 +2,7 @@ package org.prebid.server.functional.testcontainers
 
 import org.prebid.server.functional.testcontainers.container.NetworkServiceContainer
 import org.prebid.server.functional.util.SystemProperties
+import org.testcontainers.containers.InfluxDBContainer
 import org.testcontainers.containers.MySQLContainer
 import org.testcontainers.containers.Network
 import org.testcontainers.containers.localstack.LocalStackContainer
@@ -34,6 +35,13 @@ class Dependencies {
             .withInitScript("org/prebid/server/functional/db_psql_schema.sql")
             .withNetwork(network)
 
+    static final InfluxDBContainer influxdbContainer = new InfluxDBContainer<>(DockerImageName.parse("influxdb:1.8.10"))
+            .withUsername("prebid")
+            .withPassword("prebid")
+            .withAuthEnabled(false)
+            .withDatabase("prebid")
+            .withNetwork(network)
+
     static final NetworkServiceContainer networkServiceContainer = new NetworkServiceContainer(MOCKSERVER_VERSION)
             .withNetwork(network)
 
@@ -44,13 +52,13 @@ class Dependencies {
             localStackContainer = new LocalStackContainer(DockerImageName.parse("localstack/localstack:s3-latest"))
                     .withNetwork(network)
                     .withServices(S3)
-            Startables.deepStart([networkServiceContainer, mysqlContainer, localStackContainer]).join()
+            Startables.deepStart([networkServiceContainer, mysqlContainer, localStackContainer, influxdbContainer]).join()
         }
     }
 
     static void stop() {
         if (IS_LAUNCH_CONTAINERS) {
-            [networkServiceContainer, mysqlContainer, localStackContainer].parallelStream()
+            [networkServiceContainer, mysqlContainer, localStackContainer, influxdbContainer].parallelStream()
                     .forEach({ it.stop() })
         }
     }

src/test/groovy/org/prebid/server/functional/testcontainers/PbsConfig.groovy

@@ -1,5 +1,6 @@
 package org.prebid.server.functional.testcontainers
 
+import org.testcontainers.containers.InfluxDBContainer
 import org.testcontainers.containers.MySQLContainer
 import org.testcontainers.containers.PostgreSQLContainer
 
@@ -100,7 +101,6 @@ LIMIT 1
          "settings.database.idle-connection-timeout": "300"
         ].asImmutable()
     }
-
     static Map<String, String> getPostgreSqlConfig(PostgreSQLContainer postgres = Dependencies.postgresqlContainer) {
         ["settings.database.type"                   : "postgres",
          "settings.database.host"                   : postgres.getNetworkAliases().get(0),
@@ -145,7 +145,7 @@ LIMIT 1
          "currency-converter.external-rates.refresh-period-ms" : "900000"]
     }
 
-    static Map<String,String> getTargetingConfig() {
+    static Map<String, String> getTargetingConfig() {
         ["settings.targeting.truncate-attr-chars": '255']
     }
 

src/test/groovy/org/prebid/server/functional/testcontainers/container/PrebidServerContainer.groovy

@@ -75,6 +75,10 @@ class PrebidServerContainer extends GenericContainer<PrebidServerContainer> {
         getMappedPort(PROMETHEUS_PORT)
     }
 
+    String getInfluxUri() {
+        return "http://$host:$Dependencies.influxdbContainer.firstMappedPort"
+    }
+
     String getRootUri() {
         return "http://$host:$port"
     }

src/test/groovy/org/prebid/server/functional/tests/InfluxDBSpec.groovy

@@ -0,0 +1,78 @@
+package org.prebid.server.functional.tests
+
+import org.prebid.server.functional.model.AccountStatus
+import org.prebid.server.functional.model.config.AccountConfig
+import org.prebid.server.functional.model.db.Account
+import org.prebid.server.functional.model.request.auction.BidRequest
+import org.prebid.server.functional.service.PrebidServerException
+import org.prebid.server.functional.service.PrebidServerService
+
+import static io.netty.handler.codec.http.HttpResponseStatus.UNAUTHORIZED
+import static org.prebid.server.functional.testcontainers.Dependencies.influxdbContainer
+
+class InfluxDBSpec extends BaseSpec {
+
+    private static final String ACCOUNT_REJECTED_METRIC = "influx.metric.account.%s.requests.rejected.invalid-account"
+
+    private static final Map<String, String> PBS_CONFIG_WITH_INFLUX_AND_ENFORCE_VALIDATION_ACCOUNTANT = [
+            "metrics.influxdb.enabled"       : "true",
+            "metrics.influxdb.prefix"        : "influx.metric.",
+            "metrics.influxdb.host"          : influxdbContainer.getNetworkAliases().get(0),
+            "metrics.influxdb.port"          : influxdbContainer.getExposedPorts().get(0) as String,
+            "metrics.influxdb.protocol"      : "http",
+            "metrics.influxdb.database"      : influxdbContainer.database as String,
+            "metrics.influxdb.auth"          : "${influxdbContainer.username}:${influxdbContainer.password}" as String,
+            "metrics.influxdb.interval"      : "1",
+            "metrics.influxdb.connectTimeout": "5000",
+            "metrics.influxdb.readTimeout"   : "100",
+            "settings.enforce-valid-account": true as String]
+
+    private static final PrebidServerService pbsServiceWithEnforceValidAccount
+            = pbsServiceFactory.getService(PBS_CONFIG_WITH_INFLUX_AND_ENFORCE_VALIDATION_ACCOUNTANT)
+
+    def cleanupSpec() {
+        pbsServiceFactory.removeContainer(PBS_CONFIG_WITH_INFLUX_AND_ENFORCE_VALIDATION_ACCOUNTANT)
+    }
+
+    def "PBS should reject request with error and metrics when inactive account"() {
+        given: "Default basic BidRequest with inactive account id"
+        def bidRequest = BidRequest.defaultBidRequest
+
+        and: "Inactive account id"
+        def account = new Account(uuid: bidRequest.accountId, config: new AccountConfig(status: AccountStatus.INACTIVE))
+        accountDao.save(account)
+
+        when: "PBS processes auction request"
+        pbsServiceWithEnforceValidAccount.sendAuctionRequest(bidRequest)
+
+        then: "PBS should reject the entire auction"
+        def exception = thrown(PrebidServerException)
+        assert exception.statusCode == UNAUTHORIZED.code()
+        assert exception.responseBody == "Account $bidRequest.accountId is inactive"
+
+        and: "PBS wait until get metric"
+        assert pbsServiceWithEnforceValidAccount.isContainMetricByValue(ACCOUNT_REJECTED_METRIC.formatted(bidRequest.accountId))
+
+        and: "PBS metrics populated correctly"
+        def influxMetricsRequest = pbsServiceWithEnforceValidAccount.sendInfluxMetricsRequest()
+        assert influxMetricsRequest[ACCOUNT_REJECTED_METRIC.formatted(bidRequest.accountId)] == 1
+    }
+
+    def "PBS shouldn't reject request with error and metrics when active account"() {
+        given: "Default basic BidRequest with inactive account id"
+        def bidRequest = BidRequest.defaultBidRequest
+
+        and: "Inactive account id"
+        def account = new Account(uuid: bidRequest.accountId, config: new AccountConfig(status: AccountStatus.ACTIVE))
+        accountDao.save(account)
+
+        when: "PBS processes auction request"
+        def response = pbsServiceWithEnforceValidAccount.sendAuctionRequest(bidRequest)
+
+        then: "Bid response should contain seatBid"
+        assert response.seatbid.size() == 1
+
+        and: "PBs shouldn't emit metric"
+        assert !pbsServiceWithEnforceValidAccount.isContainMetricByValue(ACCOUNT_REJECTED_METRIC.formatted(bidRequest.accountId))
+    }
+}