bloodhound-1.0.0.0: tests/Test/MigrationReindexSpec.hs
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE QuasiQuotes #-}
module Test.MigrationReindexSpec (spec) where
import Data.Aeson
import Data.ByteString.Lazy.Char8 qualified as LBS
import Database.Bloodhound.Client.Cluster (BackendType (..))
import Database.Bloodhound.ElasticSearch8.Client qualified as ClientES8
import Database.Bloodhound.ElasticSearch8.Requests qualified as RequestsES8
import Database.Bloodhound.ElasticSearch8.Types qualified as Types
import Database.Bloodhound.ElasticSearch9.Requests qualified as RequestsES9
import System.Environment (lookupEnv)
import TestsUtils.Common
import TestsUtils.Import
import Prelude
------------------------------------------------------------------------------
-- Sample payloads (built from the ES9 OpenAPI spec shapes).
------------------------------------------------------------------------------
-- | Canonical @POST /_migration/reindex@ request body: reindex the
-- @my-data-stream@ data stream in @upgrade@ mode.
sampleMigrateReindexRequestBytes :: LBS.ByteString
sampleMigrateReindexRequestBytes =
"{\
\ \"source\": {\"index\": \"my-data-stream\"},\
\ \"mode\": \"upgrade\"\
\}"
-- | A @GET /_migration/reindex/{index}/_status@ response mid-flight: one
-- index in progress (half its docs reindexed), one errored, the rest
-- pending. @start_time@ is an ISO-8601 string (one of the two wire forms
-- of @_types.DateTime@).
sampleMigrateReindexStatusBytes :: LBS.ByteString
sampleMigrateReindexStatusBytes =
"{\
\ \"start_time\": \"2024-01-15T10:30:00.000Z\",\
\ \"start_time_millis\": 1705314600000,\
\ \"complete\": false,\
\ \"total_indices_in_data_stream\": 5,\
\ \"total_indices_requiring_upgrade\": 4,\
\ \"successes\": 2,\
\ \"in_progress\": [\
\ {\
\ \"index\": \"logs-app-000003\",\
\ \"total_doc_count\": 1000,\
\ \"reindexed_doc_count\": 500\
\ }\
\ ],\
\ \"pending\": 1,\
\ \"errors\": [\
\ {\
\ \"index\": \"logs-app-000001\",\
\ \"message\": \"cluster:blocked\"\
\ }\
\ ]\
\}"
-- | A completed status with @start_time@ as an epoch-millis number (the
-- other @_types.DateTime@ wire form) and an @exception@ set.
sampleMigrateReindexStatusCompleteBytes :: LBS.ByteString
sampleMigrateReindexStatusCompleteBytes =
"{\
\ \"start_time_millis\": 1705314600000,\
\ \"complete\": true,\
\ \"total_indices_in_data_stream\": 3,\
\ \"total_indices_requiring_upgrade\": 2,\
\ \"successes\": 1,\
\ \"in_progress\": [],\
\ \"pending\": 0,\
\ \"errors\": [],\
\ \"exception\": \"partial failure\"\
\}"
------------------------------------------------------------------------------
-- Spec
------------------------------------------------------------------------------
spec :: Spec
spec = describe "Migration reindex API" $ do
--------------------------------------------------------------------------
-- MigrateReindexMode
--------------------------------------------------------------------------
describe "MigrateReindexMode JSON" $ do
it "encodes the upgrade constructor" $
encode Types.MigrateReindexModeUpgrade `shouldBe` "\"upgrade\""
it "decodes the upgrade wire string" $
(decode "\"upgrade\"" :: Maybe Types.MigrateReindexMode)
`shouldBe` Just Types.MigrateReindexModeUpgrade
it "rejects an unknown mode" $
(decode "\"downgrade\"" :: Maybe Types.MigrateReindexMode)
`shouldBe` Nothing
it "rejects a non-string value" $
(decode "42" :: Maybe Types.MigrateReindexMode)
`shouldBe` Nothing
--------------------------------------------------------------------------
-- MigrateReindexSource / MigrateReindexRequest
--------------------------------------------------------------------------
describe "MigrateReindexRequest JSON" $ do
it "decodes the canonical ES request body" $
case (decode sampleMigrateReindexRequestBytes :: Maybe Types.MigrateReindexRequest) of
Nothing -> expectationFailure "expected MigrateReindexRequest decode"
Just decoded -> do
Types.migrateReindexRequestMode decoded
`shouldBe` Types.MigrateReindexModeUpgrade
Types.migrateReindexSourceIndex
(Types.migrateReindexRequestSource decoded)
`shouldBe` [qqIndexName|my-data-stream|]
it "round-trips through ToJSON/FromJSON" $ do
let req =
Types.mkMigrateReindexRequest [qqIndexName|logs-stream|]
(decode . encode) req `shouldBe` Just req
it "mkMigrateReindexRequest sets the upgrade mode" $ do
let req = Types.mkMigrateReindexRequest [qqIndexName|ds|]
Types.migrateReindexRequestMode req
`shouldBe` Types.MigrateReindexModeUpgrade
it "rejects a body missing the mode field" $
( decode
"{\"source\":{\"index\":\"ds\"}}" ::
Maybe Types.MigrateReindexRequest
)
`shouldBe` Nothing
it "rejects a body missing the source field" $
( decode "{\"mode\":\"upgrade\"}" ::
Maybe Types.MigrateReindexRequest
)
`shouldBe` Nothing
it "rejects a non-object body" $
(decode "[]" :: Maybe Types.MigrateReindexRequest)
`shouldBe` Nothing
--------------------------------------------------------------------------
-- MigrateReindexInProgress
--------------------------------------------------------------------------
describe "MigrateReindexInProgress JSON" $ do
it "decodes a canonical entry" $
case ( decode
"{\"index\":\"logs-app-000003\",\"total_doc_count\":1000,\"reindexed_doc_count\":500}" ::
Maybe Types.MigrateReindexInProgress
) of
Nothing -> expectationFailure "expected MigrateReindexInProgress decode"
Just decoded -> do
Types.migrateReindexInProgressIndex decoded
`shouldBe` [qqIndexName|logs-app-000003|]
Types.migrateReindexInProgressTotalDocCount decoded
`shouldBe` 1000
Types.migrateReindexInProgressReindexedDocCount decoded
`shouldBe` 500
it "round-trips through ToJSON/FromJSON" $ do
let entry =
Types.MigrateReindexInProgress
{ Types.migrateReindexInProgressIndex =
[qqIndexName|idx|],
Types.migrateReindexInProgressTotalDocCount = 7,
Types.migrateReindexInProgressReindexedDocCount = 3
}
(decode . encode) entry `shouldBe` Just entry
it "rejects a body missing total_doc_count" $
( decode
"{\"index\":\"idx\",\"reindexed_doc_count\":3}" ::
Maybe Types.MigrateReindexInProgress
)
`shouldBe` Nothing
--------------------------------------------------------------------------
-- MigrateReindexError
--------------------------------------------------------------------------
describe "MigrateReindexError JSON" $ do
it "decodes a canonical entry" $
case ( decode
"{\"index\":\"logs-app-000001\",\"message\":\"cluster:blocked\"}" ::
Maybe Types.MigrateReindexError
) of
Nothing -> expectationFailure "expected MigrateReindexError decode"
Just decoded -> do
Types.migrateReindexErrorIndex decoded
`shouldBe` [qqIndexName|logs-app-000001|]
Types.migrateReindexErrorMessage decoded
`shouldBe` "cluster:blocked"
it "round-trips through ToJSON/FromJSON" $ do
let err =
Types.MigrateReindexError
{ Types.migrateReindexErrorIndex = [qqIndexName|idx|],
Types.migrateReindexErrorMessage = "boom"
}
(decode . encode) err `shouldBe` Just err
--------------------------------------------------------------------------
-- MigrateReindexStatus
--------------------------------------------------------------------------
describe "MigrateReindexStatus JSON" $ do
it "decodes the canonical mid-flight response" $
case (decode sampleMigrateReindexStatusBytes :: Maybe Types.MigrateReindexStatus) of
Nothing -> expectationFailure "expected MigrateReindexStatus decode"
Just decoded -> do
Types.migrateReindexStatusStartTimeMillis decoded
`shouldBe` 1705314600000
Types.migrateReindexStatusComplete decoded `shouldBe` False
Types.migrateReindexStatusTotalIndicesInDataStream decoded
`shouldBe` 5
Types.migrateReindexStatusSuccesses decoded `shouldBe` 2
Types.migrateReindexStatusPending decoded `shouldBe` 1
length (Types.migrateReindexStatusInProgress decoded)
`shouldBe` 1
length (Types.migrateReindexStatusErrors decoded)
`shouldBe` 1
-- start_time accepted as an ISO-8601 string
Types.migrateReindexStatusStartTime decoded
`shouldSatisfy` isJust
Types.migrateReindexStatusException decoded
`shouldBe` Nothing
it "decodes a completed response with epoch-number start_time omitted and exception set" $
case (decode sampleMigrateReindexStatusCompleteBytes :: Maybe Types.MigrateReindexStatus) of
Nothing -> expectationFailure "expected MigrateReindexStatus decode"
Just decoded -> do
Types.migrateReindexStatusComplete decoded `shouldBe` True
Types.migrateReindexStatusStartTime decoded
`shouldBe` Nothing
Types.migrateReindexStatusInProgress decoded `shouldBe` []
Types.migrateReindexStatusErrors decoded `shouldBe` []
Types.migrateReindexStatusException decoded
`shouldBe` Just "partial failure"
it "tolerates missing in_progress and errors (default to empty)" $
case ( decode
"{\"start_time_millis\":1,\"complete\":true,\
\ \"total_indices_in_data_stream\":0,\
\ \"total_indices_requiring_upgrade\":0,\
\ \"successes\":0,\"pending\":0}" ::
Maybe Types.MigrateReindexStatus
) of
Nothing -> expectationFailure "expected MigrateReindexStatus decode"
Just decoded -> do
Types.migrateReindexStatusInProgress decoded `shouldBe` []
Types.migrateReindexStatusErrors decoded `shouldBe` []
it "round-trips through ToJSON/FromJSON (omitNulls drops empty arrays)" $ do
-- 'omitNulls' drops empty arrays and Nothing fields, so after a
-- round-trip an empty @in_progress@/@errors@ and absent @start_time@
-- resurface as @[]@ / @Nothing@ (the decoder defaults them).
let s =
Types.MigrateReindexStatus
{ Types.migrateReindexStatusStartTime = Nothing,
Types.migrateReindexStatusStartTimeMillis = 1,
Types.migrateReindexStatusComplete = True,
Types.migrateReindexStatusTotalIndicesInDataStream = 0,
Types.migrateReindexStatusTotalIndicesRequiringUpgrade = 0,
Types.migrateReindexStatusSuccesses = 0,
Types.migrateReindexStatusInProgress = [],
Types.migrateReindexStatusPending = 0,
Types.migrateReindexStatusErrors = [],
Types.migrateReindexStatusException = Nothing
}
(decode . encode) s `shouldBe` Just s
it "round-trips a populated status unchanged" $
case (decode sampleMigrateReindexStatusBytes :: Maybe Types.MigrateReindexStatus) of
Nothing -> expectationFailure "expected MigrateReindexStatus decode"
Just decoded -> (decode . encode) decoded `shouldBe` Just decoded
it "rejects a body missing start_time_millis" $
( decode
"{\"complete\":true,\"total_indices_in_data_stream\":0,\
\ \"total_indices_requiring_upgrade\":0,\"successes\":0,\
\ \"pending\":0}" ::
Maybe Types.MigrateReindexStatus
)
`shouldBe` Nothing
it "rejects a non-object body" $
(decode "42" :: Maybe Types.MigrateReindexStatus)
`shouldBe` Nothing
--------------------------------------------------------------------------
-- CreateIndexFromBody
--------------------------------------------------------------------------
describe "CreateIndexFromBody JSON" $ do
it "decodes a fully-populated body" $
case ( decode
"{\"mappings_override\":{\"enabled\":true},\
\ \"settings_override\":{\"index\":{\"number_of_shards\":1}},\
\ \"remove_index_blocks\":false}" ::
Maybe Types.CreateIndexFromBody
) of
Nothing -> expectationFailure "expected CreateIndexFromBody decode"
Just decoded -> do
Types.createIndexFromBodyRemoveIndexBlocks decoded
`shouldBe` Just False
Types.createIndexFromBodyMappingsOverride decoded
`shouldSatisfy` isJust
it "decodes an empty object as the default (all Nothing)" $
(decode "{}" :: Maybe Types.CreateIndexFromBody)
`shouldBe` Just Types.defaultCreateIndexFromBody
it "omits all Nothing fields from ToJSON output" $
encode Types.defaultCreateIndexFromBody `shouldBe` "{}"
it "round-trips a fully-populated body" $ do
let body =
Types.CreateIndexFromBody
{ Types.createIndexFromBodyMappingsOverride =
Just (object ["enabled" .= True]),
Types.createIndexFromBodySettingsOverride = Nothing,
Types.createIndexFromBodyRemoveIndexBlocks = Just True
}
(decode . encode) body `shouldBe` Just body
--------------------------------------------------------------------------
-- CreateIndexFromResponse
--------------------------------------------------------------------------
describe "CreateIndexFromResponse JSON" $ do
it "decodes a canonical response" $ do
let Just decoded =
decode
"{\"acknowledged\":true,\"index\":\"my-new-index\",\
\ \"shards_acknowledged\":true}" ::
Maybe Types.CreateIndexFromResponse
Types.createIndexFromResponseAcknowledged decoded `shouldBe` True
Types.createIndexFromResponseIndex decoded
`shouldBe` [qqIndexName|my-new-index|]
Types.createIndexFromResponseShardsAcknowledged decoded
`shouldBe` True
it "round-trips through ToJSON/FromJSON" $ do
let resp =
Types.CreateIndexFromResponse
{ Types.createIndexFromResponseAcknowledged = True,
Types.createIndexFromResponseIndex = [qqIndexName|dest|],
Types.createIndexFromResponseShardsAcknowledged = False
}
(decode . encode) resp `shouldBe` Just resp
it "rejects a body missing shards_acknowledged" $
( decode
"{\"acknowledged\":true,\"index\":\"dest\"}" ::
Maybe Types.CreateIndexFromResponse
)
`shouldBe` Nothing
--------------------------------------------------------------------------
-- Endpoint shape tests (ES8 builders — the canonical implementation)
--------------------------------------------------------------------------
describe "migrateReindex endpoint shape" $ do
let req = Types.mkMigrateReindexRequest [qqIndexName|my-data-stream|]
bhReq = RequestsES8.migrateReindex req
it "POSTs /_migration/reindex" $
getRawEndpoint (bhRequestEndpoint bhReq)
`shouldBe` ["_migration", "reindex"]
it "uses the POST method" $
bhRequestMethod bhReq `shouldBe` "POST"
it "carries the encoded request body" $ do
bhRequestBody bhReq `shouldSatisfy` isJust
-- the body round-trips back into the same request
let Just body = bhRequestBody bhReq
(decode body :: Maybe Types.MigrateReindexRequest) `shouldBe` Just req
it "carries no query string" $
getRawEndpointQueries (bhRequestEndpoint bhReq) `shouldBe` []
describe "cancelMigrateReindex endpoint shape" $ do
let bhReq =
RequestsES8.cancelMigrateReindex
([qqIndexName|my-data-stream|] :| [])
it "POSTs /_migration/reindex/{index}/_cancel" $
getRawEndpoint (bhRequestEndpoint bhReq)
`shouldBe` ["_migration", "reindex", "my-data-stream", "_cancel"]
it "uses the POST method" $
bhRequestMethod bhReq `shouldBe` "POST"
it "sends an empty body" $
bhRequestBody bhReq `shouldBe` Just ""
it "comma-joins multiple indices into one path segment" $ do
let bhReq' =
RequestsES8.cancelMigrateReindex
( [qqIndexName|ds-one|]
:| [[qqIndexName|ds-two|]]
)
getRawEndpoint (bhRequestEndpoint bhReq')
`shouldBe` ["_migration", "reindex", "ds-one,ds-two", "_cancel"]
describe "getMigrateReindexStatus endpoint shape" $ do
let bhReq =
RequestsES8.getMigrateReindexStatus
([qqIndexName|my-data-stream|] :| [])
it "GETs /_migration/reindex/{index}/_status" $
getRawEndpoint (bhRequestEndpoint bhReq)
`shouldBe` ["_migration", "reindex", "my-data-stream", "_status"]
it "uses the GET method" $
bhRequestMethod bhReq `shouldBe` "GET"
it "carries no request body" $
bhRequestBody bhReq `shouldBe` Nothing
describe "createIndexFrom endpoint shape" $ do
let bhReq = RequestsES8.createIndexFrom [qqIndexName|src|] [qqIndexName|dst|]
it "POSTs /_create_from/{source}/{dest}" $
getRawEndpoint (bhRequestEndpoint bhReq)
`shouldBe` ["_create_from", "src", "dst"]
it "uses the POST method" $
bhRequestMethod bhReq `shouldBe` "POST"
it "sends an empty body when no overrides are supplied" $
bhRequestBody bhReq `shouldBe` Just ""
describe "createIndexFromWith endpoint shape" $ do
let body =
Types.CreateIndexFromBody
{ Types.createIndexFromBodyMappingsOverride =
Just (object ["enabled" .= True]),
Types.createIndexFromBodySettingsOverride = Nothing,
Types.createIndexFromBodyRemoveIndexBlocks = Just False
}
bhReq =
RequestsES8.createIndexFromWith
[qqIndexName|src|]
[qqIndexName|dst|]
(Just body)
it "POSTs /_create_from/{source}/{dest}" $
getRawEndpoint (bhRequestEndpoint bhReq)
`shouldBe` ["_create_from", "src", "dst"]
it "carries the encoded override body" $ do
bhRequestBody bhReq `shouldSatisfy` isJust
let Just b = bhRequestBody bhReq
(decode b :: Maybe Types.CreateIndexFromBody) `shouldBe` Just body
describe "ES9 aliases match the ES8 builders" $ do
let indices = [qqIndexName|my-data-stream|] :| []
it "migrateReindex produces the same endpoint as ES8" $ do
let req = Types.mkMigrateReindexRequest [qqIndexName|ds|]
getRawEndpoint
(bhRequestEndpoint (RequestsES9.migrateReindex req))
`shouldBe` getRawEndpoint
(bhRequestEndpoint (RequestsES8.migrateReindex req))
it "getMigrateReindexStatus produces the same endpoint as ES8" $
getRawEndpoint
(bhRequestEndpoint (RequestsES9.getMigrateReindexStatus indices))
`shouldBe` getRawEndpoint
(bhRequestEndpoint (RequestsES8.getMigrateReindexStatus indices))
it "cancelMigrateReindex produces the same endpoint as ES8" $
getRawEndpoint
(bhRequestEndpoint (RequestsES9.cancelMigrateReindex indices))
`shouldBe` getRawEndpoint
(bhRequestEndpoint (RequestsES8.cancelMigrateReindex indices))
it "createIndexFrom produces the same endpoint as ES8" $
getRawEndpoint
( bhRequestEndpoint
(RequestsES9.createIndexFrom [qqIndexName|s|] [qqIndexName|d|])
)
`shouldBe` getRawEndpoint
( bhRequestEndpoint
(RequestsES8.createIndexFrom [qqIndexName|s|] [qqIndexName|d|])
)
--------------------------------------------------------------------------
-- Live integration (gated on ES8/ES9; mutating ops further gated
-- behind ES_TEST_RUN_MUTATING so cluster state is not perturbed by
-- default).
--------------------------------------------------------------------------
backendSpecific
[ElasticSearch8, ElasticSearch9]
$ describe "Migration reindex API (live integration)"
$ do
it "getMigrateReindexStatus returns a parseable response when ES_TEST_RUN_MUTATING=true" $
withTestEnv $ do
-- Querying a non-existent data stream surfaces as an EsError
-- (404) rather than a decodable status, so this is gated on
-- the mutating flag alongside the cluster-perturbing ops; a
-- meaningful live assertion requires a real data stream.
enabled <- liftIO $ lookupEnv "ES_TEST_RUN_MUTATING"
case enabled of
Just "true" -> do
result <-
ClientES8.getMigrateReindexStatus
([qqIndexName|bloodhound-missing-stream|] :| [])
liftIO $
Types.migrateReindexStatusComplete result
`shouldSatisfy` (const True :: Bool -> Bool)
_ ->
liftIO $
pendingWith
"skipped: set ES_TEST_RUN_MUTATING=true to run this test"
it "migrateReindex returns Acknowledged when ES_TEST_RUN_MUTATING=true" $
withTestEnv $ do
enabled <- liftIO $ lookupEnv "ES_TEST_RUN_MUTATING"
case enabled of
Just "true" -> do
ack <-
ClientES8.migrateReindex
( Types.mkMigrateReindexRequest
[qqIndexName|bloodhound-missing-stream|]
)
liftIO $ isAcknowledged ack `shouldBe` True
_ ->
liftIO $
pendingWith
"skipped: set ES_TEST_RUN_MUTATING=true to run this mutating test"
it "createIndexFrom returns acknowledged when ES_TEST_RUN_MUTATING=true" $
withTestEnv $ do
enabled <- liftIO $ lookupEnv "ES_TEST_RUN_MUTATING"
case enabled of
Just "true" -> do
resp <-
ClientES8.createIndexFrom
[qqIndexName|bloodhound-tests-twitter-1|]
[qqIndexName|bloodhound-createfrom-dest|]
liftIO $
Types.createIndexFromResponseAcknowledged resp
`shouldBe` True
_ ->
liftIO $
pendingWith
"skipped: set ES_TEST_RUN_MUTATING=true to run this mutating test"