packages feed

distributed-process-zookeeper-0.2.0.0: test/Spec.hs

{-# OPTIONS_GHC -fno-warn-missing-signatures #-}
{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE FlexibleContexts #-}

module Main (main, spec) where

import           Test.Hspec
import           Control.Concurrent.Lifted
import           Control.Distributed.Process.MonadBaseControl()
import           Control.Distributed.Process hiding (bracket)
import           Control.Distributed.Process.Zookeeper
import           Control.Distributed.Process.Serializable (Serializable)
import qualified Database.Zookeeper as ZK
import Control.Distributed.Process.Node (initRemoteTable)
import Control.DeepSeq (NFData, deepseq)
import Control.Monad.Trans.Control (MonadBaseControl)
import Control.Monad (void)
import Control.Monad.IO.Class (MonadIO)
import Control.Exception.Enclosed (catchAny)
import Control.Exception.Lifted (throw)
import Data.ByteString.Char8 (pack)

main :: IO ()
main = hspec spec

spec :: Spec
spec = do
    describe "Test harness" $ do

            it "can do basic Process messaging" $ do
                testBoot $  do
                    parent <- getSelfPid
                    child <- spawnLocal $ do
                        saysee <- texpect :: Process String
                        send parent ("I said: " ++ saysee)
                    send child "foo"
                    said <- texpect :: Process String
                    return said
                `shouldReturn` "I said: foo"

    describe "C.D.P. Zookeeper" $ do

            it "will register local names with zookeeper" $ do
                testBoot $  do
                    self <- getSelfPid
                    register "testy" self
                    threadDelay 20000
                    [found] <- getCapable "testy"
                    return (found == self)
                `shouldReturn` True

            it "will only register prefixed local names if configured" $ do
                testBootWith defaultConfig {registerPrefix = "zk:"} zookeepers  $  do
                    self <- getSelfPid
                    register "testr" self
                    register "zk:testr2" self
                    threadDelay 10000
                    [] <- getCapable "testr"
                    [found] <- getCapable "zk:testr2"
                    return (found == self)
                `shouldReturn` True

            it "enables broadcast to specific services" $ do
                testBoot $ do
                    one <- newEmptyMVar
                    two <- newEmptyMVar
                    let slave mv = void $ fork $ testBoot $
                           do self <- getSelfPid
                              register "test-broadcast" self
                              threadDelay 50000 --finish registration
                              putMVar mv ()
                              "foundme" <- expect
                              putMVar mv ()
                              expect :: Process ()
                    slave one
                    slave two
                    takeMVar one
                    takeMVar two --hopefully they are both registered by now!
                    nsendCapable "test-broadcast" "foundme"
                    takeMVar one
                    takeMVar two
                `shouldReturn` ()

            it "enables broadcast to all nodes" $ do
                testBoot $ do
                    let slave mv =
                           do self <- getSelfPid
                              register "test-broadcast" self
                              threadDelay 50000 --finish registration
                              putMVar mv ()
                              "foundme" <- expect
                              putMVar mv ()
                    one <- forkSlave slave
                    two <- forkSlave slave
                    takeMVar one
                    takeMVar two --hopefully they are both registered by now!
                    nsendPeers "test-broadcast" "foundme"
                    takeMVar one
                    takeMVar two
                `shouldReturn` ()

            it "will run a single process with a global name" $ do
                testBoot $ do
                    pid <- getSelfPid
                    let racer mv =
                           do void $ registerCandidate "test-global" $
                                      do send pid ()
                                         expect :: Process ()
                              putMVar mv ()
                              expect :: Process ()

                    one <- forkSlave racer
                    two <- forkSlave racer
                    takeMVar one
                    takeMVar two
                    -- we should only get one
                    () <- expect :: Process ()
                    expectTimeout 100000
                `shouldReturn` (Nothing :: Maybe ())

            it "will elect a new global when one exits" $ do
                testBoot $ do
                    pid <- getSelfPid

                    let racer mv =
                           do Right gpid <- registerCandidate "test-global2" $
                                              do send pid ()
                                                 expect :: Process ()
                              putMVar mv gpid
                              expect :: Process ()

                    one <- forkSlave racer
                    two <- forkSlave racer
                    three <- forkSlave racer
                    first <- takeMVar one
                    void $ takeMVar two
                    void $ takeMVar three
                    exit first "because"
                    () <- expect :: Process ()
                    () <- expect :: Process ()
                    Just second <- whereisGlobal "test-global2"
                    -- make sure only ONE candidate is elected on exit of
                    -- first
                    Nothing <- expectTimeout 100000 :: Process (Maybe ())
                    return (first /= second)
                `shouldReturn` True

            it "will not leak processes when a candidate is attempted to register again" $ do
                testBoot $ do
                    nodeid <- getSelfNode

                    Right gpid <- registerCandidate "test-global3" $
                                      (expect :: Process ())

                    Right stats <- getNodeStats nodeid
                    let b4 = nodeStatsProcesses stats

                    Right same <- registerCandidate "test-global3" $
                                      (expect :: Process ())

                    Right stats' <- getNodeStats nodeid
                    let after' = nodeStatsProcesses stats'

                    return ((b4 + 1 == after') && (gpid == same))
                `shouldReturn` True

            it "will clear the name cache when a service exits" $ do
                testBoot $ do
                    parent <- getSelfPid
                    void $ spawnLocal $
                             do self <- getSelfPid
                                register "test-cache" self
                                threadDelay 10000
                                (expect :: Process ())
                                send parent ()
                    pid <- waitZKRegistration "test-cache"
                    send pid ()
                    (expect :: Process ())
                    threadDelay 10000
                    getCapable "test-cache"
                `shouldReturn` []

            it "will clear the name cache when a global process exits" $ do
                testBoot $ do
                    Right gpid <- registerCandidate "test-cache2" $
                                      (expect :: Process ())
                    send gpid ()
                    threadDelay 100000
                    whereisGlobal "test-cache2"
                `shouldReturn` Nothing

            it "will strictly evaluate command arguments in the client process" $ do
                testBoot $ do
                    result <- getSelfPid >>= registerZK (error "asplode")
                    return result
                    -- if the Name field of the Register constructor were
                    -- not strict, we'd get a ProcessLinkException here
                    -- because it would be evaluated in the Controller
                    -- process
                `shouldThrow` anyErrorCall

            describe "supports authorization" $ do

                it "can setup with restricted acl" $ do
                    testBootAuth "user:rightpassword" $ do
                        -- we just run this for the initialization with
                        -- the "correct" credentials and restricted ACL
                        return ()
                    `shouldReturn` ()

                it "will fail if there is an ACL with different credentials" $ do
                    testBootAuth "user:wrongpassword" $ do
                        return ()
                    `shouldThrow` linkException

linkException :: ProcessLinkException -> Bool
linkException _ = True

zookeepers = "localhost:2181"

forkSlave ma =
 do mv <- newEmptyMVar
    void $ fork $ testBoot (ma mv)
    return mv

testBoot :: (MonadBaseControl IO io, MonadIO io, Show a, NFData a)
         => Process a -> io a
testBoot = testBootWith defaultConfig zookeepers -- {logTrace = sayTrace}

testBootAuth :: (MonadBaseControl IO io, MonadIO io, Show a, NFData a)
             => String -> Process a -> io a
testBootAuth creds = testBootWith defaultConfig { credentials = Just ("digest", pack creds)
                                          , acl = ZK.CreatorAll
                                          }
                            (zookeepers ++ "/testauth")

testBootWith :: (MonadBaseControl IO io, MonadIO io, Show a, NFData a)
             => Config -> String -> Process a -> io a
testBootWith config servers ma = testProcessTimeout 1000 (ma >>= waitReturn )
                                      (liftIO . bootstrapWith config
                                                              "localhost"
                                                              "0"
                                                              servers
                                                              initRemoteTable)

waitReturn a = do threadDelay 100000; return a

testProcessTimeout :: (Show a, NFData a, MonadBaseControl IO io)
                => Int -> Process a -> (Process () -> io ()) -> io a
testProcessTimeout timeout ma runProc = do
    resultMV <- newEmptyMVar
    void . fork $
        catchAny (runProc $
                    do a <- ma
                       a `deepseq` putMVar resultMV a)
                 (\e -> putMVar resultMV (throw e))
    !result <- waitMVar resultMV
    return result
  where
    waitMVar mv = loop 0
      where loop lapsed
              | lapsed < timeout =
                  tryTakeMVar mv >>= maybe (delay >> loop (lapsed + 1))
                                           return
              | otherwise = error "Execution of Process test timed-out."
            delay = threadDelay 2000

texpect :: Serializable a => Process a
texpect = do
    gotit <- expectTimeout 500000 -- 500ms may as well be never
    case gotit of
        Nothing -> error "Timed out in test expect"
        Just v -> return v

waitZKRegistration name = loop
  where loop = do found <- getCapable name
                  case found of
                    []       -> threadDelay 1000 >> loop
                    pid : _ -> return pid