bloodhound-1.0.0.0: tests/Test/IngestPipelineSpec.hs
{-# LANGUAGE OverloadedStrings #-}
module Test.IngestPipelineSpec (spec) where
import Control.Monad.IO.Class (liftIO)
import Data.Aeson (Value (..), decode, encode, object, (.=))
import Data.Aeson.Key qualified as K
import Data.Aeson.KeyMap qualified as KM
import Data.Aeson.Types (parseMaybe)
import Data.ByteString.Lazy.Char8 qualified as LBS
import Data.Map.Strict qualified as Map
import Data.Maybe (isJust)
import Data.Text qualified as Text
import Data.Time.Clock.POSIX (getPOSIXTime)
import Data.Vector qualified as V
import Database.Bloodhound.Client.Cluster (BackendType (..))
import Database.Bloodhound.Common.Client qualified as CommonClient
import Database.Bloodhound.Common.Requests qualified as Common
import TestsUtils.Common
import TestsUtils.Import
import Prelude
-- | Fixture mimicking the @GET /_ingest/pipeline@ response (two pipelines,
-- keyed by id). Adapted from the shape documented at
-- https://www.elastic.co/guide/en/elasticsearch/reference/current/get-pipeline-api.html.
sampleListResponse :: LBS.ByteString
sampleListResponse =
"{\
\ \"my-pipeline\": {\
\ \"description\": \"Pipeline for synthetic test fixture\",\
\ \"version\": 1,\
\ \"last_modified\": \"2023-06-01T12:34:56.789Z\",\
\ \"processors\": [\
\ { \"set\": { \"field\": \"foo\", \"value\": \"bar\" } }\
\ ]\
\ },\
\ \"other-pipeline\": {\
\ \"version\": 3,\
\ \"processors\": [\
\ { \"uppercase\": { \"field\": \"name\" } }\
\ ]\
\ }\
\}"
-- | Fixture mimicking the @GET /_ingest/pipeline/{id}@ response. ES still
-- wraps the single pipeline in a one-entry object keyed by id.
sampleSingleResponse :: LBS.ByteString
sampleSingleResponse =
"{\
\ \"my-pipeline\": {\
\ \"description\": \"single pipeline fixture\",\
\ \"version\": 2,\
\ \"last_modified\": \"2024-01-15T08:00:00.000Z\",\
\ \"processors\": [\
\ { \"uppercase\": { \"field\": \"name\" } }\
\ ]\
\ }\
\}"
-- | Minimal ingest pipeline body used by the live integration tests: a
-- single @uppercase@ processor that upper-cases the @name@ field. Small
-- enough to be accepted by every supported ES / OpenSearch version, and
-- observable enough that the simulate round-trip can assert the
-- transformation was applied.
sampleUppercasePipeline :: IngestPipeline
sampleUppercasePipeline =
IngestPipeline $
object
[ "description" .= ("uppercase the name field" :: Text.Text),
"processors"
.= [ object
[ "uppercase"
.= object ["field" .= ("name" :: Text.Text)]
]
]
]
spec :: Spec
spec = describe "Ingest Pipelines API" $ do
describe "PipelineId JSON" $ do
it "round-trips as a bare JSON string" $ do
let Just decoded = decode "\"my-pipeline\"" :: Maybe PipelineId
unPipelineId decoded `shouldBe` "my-pipeline"
encode (PipelineId "my-pipeline") `shouldBe` "\"my-pipeline\""
it "rejects a non-string value" $ do
let decoded = decode "42" :: Maybe PipelineId
decoded `shouldBe` Nothing
describe "IngestPipeline JSON (PUT body)" $ do
it "encodes the carried Value verbatim (no 'pipeline' wrapping)" $ do
let body =
object
[ "description" .= ("synthetic" :: Text.Text),
"processors"
.= object ["set" .= object ["field" .= ("foo" :: Text.Text), "value" .= ("bar" :: Text.Text)]]
]
encoded = encode (IngestPipeline body)
encoded `shouldBe` encode body
it "decodes an object body verbatim" $ do
let Just decoded = decode samplePutBodyBytes :: Maybe IngestPipeline
Just expected = decode samplePutBodyBytes :: Maybe Value
unIngestPipeline decoded `shouldBe` expected
it "round-trips through ToJSON/FromJSON" $ do
let Just decoded = decode samplePutBodyBytes :: Maybe IngestPipeline
(decode . encode) decoded `shouldBe` Just decoded
it "rejects a non-object body (array)" $ do
let decoded = decode "[1,2,3]" :: Maybe IngestPipeline
decoded `shouldBe` Nothing
it "rejects a non-object body (bare number)" $ do
let decoded = decode "5" :: Maybe IngestPipeline
decoded `shouldBe` Nothing
it "rejects invalid JSON" $ do
let decoded = decode "{ this is not json" :: Maybe IngestPipeline
decoded `shouldBe` Nothing
describe "IngestPipelineInfo JSON" $ do
it "decodes a single pipeline body (id stays as the placeholder)" $ do
let Just (Object o) = decode sampleSingleResponse
Just body = KM.lookup "my-pipeline" o
decoded = parseMaybe parseJSON body :: Maybe IngestPipelineInfo
ingestPipelineInfoId <$> decoded `shouldBe` Just (PipelineId "")
ingestPipelineInfoVersion <$> decoded `shouldBe` Just (Just 2)
ingestPipelineInfoDescription <$> decoded `shouldBe` Just (Just "single pipeline fixture")
ingestPipelineInfoLastModified <$> decoded `shouldBe` Just (Just "2024-01-15T08:00:00.000Z")
(ingestPipelineInfoProcessors <$> decoded) `shouldSatisfy` maybe False (maybe False hasProcessorsShape)
it "tolerates a body missing every optional field except processors" $ do
let decoded = decode "{ \"processors\": [] }" :: Maybe IngestPipelineInfo
ingestPipelineInfoId <$> decoded `shouldBe` Just (PipelineId "")
ingestPipelineInfoVersion <$> decoded `shouldBe` Just Nothing
ingestPipelineInfoDescription <$> decoded `shouldBe` Just Nothing
ingestPipelineInfoLastModified <$> decoded `shouldBe` Just Nothing
it "tolerates a body with no processors field at all" $ do
let decoded = decode "{ \"description\": \"empty\" }" :: Maybe IngestPipelineInfo
ingestPipelineInfoProcessors <$> decoded `shouldBe` Just Nothing
describe "IngestPipelines JSON (keyed-by-id response walk)" $ do
it "decodes a multi-entry response, stamping each id from the JSON key" $ do
let Just decoded = decode sampleListResponse :: Maybe IngestPipelines
entries = unIngestPipelines decoded
length entries `shouldBe` 2
map (unPipelineId . ingestPipelineInfoId) entries `shouldSatisfy` elem "my-pipeline"
map (unPipelineId . ingestPipelineInfoId) entries `shouldSatisfy` elem "other-pipeline"
it "decodes the single-entry response as a singleton list" $ do
let Just decoded = decode sampleSingleResponse :: Maybe IngestPipelines
length (unIngestPipelines decoded) `shouldBe` 1
it "round-trips a multi-entry response through ToJSON/FromJSON" $ do
let Just decoded = decode sampleListResponse :: Maybe IngestPipelines
(decode . encode) decoded `shouldBe` Just decoded
-- Pindocumentation: 'IngestPipelineInfo''s 'ToJSON' uses 'omitNulls',
-- which drops empty arrays. A standalone @decode . encode :: IngestPipelineInfo
-- -> Maybe IngestPipelineInfo@ therefore loses a 'Just (Array [])'
-- processors field (turning it into 'Nothing'). This matches the ILM
-- module's behavior and is intentional — round-trip via 'IngestPipelines'
-- (which preserves ids and bodies) for full fidelity.
it "omitNulls drops an empty processors array on ToJSON round-trip" $ do
let info =
IngestPipelineInfo
{ ingestPipelineInfoId = PipelineId "x",
ingestPipelineInfoVersion = Nothing,
ingestPipelineInfoDescription = Nothing,
ingestPipelineInfoLastModified = Nothing,
ingestPipelineInfoProcessors = Just (Array mempty)
}
encoded = encode info
encoded `shouldBe` "{}"
it "decodes an empty object as an empty list" $ do
let Just decoded = decode "{}" :: Maybe IngestPipelines
unIngestPipelines decoded `shouldBe` []
it "rejects a non-object response (array)" $ do
let decoded = decode "[1,2,3]" :: Maybe IngestPipelines
decoded `shouldBe` Nothing
describe "SimulateIngestPipelineRequest JSON" $ do
it "emits both pipeline and docs when both are present" $ do
let req =
SimulateIngestPipelineRequest
{ simulateIngestPipelineRequestPipeline = Just sampleUppercasePipeline,
simulateIngestPipelineRequestDocs = [object ["_source" .= object ["name" .= ("hello" :: Text.Text)]]]
}
Object o = toJSON req
KM.member "pipeline" o `shouldBe` True
KM.member "docs" o `shouldBe` True
it "omits the pipeline field when it is Nothing" $ do
let req =
SimulateIngestPipelineRequest
{ simulateIngestPipelineRequestPipeline = Nothing,
simulateIngestPipelineRequestDocs = []
}
Object o = toJSON req
KM.member "pipeline" o `shouldBe` False
KM.member "docs" o `shouldBe` True
it "round-trips through ToJSON/FromJSON" $ do
let req =
SimulateIngestPipelineRequest
{ simulateIngestPipelineRequestPipeline = Just sampleUppercasePipeline,
simulateIngestPipelineRequestDocs =
[ object
[ "_index" .= ("ix" :: Text.Text),
"_source" .= object ["name" .= ("hello" :: Text.Text)]
]
]
}
(decode . encode) req `shouldBe` Just req
describe "SimulateResult JSON" $ do
it "decodes a non-verbose response verbatim" $ do
let decoded =
decode
"{\
\ \"docs\": [\
\ { \"doc\": { \"_index\": \"ix\", \"_source\": { \"name\": \"HELLO\" } } }\
\ ]\
\}" ::
Maybe SimulateResult
decoded `shouldSatisfy` isJust
it "rejects a non-object response" $ do
let decoded = decode "[1,2,3]" :: Maybe SimulateResult
decoded `shouldBe` Nothing
describe "putIngestPipeline endpoint shape" $ do
let body = IngestPipeline (object ["description" .= ("synthetic" :: Text.Text)])
req = Common.putIngestPipeline (PipelineId "my-pipeline") body
it "PUTs /_ingest/pipeline/{id}" $ do
getRawEndpoint (bhRequestEndpoint req) `shouldBe` ["_ingest", "pipeline", "my-pipeline"]
it "uses the PUT method" $ do
bhRequestMethod req `shouldBe` "PUT"
it "attaches the encoded pipeline as the request body" $ do
bhRequestBody req `shouldBe` Just (encode body)
it "carries no query string" $ do
getRawEndpointQueries (bhRequestEndpoint req) `shouldBe` []
it "uses a distinct id in the path when given a different PipelineId" $ do
let req' = Common.putIngestPipeline (PipelineId "other") body
getRawEndpoint (bhRequestEndpoint req') `shouldBe` ["_ingest", "pipeline", "other"]
describe "getIngestPipeline endpoint shape" $ do
it "GETs /_ingest/pipeline when given Nothing" $ do
let req = Common.getIngestPipeline Nothing
getRawEndpoint (bhRequestEndpoint req) `shouldBe` ["_ingest", "pipeline"]
getRawEndpointQueries (bhRequestEndpoint req) `shouldBe` []
it "GETs /_ingest/pipeline/{id} when given Just pid" $ do
let req = Common.getIngestPipeline (Just (PipelineId "my-pipeline"))
getRawEndpoint (bhRequestEndpoint req) `shouldBe` ["_ingest", "pipeline", "my-pipeline"]
getRawEndpointQueries (bhRequestEndpoint req) `shouldBe` []
it "uses the GET method" $ do
let req = Common.getIngestPipeline Nothing
bhRequestMethod req `shouldBe` "GET"
it "carries no body" $ do
let req = Common.getIngestPipeline Nothing
bhRequestBody req `shouldBe` Nothing
describe "deleteIngestPipeline endpoint shape" $ do
it "DELETEs /_ingest/pipeline/{id}" $ do
let req = Common.deleteIngestPipeline (PipelineId "my-pipeline")
getRawEndpoint (bhRequestEndpoint req) `shouldBe` ["_ingest", "pipeline", "my-pipeline"]
getRawEndpointQueries (bhRequestEndpoint req) `shouldBe` []
it "uses the DELETE method" $ do
let req = Common.deleteIngestPipeline (PipelineId "my-pipeline")
bhRequestMethod req `shouldBe` "DELETE"
it "forwards a different id into the path" $ do
let req = Common.deleteIngestPipeline (PipelineId "other")
getRawEndpoint (bhRequestEndpoint req) `shouldBe` ["_ingest", "pipeline", "other"]
describe "simulateIngestPipeline endpoint shape" $ do
let docs = [object ["_source" .= object ["name" .= ("hello" :: Text.Text)]]]
it "POSTs /_ingest/pipeline/_simulate when given Nothing" $ do
let req = Common.simulateIngestPipeline Nothing sampleUppercasePipeline docs
getRawEndpoint (bhRequestEndpoint req) `shouldBe` ["_ingest", "pipeline", "_simulate"]
getRawEndpointQueries (bhRequestEndpoint req) `shouldBe` []
it "POSTs /_ingest/pipeline/{id}/_simulate when given Just pid" $ do
let req = Common.simulateIngestPipeline (Just (PipelineId "my-pipeline")) sampleUppercasePipeline docs
getRawEndpoint (bhRequestEndpoint req) `shouldBe` ["_ingest", "pipeline", "my-pipeline", "_simulate"]
getRawEndpointQueries (bhRequestEndpoint req) `shouldBe` []
it "uses the POST method" $ do
let req = Common.simulateIngestPipeline Nothing sampleUppercasePipeline docs
bhRequestMethod req `shouldBe` "POST"
it "attaches the encoded SimulateIngestPipelineRequest as the body" $ do
let req = Common.simulateIngestPipeline Nothing sampleUppercasePipeline docs
expectedBody = encode (SimulateIngestPipelineRequest (Just sampleUppercasePipeline) docs)
bhRequestBody req `shouldBe` Just expectedBody
describe "GrokPatterns JSON" $ do
it "decodes a patterns response (unwrapping the 'patterns' key)" $ do
let bytes =
"{\
\ \"patterns\": {\
\ \"PATH\": \"(?:%{UNIXPATH}|%{WINPATH})\",\
\ \"BACULA_CAPACITY\": \"%{INT}{1,3}(,%{INT}{3})*\"\
\ }\
\}"
Just decoded = decode bytes :: Maybe GrokPatterns
m = unGrokPatterns decoded
Map.size m `shouldBe` 2
Map.lookup "PATH" m `shouldBe` Just "(?:%{UNIXPATH}|%{WINPATH})"
it "decodes an empty patterns map" $ do
let Just decoded = decode "{\"patterns\":{}}" :: Maybe GrokPatterns
unGrokPatterns decoded `shouldBe` Map.empty
it "rejects a response missing the patterns key" $ do
let decoded = decode "{\"foo\":{}}" :: Maybe GrokPatterns
decoded `shouldBe` Nothing
it "round-trips through ToJSON/FromJSON" $ do
let gp = GrokPatterns (Map.fromList [("IP", "%{IPV4}|%{IPV6}"), ("WORD", "\\b\\w+\\b")])
(decode . encode) gp `shouldBe` Just gp
it "encodes back under the 'patterns' key" $ do
let gp = GrokPatterns (Map.singleton "IP" "%{IPV4}|%{IPV6}")
encode gp `shouldBe` "{\"patterns\":{\"IP\":\"%{IPV4}|%{IPV6}\"}}"
describe "getGrokPatterns endpoint shape" $ do
let req = Common.getGrokPatterns
it "GETs /_ingest/processor/grok" $ do
getRawEndpoint (bhRequestEndpoint req) `shouldBe` ["_ingest", "processor", "grok"]
it "uses the GET method" $ do
bhRequestMethod req `shouldBe` "GET"
it "carries no request body" $ do
bhRequestBody req `shouldBe` Nothing
it "carries no query string" $ do
getRawEndpointQueries (bhRequestEndpoint req) `shouldBe` []
-- ------------------------------------------------------------------------
-- Live integration: round-trip through a real ES / OpenSearch cluster.
-- Ingest pipelines are universal across every supported backend, so we
-- run on the full set.
-- ------------------------------------------------------------------------
backendSpecific
[ElasticSearch7, ElasticSearch8, ElasticSearch9, OpenSearch1, OpenSearch2, OpenSearch3]
$ describe "Ingest pipeline (live integration)"
$ do
it "round-trips a pipeline via putIngestPipeline + getIngestPipeline (Just pid)" $
withTestEnv $ do
suffix <- liftIO $ show @Int . floor <$> getPOSIXTime
let pid =
PipelineId $
Text.pack $
"bloodhound_test_ingest_live_put_" <> suffix
_ <- CommonClient.putIngestPipeline pid sampleUppercasePipeline
infos <- CommonClient.getIngestPipeline (Just pid)
liftIO $
case infos of
[info] -> do
ingestPipelineInfoId info `shouldBe` pid
ingestPipelineInfoProcessors info `shouldSatisfy` maybe False hasProcessorsShape
_ ->
expectationFailure $
"expected exactly one pipeline, got " <> show (length infos)
-- cleanup so the next run starts clean
_ <- CommonClient.deleteIngestPipeline pid
pure ()
it "lists the created pipeline via getIngestPipeline Nothing" $
withTestEnv $ do
suffix <- liftIO $ show @Int . floor <$> getPOSIXTime
let pid =
PipelineId $
Text.pack $
"bloodhound_test_ingest_live_list_" <> suffix
_ <- CommonClient.putIngestPipeline pid sampleUppercasePipeline
infos <- CommonClient.getIngestPipeline Nothing
liftIO $
map (unPipelineId . ingestPipelineInfoId) infos
`shouldSatisfy` elem (unPipelineId pid)
_ <- CommonClient.deleteIngestPipeline pid
pure ()
it "deleteIngestPipeline removes the pipeline" $
withTestEnv $ do
suffix <- liftIO $ show @Int . floor <$> getPOSIXTime
let pid =
PipelineId $
Text.pack $
"bloodhound_test_ingest_live_del_" <> suffix
_ <- CommonClient.putIngestPipeline pid sampleUppercasePipeline
_ <- CommonClient.deleteIngestPipeline pid
infos <- CommonClient.getIngestPipeline Nothing
liftIO $
map (unPipelineId . ingestPipelineInfoId) infos
`shouldSatisfy` notElem (unPipelineId pid)
it "simulateIngestPipeline upper-cases the name field" $
withTestEnv $ do
let docs =
[ object
[ "_index" .= ("ix" :: Text.Text),
"_source" .= object ["name" .= ("hello world" :: Text.Text)]
]
]
result <-
CommonClient.simulateIngestPipeline Nothing sampleUppercasePipeline docs
let body = unSimulateResult result
-- Walk the response: docs[0].doc._source.name == "HELLO WORLD"
key :: Text.Text -> Value -> Maybe Value
key k (Object o) = KM.lookup (K.fromText k) o
key _ _ = Nothing
walkName :: Value -> Maybe Value
walkName top = do
docsArr <- key "docs" top
firstDoc <- firstArrayElement docsArr
doc <- key "doc" firstDoc
src <- key "_source" doc
key "name" src
firstArrayElement :: Value -> Maybe Value
firstArrayElement (Array a) = case V.toList a of (x : _) -> Just x; _ -> Nothing
firstArrayElement _ = Nothing
liftIO $ walkName body `shouldBe` Just (String "HELLO WORLD")
it "simulateIngestPipeline (Just pid) runs the STORED pipeline; the inline body is ignored" $
withTestEnv $ do
suffix <- liftIO $ show @Int . floor <$> getPOSIXTime
let pid =
PipelineId $
Text.pack $
"bloodhound_test_ingest_live_simid_" <> suffix
-- Store a pipeline that uppercases @name@ — we will assert this
-- transformation runs.
_ <- CommonClient.putIngestPipeline pid sampleUppercasePipeline
-- Simulate against the stored id but supply a DIFFERENT inline
-- body (a no-op @set@ of an unrelated field). ES must run the
-- stored pipeline, not the inline one — so @name@ should be
-- upper-cased.
let inlineBody =
IngestPipeline $
object
[ "description" .= ("inline no-op, should be ignored" :: Text.Text),
"processors"
.= [ object
[ "set"
.= object
[ "field" .= ("inline_marker" :: Text.Text),
"value" .= ("inline" :: Text.Text)
]
]
]
]
docs =
[ object
[ "_index" .= ("ix" :: Text.Text),
"_source" .= object ["name" .= ("store me" :: Text.Text)]
]
]
result <- CommonClient.simulateIngestPipeline (Just pid) inlineBody docs
let body = unSimulateResult result
key :: Text.Text -> Value -> Maybe Value
key k (Object o) = KM.lookup (K.fromText k) o
key _ _ = Nothing
walkName :: Value -> Maybe Value
walkName top = do
docsArr <- key "docs" top
firstDoc <- firstArrayElement docsArr
doc <- key "doc" firstDoc
src <- key "_source" doc
key "name" src
firstArrayElement :: Value -> Maybe Value
firstArrayElement (Array a) = case V.toList a of (x : _) -> Just x; _ -> Nothing
firstArrayElement _ = Nothing
liftIO $ walkName body `shouldBe` Just (String "STORE ME")
_ <- CommonClient.deleteIngestPipeline pid
pure ()
-- | Sample PUT body shaped like a real ingest pipeline: a @description@
-- and a single @set@ processor. Used by the JSON round-trip tests.
samplePutBodyBytes :: LBS.ByteString
samplePutBodyBytes =
"{\
\ \"description\": \"example pipeline\",\
\ \"processors\": [\
\ { \"set\": { \"field\": \"foo\", \"value\": \"bar\" } }\
\ ],\
\ \"version\": 1\
\}"
-- | True iff the given 'Value' is an array-shaped processors field. Used
-- to assert the opaque processors body without modelling the ~30
-- processor types.
hasProcessorsShape :: Value -> Bool
hasProcessorsShape (Array _) = True
hasProcessorsShape _ = False