bloodhound-1.0.0.0: tests/Test/TasksSpec.hs
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE QuasiQuotes #-}
{-# LANGUAGE TypeApplications #-}
module Test.TasksSpec (spec) where
import Control.Concurrent (threadDelay)
import Data.List qualified as L
import Data.Text qualified as T
import Numeric.Natural
import TestsUtils.Common
import TestsUtils.Import
spec :: Spec
spec = do
-- ------------------------------------------------------------------ --
-- FromJSON decoding (pure, no ES required) --
-- ------------------------------------------------------------------ --
describe "TaskListResponse FromJSON" $ do
let singleTask =
"{\"nodes\":{\"n1\":{"
<> "\"name\":\"node-1\","
<> "\"transport_address\":\"127.0.0.1:9300\","
<> "\"host\":\"127.0.0.1\","
<> "\"ip\":\"127.0.0.1\","
<> "\"tasks\":{\"n1:42\":{"
<> "\"node\":\"n1\","
<> "\"id\":42,"
<> "\"type\":\"transport\","
<> "\"action\":\"indices:data/write/reindex\","
<> "\"description\":\"reindex [bloodhound-tests-twitter-1]\","
<> "\"start_time_in_millis\":1700000000000,"
<> "\"running_time_in_nanos\":5000000000,"
<> "\"cancellable\":true,"
<> "\"parent_task_id\":\"n1:1\""
<> "}}}}}"
it "parses a nodes-grouped response and stamps the node id" $ do
let parsed = decode singleTask :: Maybe TaskListResponse
parsed `shouldSatisfy` isJust
case parsed of
Just resp -> do
length (taskListResponseNodes resp) `shouldBe` 1
case taskListResponseNodes resp of
(node : _) -> taskListNodeId node `shouldBe` "n1"
_ -> expectationFailure "expected at least one node"
Nothing -> expectationFailure "expected TaskListResponse parse"
it "flattens to a single TaskInfo with the expected fields" $
case decode singleTask :: Maybe TaskListResponse of
Just resp ->
case taskListFlat resp of
[t] -> do
taskInfoNode t `shouldBe` "n1"
taskInfoId t `shouldBe` 42
taskInfoType t `shouldBe` "transport"
taskInfoAction t `shouldBe` "indices:data/write/reindex"
taskInfoDescription t `shouldBe` Just "reindex [bloodhound-tests-twitter-1]"
taskInfoStartTimeInMillis t `shouldBe` 1700000000000
taskInfoRunningTimeInNanos t `shouldBe` 5000000000
taskInfoCancellable t `shouldBe` True
taskInfoParentTaskId t `shouldBe` Just "n1:1"
_ -> expectationFailure "expected exactly one task"
Nothing -> expectationFailure "expected TaskListResponse parse"
it "parses a node with no tasks into an empty flat list" $ do
let payload =
"{\"nodes\":{\"n1\":{\"name\":\"node-1\",\"tasks\":{}}}}"
case decode payload :: Maybe TaskListResponse of
Just resp -> taskListFlat resp `shouldBe` []
Nothing -> expectationFailure "expected TaskListResponse parse"
it "tolerates a missing tasks key on a node" $ do
let payload =
"{\"nodes\":{\"n1\":{\"name\":\"node-1\"}}}"
case decode payload :: Maybe TaskListResponse of
Just resp -> taskListFlat resp `shouldBe` []
Nothing -> expectationFailure "expected TaskListResponse parse"
it "parses multiple nodes and tasks" $ do
let payload =
"{\"nodes\":{"
<> "\"n1\":{\"tasks\":{\"n1:1\":{\"node\":\"n1\",\"id\":1,\"type\":\"transport\",\"action\":\"a1\",\"start_time_in_millis\":1,\"running_time_in_nanos\":1,\"cancellable\":false}}},"
<> "\"n2\":{\"tasks\":{\"n2:2\":{\"node\":\"n2\",\"id\":2,\"type\":\"transport\",\"action\":\"a2\",\"start_time_in_millis\":2,\"running_time_in_nanos\":2,\"cancellable\":true}}}"
<> "}}"
case decode payload :: Maybe TaskListResponse of
Just resp -> do
let tasks = taskListFlat resp
L.sort (taskInfoAction <$> tasks) `shouldBe` ["a1", "a2"]
-- payload omits description / parent_task_id, so the .:?
-- decode path must yield Nothing for every task.
let descriptions = taskInfoDescription <$> tasks
parents = taskInfoParentTaskId <$> tasks
all isNothing descriptions `shouldBe` True
all isNothing parents `shouldBe` True
Nothing -> expectationFailure "expected TaskListResponse parse"
-- ------------------------------------------------------------------ --
-- Task FromJSON (pure, no ES required) --
-- ------------------------------------------------------------------ --
describe "Task FromJSON" $ do
let fullTask =
"{\"node\":\"n1\","
<> "\"id\":42,"
<> "\"type\":\"transport\","
<> "\"action\":\"indices:data/write/reindex\","
<> "\"status\":{\"created\":0,\"batches\":1},"
<> "\"description\":\"reindex [bloodhound-tests-twitter-1]\","
<> "\"start_time_in_millis\":1700000000000,"
<> "\"running_time_in_nanos\":5000000000,"
<> "\"cancellable\":true,"
<> "\"parent_task_id\":\"n1:1\","
<> "\"headers\":{\"X-Opaque-Id\":\"abc\"}"
<> "}"
it "decodes parent_task_id and headers when present" $
case decode fullTask :: Maybe (Task Value) of
Just t -> do
taskParentTaskId t `shouldBe` Just "n1:1"
taskHeaders t `shouldSatisfy` isJust
Nothing -> expectationFailure "expected Task parse"
it "yields Nothing for parent_task_id and headers when absent" $ do
let payload =
"{\"node\":\"n1\","
<> "\"id\":42,"
<> "\"type\":\"transport\","
<> "\"action\":\"indices:data/write/reindex\","
<> "\"status\":{},"
<> "\"description\":\"reindex\","
<> "\"start_time_in_millis\":1700000000000,"
<> "\"running_time_in_nanos\":5000000000,"
<> "\"cancellable\":true"
<> "}"
case decode payload :: Maybe (Task Value) of
Just t -> do
taskParentTaskId t `shouldBe` Nothing
taskHeaders t `shouldBe` Nothing
Nothing -> expectationFailure "expected Task parse"
it "preserves the \"-1\" sentinel for parent_task_id verbatim" $ do
let payload =
"{\"node\":\"n1\","
<> "\"id\":42,"
<> "\"type\":\"transport\","
<> "\"action\":\"indices:data/write/reindex\","
<> "\"status\":{},"
<> "\"description\":\"reindex\","
<> "\"start_time_in_millis\":1700000000000,"
<> "\"running_time_in_nanos\":5000000000,"
<> "\"cancellable\":true,"
<> "\"parent_task_id\":\"-1\""
<> "}"
case decode payload :: Maybe (Task Value) of
Just t -> taskParentTaskId t `shouldBe` Just "-1"
Nothing -> expectationFailure "expected Task parse"
-- ------------------------------------------------------------------ --
-- TaskResponse FromJSON (pure, no ES required) --
-- --
-- Elasticsearch emits the completed-task result under the --
-- historically-misspelled @reponse@ key, while OpenSearch emits it --
-- under the correctly-spelled @response@ key. The decoder must --
-- accept both (preferring @response@) or silently drop the result on --
-- OpenSearch. --
-- ------------------------------------------------------------------ --
describe "TaskResponse FromJSON" $ do
let minTask =
"\"task\":{\"node\":\"n1\","
<> "\"id\":42,"
<> "\"type\":\"transport\","
<> "\"action\":\"indices:data/write/reindex\","
<> "\"status\":{},"
<> "\"description\":\"reindex\","
<> "\"start_time_in_millis\":1,"
<> "\"running_time_in_nanos\":1,"
<> "\"cancellable\":true"
<> "}"
it "decodes the ES-style \"reponse\" (historical typo) result key" $ do
let payload = "{\"completed\":true," <> minTask <> ",\"reponse\":\"es-result\"}"
case decode payload :: Maybe (TaskResponse Value) of
Just r -> taskResponseResponse r `shouldBe` Just (String "es-result")
Nothing -> expectationFailure "expected TaskResponse parse"
it "decodes the OpenSearch \"response\" result key (regression: was silently dropped)" $ do
let payload = "{\"completed\":true," <> minTask <> ",\"response\":\"os-result\"}"
case decode payload :: Maybe (TaskResponse Value) of
Just r -> taskResponseResponse r `shouldBe` Just (String "os-result")
Nothing -> expectationFailure "expected TaskResponse parse"
it "prefers \"response\" over \"reponse\" when both keys are present" $ do
let payload = "{\"completed\":true," <> minTask <> ",\"response\":\"os\",\"reponse\":\"es\"}"
case decode payload :: Maybe (TaskResponse Value) of
Just r -> taskResponseResponse r `shouldBe` Just (String "os")
Nothing -> expectationFailure "expected TaskResponse parse"
it "yields Nothing for the result when neither key is present" $ do
let payload = "{\"completed\":false," <> minTask <> "}"
case decode payload :: Maybe (TaskResponse Value) of
Just r -> taskResponseResponse r `shouldBe` Nothing
Nothing -> expectationFailure "expected TaskResponse parse"
-- ------------------------------------------------------------------ --
-- taskListOptionsParams: pure URI rendering (no ES required) --
-- ------------------------------------------------------------------ --
describe "taskListOptionsParams URI rendering" $ do
let normalize :: [(T.Text, Maybe T.Text)] -> [(T.Text, Maybe T.Text)]
normalize = L.sort
it "defaultTaskListOptions emits no params" $
taskListOptionsParams defaultTaskListOptions
`shouldBe` []
it "renders nodes as a comma-joined list" $ do
let opts =
defaultTaskListOptions
{ taskListOptionsNodes = Just ["node-a", "node-b"]
}
taskListOptionsParams opts
`shouldBe` [("nodes", Just "node-a,node-b")]
it "renders actions as a comma-joined list" $ do
let opts =
defaultTaskListOptions
{ taskListOptionsActions =
Just ["indices:data/write/reindex", "indices:data/read/search"]
}
taskListOptionsParams opts
`shouldBe` [("actions", Just "indices:data/write/reindex,indices:data/read/search")]
it "renders booleans as lowercase true/false" $ do
let opts =
defaultTaskListOptions
{ taskListOptionsDetailed = Just True,
taskListOptionsWaitForCompletion = Just False
}
normalize (taskListOptionsParams opts)
`shouldBe` [ ("detailed", Just "true"),
("wait_for_completion", Just "false")
]
it "renders parent_task_id verbatim" $ do
let opts =
defaultTaskListOptions
{ taskListOptionsParentTaskId = Just "n1:1"
}
taskListOptionsParams opts
`shouldBe` [("parent_task_id", Just "n1:1")]
it "renders timeout as <n><suffix>" $ do
let opts =
defaultTaskListOptions
{ taskListOptionsTimeout = Just (TimeUnitSeconds, 30)
}
taskListOptionsParams opts
`shouldBe` [("timeout", Just "30s")]
it "renders group_by via renderTaskListGroupBy" $ do
let opts =
defaultTaskListOptions
{ taskListOptionsGroupBy = Just TaskListGroupByParents
}
taskListOptionsParams opts
`shouldBe` [("group_by", Just "parents")]
it "emits every param together when all are set" $ do
let opts =
defaultTaskListOptions
{ taskListOptionsNodes = Just ["n1"],
taskListOptionsActions = Just ["indices:data/write/reindex"],
taskListOptionsDetailed = Just True,
taskListOptionsParentTaskId = Just "n1:0",
taskListOptionsWaitForCompletion = Just True,
taskListOptionsTimeout = Just (TimeUnitMilliseconds, 500),
taskListOptionsGroupBy = Just TaskListGroupByNodes
}
normalize (taskListOptionsParams opts)
`shouldBe` [ ("actions", Just "indices:data/write/reindex"),
("detailed", Just "true"),
("group_by", Just "nodes"),
("nodes", Just "n1"),
("parent_task_id", Just "n1:0"),
("timeout", Just "500ms"),
("wait_for_completion", Just "true")
]
-- ------------------------------------------------------------------ --
-- taskGetOptionsParams: pure URI rendering (no ES required) --
-- ------------------------------------------------------------------ --
describe "taskGetOptionsParams URI rendering" $ do
let normalize :: [(T.Text, Maybe T.Text)] -> [(T.Text, Maybe T.Text)]
normalize = L.sort
it "defaultTaskGetOptions emits no params" $
taskGetOptionsParams defaultTaskGetOptions
`shouldBe` []
it "renders wait_for_completion as lowercase true/false" $ do
let trueOpts = defaultTaskGetOptions {taskGetOptionsWaitForCompletion = Just True}
falseOpts = defaultTaskGetOptions {taskGetOptionsWaitForCompletion = Just False}
taskGetOptionsParams trueOpts
`shouldBe` [("wait_for_completion", Just "true")]
taskGetOptionsParams falseOpts
`shouldBe` [("wait_for_completion", Just "false")]
it "renders timeout as <n><suffix>" $ do
let opts = defaultTaskGetOptions {taskGetOptionsTimeout = Just (TimeUnitSeconds, 30)}
taskGetOptionsParams opts
`shouldBe` [("timeout", Just "30s")]
it "emits both params together when all are set" $ do
let opts =
defaultTaskGetOptions
{ taskGetOptionsWaitForCompletion = Just True,
taskGetOptionsTimeout = Just (TimeUnitMilliseconds, 500)
}
normalize (taskGetOptionsParams opts)
`shouldBe` [ ("timeout", Just "500ms"),
("wait_for_completion", Just "true")
]
-- ------------------------------------------------------------------ --
-- Live integration (requires ES/OpenSearch at ES_TEST_SERVER) --
-- ------------------------------------------------------------------ --
describe "listTasks live" $ do
it "returns the responding node when called unfiltered" $
withTestEnv $ do
resp <- performBHRequest $ listTasks Nothing
liftIO $ length (taskListResponseNodes resp) `shouldSatisfy` (>= 1)
it "accepts detailed=true and still parses" $
withTestEnv $ do
let opts = defaultTaskListOptions {taskListOptionsDetailed = Just True}
resp <- performBHRequest $ listTasks (Just opts)
liftIO $ length (taskListResponseNodes resp) `shouldSatisfy` (>= 1)
-- ------------------------------------------------------------------ --
-- getTask / getTaskWith live (requires ES/OpenSearch at ES_TEST_SERVER) --
-- ------------------------------------------------------------------ --
describe "getTaskWith live" $ do
it "getTask polls a reindex task to completion" $
withTestEnv $ do
_ <- insertData
let target = [qqIndexName|bloodhound-tests-twitter-gettask-default|]
_ <- tryPerformBHRequest $ deleteIndex target
_ <-
performBHRequest $
createIndex (IndexSettings (ShardCount 1) (ReplicaCount 0) defaultIndexMappingsLimits) target
taskNodeId <- assertRight =<< tryPerformBHRequest (reindexAsync $ mkReindexRequest testIndex target)
resp <- waitForCompletion 10 taskNodeId
liftIO $ taskResponseCompleted resp `shouldBe` True
it "getTaskWith forwards wait_for_completion=true and returns a completed TaskResponse" $
withTestEnv $ do
_ <- insertData
let target = [qqIndexName|bloodhound-tests-twitter-gettask-opts|]
_ <- tryPerformBHRequest $ deleteIndex target
_ <-
performBHRequest $
createIndex (IndexSettings (ShardCount 1) (ReplicaCount 0) defaultIndexMappingsLimits) target
taskNodeId <- assertRight =<< tryPerformBHRequest (reindexAsync $ mkReindexRequest testIndex target)
-- The single-doc reindex typically finishes in milliseconds, so
-- wait_for_completion=true returns immediately with a completed
-- TaskResponse rather than exercising the blocking path; the
-- assertion holds either way.
let opts =
defaultTaskGetOptions
{ taskGetOptionsWaitForCompletion = Just True
}
resp <- assertRight =<< tryPerformBHRequest (getTaskWith @ReindexResponse opts taskNodeId)
liftIO $ taskResponseCompleted resp `shouldBe` True
-- ------------------------------------------------------------------ --
-- cancelTask live (requires ES/OpenSearch at ES_TEST_SERVER) --
-- ------------------------------------------------------------------ --
describe "cancelTask live" $ do
it "cancels a running async reindex and lists the cancelled task" $
withTestEnv $ do
_ <- insertData
let target = [qqIndexName|bloodhound-tests-twitter-cancel-target|]
_ <- tryPerformBHRequest $ deleteIndex target
_ <-
performBHRequest $
createIndex (IndexSettings (ShardCount 1) (ReplicaCount 0) defaultIndexMappingsLimits) target
taskNodeId <- assertRight =<< tryPerformBHRequest (reindexAsync $ mkReindexRequest testIndex target)
-- Small race window: the single-doc reindex may finish before the
-- cancel lands, in which case the response lists no nodes. When the
-- task is still running the cancel response lists it (>= 1 node).
cancelled <- performBHRequest $ cancelTask taskNodeId
liftIO $ length (taskListResponseNodes cancelled) `shouldSatisfy` (>= 1)
assertRight :: (Show a, MonadFail m) => Either a b -> m b
assertRight (Left x) = fail $ "Expected Right, got Left: " <> show x
assertRight (Right x) = pure x
-- | Poll getTask until the task reports completed or the retry budget
-- runs out. The response is returned so its @completed@ flag can be
-- asserted.
waitForCompletion ::
(MonadBH m, MonadFail m) =>
Natural ->
TaskNodeId ->
m (TaskResponse ReindexResponse)
waitForCompletion 0 taskNodeId =
fail $ "Timed out waiting for task to complete, taskNodeId = " <> show taskNodeId
waitForCompletion n taskNodeId = do
resp <- assertRight =<< tryPerformBHRequest (getTask taskNodeId)
if taskResponseCompleted resp
then pure resp
else liftIO (threadDelay 100000) >> waitForCompletion (n - 1) taskNodeId