mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-08-27 22:34:59 +00:00
agent: fix race condition in subscription worker (results in stuck delivery), better error handling in batch operations (#962)
* debugging test: join connection when reply queue creation fails * more logs * more logs * more logs, fixed bug * cleanup * catch all errors * simplify * comment * remove client ID counter
This commit is contained in:
Evgeny Poberezkin
@@ -38,6 +38,7 @@ module Simplex.Messaging.Agent
|
||||
AgentErrorMonad,
|
||||
SubscriptionsInfo (..),
|
||||
getSMPAgentClient,
|
||||
getSMPAgentClient_,
|
||||
disconnectAgentClient,
|
||||
resumeAgentClient,
|
||||
withConnLock,
|
||||
@@ -176,11 +177,15 @@ import UnliftIO.STM
|
||||
|
||||
-- | Creates an SMP agent client instance
|
||||
getSMPAgentClient :: (MonadRandom m, MonadUnliftIO m) => AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient
|
||||
getSMPAgentClient cfg initServers store backgroundMode =
|
||||
getSMPAgentClient = getSMPAgentClient_ 1
|
||||
{-# INLINE getSMPAgentClient #-}
|
||||
|
||||
getSMPAgentClient_ :: (MonadRandom m, MonadUnliftIO m) => Int -> AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient
|
||||
getSMPAgentClient_ clientId cfg initServers store backgroundMode =
|
||||
liftIO (newSMPAgentEnv cfg store) >>= runReaderT runAgent
|
||||
where
|
||||
runAgent = do
|
||||
c <- getAgentClient initServers
|
||||
c <- getAgentClient clientId initServers
|
||||
void $ runAgentThreads c `forkFinally` const (disconnectAgentClient c)
|
||||
pure c
|
||||
runAgentThreads c
|
||||
@@ -461,8 +466,9 @@ withAgentEnv :: AgentClient -> ReaderT Env m a -> m a
|
||||
withAgentEnv c = (`runReaderT` agentEnv c)
|
||||
|
||||
-- | Creates an SMP agent client instance that receives commands and sends responses via 'TBQueue's.
|
||||
getAgentClient :: AgentMonad' m => InitialAgentServers -> m AgentClient
|
||||
getAgentClient initServers = ask >>= atomically . newAgentClient initServers
|
||||
getAgentClient :: AgentMonad' m => Int -> InitialAgentServers -> m AgentClient
|
||||
getAgentClient clientId initServers = ask >>= atomically . newAgentClient clientId initServers
|
||||
{-# INLINE getAgentClient #-}
|
||||
|
||||
logConnection :: MonadUnliftIO m => AgentClient -> Bool -> m ()
|
||||
logConnection c connected =
|
||||
|
||||
@@ -154,6 +154,7 @@ import Data.Text.Encoding
|
||||
import Data.Time (UTCTime, defaultTimeLocale, formatTime, getCurrentTime)
|
||||
import Data.Time.Clock.System (getSystemTime)
|
||||
import Data.Word (Word16)
|
||||
-- import GHC.Conc (unsafeIOToSTM)
|
||||
import Network.Socket (HostName)
|
||||
import Simplex.FileTransfer.Client (XFTPChunkSpec (..), XFTPClient, XFTPClientConfig (..), XFTPClientError)
|
||||
import qualified Simplex.FileTransfer.Client as X
|
||||
@@ -296,11 +297,11 @@ getAgentWorker' toW fromW name hasWork c key ws work = do
|
||||
whenExists w
|
||||
| hasWork = hasWorkToDo (toW w) $> w
|
||||
| otherwise = pure w
|
||||
runWorker w = runWorkerAsync (toW w) . void $ runExceptT runWork
|
||||
runWorker w = runWorkerAsync (toW w) runWork
|
||||
where
|
||||
runWork :: ExceptT AgentErrorType m ()
|
||||
runWork = tryAgentError (work w) >>= restartOrDelete
|
||||
restartOrDelete :: Either AgentErrorType () -> ExceptT AgentErrorType m ()
|
||||
runWork :: m ()
|
||||
runWork = tryAgentError' (work w) >>= restartOrDelete
|
||||
restartOrDelete :: Either AgentErrorType () -> m ()
|
||||
restartOrDelete e_ = do
|
||||
t <- liftIO getSystemTime
|
||||
maxRestarts <- asks $ maxWorkerRestartsPerMin . config
|
||||
@@ -382,8 +383,8 @@ data AgentStatsKey = AgentStatsKey
|
||||
}
|
||||
deriving (Eq, Ord, Show)
|
||||
|
||||
newAgentClient :: InitialAgentServers -> Env -> STM AgentClient
|
||||
newAgentClient InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
|
||||
newAgentClient :: Int -> InitialAgentServers -> Env -> STM AgentClient
|
||||
newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
|
||||
let qSize = tbqSize $ config agentEnv
|
||||
active <- newTVar True
|
||||
rcvQ <- newTBQueue qSize
|
||||
@@ -417,7 +418,6 @@ newAgentClient InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do
|
||||
smpSubWorkers <- TM.empty
|
||||
asyncClients <- newTAsyncs
|
||||
agentStats <- TM.empty
|
||||
clientId <- stateTVar (clientCounter agentEnv) $ \i -> let i' = i + 1 in (i', i')
|
||||
return
|
||||
AgentClient
|
||||
{ active,
|
||||
@@ -547,21 +547,22 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers} tSess =
|
||||
where
|
||||
newSubWorker v = do
|
||||
subWorkerId <- atomically $ stateTVar (workerSeq c) $ \next -> (next, next + 1)
|
||||
subWorkerAsync <- async $ runSubWorker subWorkerId `E.catchAny` const (atomically $ cleanup subWorkerId)
|
||||
subWorkerAsync <- async $ void (E.tryAny runSubWorker) >> atomically (cleanup v subWorkerId)
|
||||
atomically $ putTMVar v SubWorker {subWorkerId, subWorkerAsync}
|
||||
runSubWorker swId = do
|
||||
runSubWorker = do
|
||||
ri <- asks $ reconnectInterval . config
|
||||
timeoutCounts <- newTVarIO 0
|
||||
withRetryInterval ri $ \_ loop -> do
|
||||
pending <- atomically $ do
|
||||
qs <- RQ.getSessQueues tSess (pendingSubs c)
|
||||
when (null qs) $ cleanup swId
|
||||
pure qs
|
||||
pending <- atomically . RQ.getSessQueues tSess $ pendingSubs c
|
||||
forM_ (L.nonEmpty pending) $ \qs -> do
|
||||
void . runExceptT $ reconnectSMPClient timeoutCounts c tSess qs `catchAgentError` \_ -> pure ()
|
||||
void . tryAgentError' $ reconnectSMPClient timeoutCounts c tSess qs
|
||||
loop
|
||||
cleanup :: Int -> STM ()
|
||||
cleanup swId = removeTSessVar ((swId ==) . subWorkerId) tSess smpSubWorkers
|
||||
cleanup :: TMVar SubWorker -> Int -> STM ()
|
||||
cleanup v swId = do
|
||||
-- Here we wait until TMVar is not empty to prevent worker cleanup happening before worker is added to TMVar.
|
||||
-- Not waiting may result in terminated worker remaining in the map.
|
||||
whenM (isEmptyTMVar v) retry
|
||||
removeTSessVar ((swId ==) . subWorkerId) tSess smpSubWorkers
|
||||
|
||||
reconnectSMPClient :: forall m. AgentMonad m => TVar Int -> AgentClient -> SMPTransportSession -> NonEmpty RcvQueue -> m ()
|
||||
reconnectSMPClient tc c tSess@(_, srv, _) qs = do
|
||||
@@ -693,6 +694,9 @@ newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient clientC
|
||||
then retryAction
|
||||
else atomically $ do
|
||||
putTMVar clientVar (Left e)
|
||||
-- TODO This can result in removing some other client from the map.
|
||||
-- We need to identify these clients before they are connected and only remove if it's the same client in the map.
|
||||
-- probably ClientVar needs it's own ID at a point it's created, and not rely on session ID of the connected client.
|
||||
TM.delete tSess clients
|
||||
throwError e
|
||||
tryConnectAsync :: m ()
|
||||
@@ -1062,7 +1066,7 @@ sendTSessionBatches statCmd statBatchSize toRQ action c qs =
|
||||
in M.alter (Just . maybe [q] (q <|)) tSess m
|
||||
sendClientBatch :: (SMPTransportSession, NonEmpty q) -> m (BatchResponses AgentErrorType r)
|
||||
sendClientBatch (tSess@(userId, srv, _), qs') =
|
||||
runExceptT (getSMPServerClient c tSess) >>= \case
|
||||
tryAgentError' (getSMPServerClient c tSess) >>= \case
|
||||
Left e -> pure $ L.map ((,Left e) . toRQ) qs'
|
||||
Right smp -> liftIO $ do
|
||||
logServer "-->" c srv (bshow (length qs') <> " queues") statCmd
|
||||
|
||||
@@ -19,6 +19,7 @@ module Simplex.Messaging.Agent.Env.SQLite
|
||||
defaultAgentConfig,
|
||||
defaultReconnectInterval,
|
||||
tryAgentError,
|
||||
tryAgentError',
|
||||
catchAgentError,
|
||||
agentFinally,
|
||||
Env (..),
|
||||
@@ -33,6 +34,7 @@ module Simplex.Messaging.Agent.Env.SQLite
|
||||
)
|
||||
where
|
||||
|
||||
import Control.Monad
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.IO.Unlift
|
||||
import Control.Monad.Reader
|
||||
@@ -112,8 +114,7 @@ data AgentConfig = AgentConfig
|
||||
certificateFile :: FilePath,
|
||||
e2eEncryptVRange :: VersionRange,
|
||||
smpAgentVRange :: VersionRange,
|
||||
smpClientVRange :: VersionRange,
|
||||
initialClientId :: Int
|
||||
smpClientVRange :: VersionRange
|
||||
}
|
||||
|
||||
defaultReconnectInterval :: RetryInterval
|
||||
@@ -184,15 +185,13 @@ defaultAgentConfig =
|
||||
certificateFile = "/etc/opt/simplex-agent/agent.crt",
|
||||
e2eEncryptVRange = supportedE2EEncryptVRange,
|
||||
smpAgentVRange = supportedSMPAgentVRange,
|
||||
smpClientVRange = supportedSMPClientVRange,
|
||||
initialClientId = 0
|
||||
smpClientVRange = supportedSMPClientVRange
|
||||
}
|
||||
|
||||
data Env = Env
|
||||
{ config :: AgentConfig,
|
||||
store :: SQLiteStore,
|
||||
random :: TVar ChaChaDRG,
|
||||
clientCounter :: TVar Int,
|
||||
randomServer :: TVar StdGen,
|
||||
ntfSupervisor :: NtfSupervisor,
|
||||
xftpAgent :: XFTPAgent,
|
||||
@@ -200,14 +199,13 @@ data Env = Env
|
||||
}
|
||||
|
||||
newSMPAgentEnv :: AgentConfig -> SQLiteStore -> IO Env
|
||||
newSMPAgentEnv config@AgentConfig {initialClientId} store = do
|
||||
newSMPAgentEnv config store = do
|
||||
random <- C.newRandom
|
||||
clientCounter <- newTVarIO initialClientId
|
||||
randomServer <- newTVarIO =<< liftIO newStdGen
|
||||
ntfSupervisor <- atomically . newNtfSubSupervisor $ tbqSize config
|
||||
xftpAgent <- atomically newXFTPAgent
|
||||
multicastSubscribers <- newTMVarIO 0
|
||||
pure Env {config, store, random, clientCounter, randomServer, ntfSupervisor, xftpAgent, multicastSubscribers}
|
||||
pure Env {config, store, random, randomServer, ntfSupervisor, xftpAgent, multicastSubscribers}
|
||||
|
||||
createAgentStore :: FilePath -> ScrubbedBytes -> Bool -> MigrationConfirmation -> IO (Either MigrationError SQLiteStore)
|
||||
createAgentStore dbFilePath dbKey keepKey = createSQLiteStore dbFilePath dbKey keepKey Migrations.app
|
||||
@@ -250,6 +248,11 @@ tryAgentError :: AgentMonad m => m a -> m (Either AgentErrorType a)
|
||||
tryAgentError = tryAllErrors mkInternal
|
||||
{-# INLINE tryAgentError #-}
|
||||
|
||||
-- unlike runExceptT, this ensures we catch IO exceptions as well
|
||||
tryAgentError' :: AgentMonad' m => ExceptT AgentErrorType m a -> m (Either AgentErrorType a)
|
||||
tryAgentError' = fmap join . runExceptT . tryAgentError
|
||||
{-# INLINE tryAgentError' #-}
|
||||
|
||||
catchAgentError :: AgentMonad m => m a -> (AgentErrorType -> m a) -> m a
|
||||
catchAgentError = catchAllErrors mkInternal
|
||||
{-# INLINE catchAgentError #-}
|
||||
|
||||
@@ -48,9 +48,11 @@ runSMPAgentBlocking (ATransport t) cfg@AgentConfig {tcpPort, caCertificateFile,
|
||||
smpAgent _ = do
|
||||
-- tlsServerParams is not in Env to avoid breaking functional API w/t key and certificate generation
|
||||
tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
|
||||
clientId <- newTVarIO 0
|
||||
runTransportServer started tcpPort tlsServerParams defaultTransportServerConfig $ \(h :: c) -> do
|
||||
liftIO . putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion
|
||||
c <- getAgentClient initServers
|
||||
cId <- atomically $ stateTVar clientId $ \i -> (i + 1, i + 1)
|
||||
c <- getAgentClient cId initServers
|
||||
logConnection c True
|
||||
race_ (connectClient h c) (runAgentClient c)
|
||||
`E.finally` disconnectAgentClient c
|
||||
|
||||
@@ -375,8 +375,8 @@ runTestCfg2 aCfg bCfg baseMsgId runTest =
|
||||
|
||||
withAgentClientsCfg2 :: AgentConfig -> AgentConfig -> (AgentClient -> AgentClient -> IO ()) -> IO ()
|
||||
withAgentClientsCfg2 aCfg bCfg runTest = do
|
||||
a <- getSMPAgentClient' aCfg initAgentServers testDB
|
||||
b <- getSMPAgentClient' bCfg initAgentServers testDB2
|
||||
a <- getSMPAgentClient' 1 aCfg initAgentServers testDB
|
||||
b <- getSMPAgentClient' 2 bCfg initAgentServers testDB2
|
||||
runTest a b
|
||||
disconnectAgentClient a
|
||||
disconnectAgentClient b
|
||||
@@ -421,9 +421,9 @@ runAgentClientTest alice bob baseId = do
|
||||
|
||||
testAgentClient3 :: HasCallStack => IO ()
|
||||
testAgentClient3 = do
|
||||
a <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
b <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
c <- getSMPAgentClient' agentCfg initAgentServers testDB3
|
||||
a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
c <- getSMPAgentClient' 3 agentCfg initAgentServers testDB3
|
||||
runRight_ $ do
|
||||
(aIdForB, bId) <- makeConnection a b
|
||||
(aIdForC, cId) <- makeConnection a c
|
||||
@@ -496,7 +496,7 @@ testAsyncInitiatingOffline =
|
||||
(bobId, cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
|
||||
disconnectAgentClient alice
|
||||
aliceId <- joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe
|
||||
alice' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
alice' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB
|
||||
subscribeConnection alice' bobId
|
||||
("", _, CONF confId _ "bob's connInfo") <- get alice'
|
||||
allowConnection alice' bobId confId "alice's connInfo"
|
||||
@@ -513,7 +513,7 @@ testAsyncJoiningOfflineBeforeActivation =
|
||||
disconnectAgentClient bob
|
||||
("", _, CONF confId _ "bob's connInfo") <- get alice
|
||||
allowConnection alice bobId confId "alice's connInfo"
|
||||
bob' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
bob' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB2
|
||||
subscribeConnection bob' aliceId
|
||||
get alice ##> ("", bobId, CON)
|
||||
get bob' ##> ("", aliceId, INFO "alice's connInfo")
|
||||
@@ -527,11 +527,11 @@ testAsyncBothOffline =
|
||||
disconnectAgentClient alice
|
||||
aliceId <- joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe
|
||||
disconnectAgentClient bob
|
||||
alice' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
alice' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB
|
||||
subscribeConnection alice' bobId
|
||||
("", _, CONF confId _ "bob's connInfo") <- get alice'
|
||||
allowConnection alice' bobId confId "alice's connInfo"
|
||||
bob' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
bob' <- liftIO $ getSMPAgentClient' 4 agentCfg initAgentServers testDB2
|
||||
subscribeConnection bob' aliceId
|
||||
get alice' ##> ("", bobId, CON)
|
||||
get bob' ##> ("", aliceId, INFO "alice's connInfo")
|
||||
@@ -574,8 +574,8 @@ testAsyncHelloTimeout = do
|
||||
testAllowConnectionClientRestart :: HasCallStack => ATransport -> IO ()
|
||||
testAllowConnectionClientRestart t = do
|
||||
let initAgentServersSrv2 = initAgentServers {smp = userServers [noAuthSrv testSMPServer2]}
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg initAgentServersSrv2 testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg initAgentServersSrv2 testDB2
|
||||
withSmpServerStoreLogOn t testPort $ \_ -> do
|
||||
(aliceId, bobId, confId) <-
|
||||
withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do
|
||||
@@ -595,7 +595,7 @@ testAllowConnectionClientRestart t = do
|
||||
threadDelay 100000 -- give time to enqueue confirmation (enqueueConfirmation)
|
||||
disconnectAgentClient alice
|
||||
|
||||
alice2 <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
|
||||
|
||||
withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do
|
||||
runRight $ do
|
||||
@@ -613,8 +613,8 @@ testAllowConnectionClientRestart t = do
|
||||
|
||||
testIncreaseConnAgentVersion :: HasCallStack => ATransport -> IO ()
|
||||
testIncreaseConnAgentVersion t = do
|
||||
alice <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
|
||||
withSmpServerStoreMsgLogOn t testPort $ \_ -> do
|
||||
(aliceId, bobId) <- runRight $ do
|
||||
(aliceId, bobId) <- makeConnection alice bob
|
||||
@@ -626,7 +626,7 @@ testIncreaseConnAgentVersion t = do
|
||||
-- version doesn't increase if incompatible
|
||||
|
||||
disconnectAgentClient alice
|
||||
alice2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
|
||||
alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
|
||||
|
||||
runRight_ $ do
|
||||
subscribeConnection alice2 bobId
|
||||
@@ -637,7 +637,7 @@ testIncreaseConnAgentVersion t = do
|
||||
-- version increases if compatible
|
||||
|
||||
disconnectAgentClient bob
|
||||
bob2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
|
||||
bob2 <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
|
||||
|
||||
runRight_ $ do
|
||||
subscribeConnection bob2 aliceId
|
||||
@@ -648,7 +648,7 @@ testIncreaseConnAgentVersion t = do
|
||||
-- version doesn't decrease, even if incompatible
|
||||
|
||||
disconnectAgentClient alice2
|
||||
alice3 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 2 2} initAgentServers testDB
|
||||
alice3 <- getSMPAgentClient' 5 agentCfg {smpAgentVRange = mkVersionRange 2 2} initAgentServers testDB
|
||||
|
||||
runRight_ $ do
|
||||
subscribeConnection alice3 bobId
|
||||
@@ -657,7 +657,7 @@ testIncreaseConnAgentVersion t = do
|
||||
checkVersion bob2 aliceId 3
|
||||
|
||||
disconnectAgentClient bob2
|
||||
bob3 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 1} initAgentServers testDB2
|
||||
bob3 <- getSMPAgentClient' 6 agentCfg {smpAgentVRange = mkVersionRange 1 1} initAgentServers testDB2
|
||||
|
||||
runRight_ $ do
|
||||
subscribeConnection bob3 aliceId
|
||||
@@ -674,8 +674,8 @@ checkVersion c connId v = do
|
||||
|
||||
testIncreaseConnAgentVersionMaxCompatible :: HasCallStack => ATransport -> IO ()
|
||||
testIncreaseConnAgentVersionMaxCompatible t = do
|
||||
alice <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB2
|
||||
withSmpServerStoreMsgLogOn t testPort $ \_ -> do
|
||||
(aliceId, bobId) <- runRight $ do
|
||||
(aliceId, bobId) <- makeConnection alice bob
|
||||
@@ -687,9 +687,9 @@ testIncreaseConnAgentVersionMaxCompatible t = do
|
||||
-- version increases to max compatible
|
||||
|
||||
disconnectAgentClient alice
|
||||
alice2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
|
||||
alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
|
||||
disconnectAgentClient bob
|
||||
bob2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB2
|
||||
bob2 <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB2
|
||||
|
||||
runRight_ $ do
|
||||
subscribeConnection alice2 bobId
|
||||
@@ -702,8 +702,8 @@ testIncreaseConnAgentVersionMaxCompatible t = do
|
||||
|
||||
testIncreaseConnAgentVersionStartDifferentVersion :: HasCallStack => ATransport -> IO ()
|
||||
testIncreaseConnAgentVersionStartDifferentVersion t = do
|
||||
alice <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 2} initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
|
||||
withSmpServerStoreMsgLogOn t testPort $ \_ -> do
|
||||
(aliceId, bobId) <- runRight $ do
|
||||
(aliceId, bobId) <- makeConnection alice bob
|
||||
@@ -715,7 +715,7 @@ testIncreaseConnAgentVersionStartDifferentVersion t = do
|
||||
-- version increases to max compatible
|
||||
|
||||
disconnectAgentClient alice
|
||||
alice2 <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
|
||||
alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
|
||||
|
||||
runRight_ $ do
|
||||
subscribeConnection alice2 bobId
|
||||
@@ -727,8 +727,8 @@ testIncreaseConnAgentVersionStartDifferentVersion t = do
|
||||
|
||||
testDeliverClientRestart :: HasCallStack => ATransport -> IO ()
|
||||
testDeliverClientRestart t = do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
|
||||
(aliceId, bobId) <- withSmpServerStoreMsgLogOn t testPort $ \_ -> do
|
||||
runRight $ do
|
||||
@@ -743,7 +743,7 @@ testDeliverClientRestart t = do
|
||||
|
||||
disconnectAgentClient bob
|
||||
|
||||
bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
bob2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
|
||||
|
||||
withSmpServerStoreMsgLogOn t testPort $ \_ -> do
|
||||
runRight_ $ do
|
||||
@@ -758,8 +758,8 @@ testDeliverClientRestart t = do
|
||||
|
||||
testDuplicateMessage :: HasCallStack => ATransport -> IO ()
|
||||
testDuplicateMessage t = do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
(aliceId, bobId, bob1) <- withSmpServerStoreMsgLogOn t testPort $ \_ -> do
|
||||
(aliceId, bobId) <- runRight $ makeConnection alice bob
|
||||
runRight_ $ do
|
||||
@@ -769,7 +769,7 @@ testDuplicateMessage t = do
|
||||
disconnectAgentClient bob
|
||||
|
||||
-- if the agent user did not send ACK, the message will be delivered again
|
||||
bob1 <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
bob1 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
|
||||
runRight_ $ do
|
||||
subscribeConnection bob1 aliceId
|
||||
get bob1 =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
|
||||
@@ -790,8 +790,8 @@ testDuplicateMessage t = do
|
||||
disconnectAgentClient alice
|
||||
disconnectAgentClient bob1
|
||||
|
||||
alice2 <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
alice2 <- getSMPAgentClient' 4 agentCfg initAgentServers testDB
|
||||
bob2 <- getSMPAgentClient' 5 agentCfg initAgentServers testDB2
|
||||
|
||||
withSmpServerStoreMsgLogOn t testPort $ \_ -> do
|
||||
runRight_ $ do
|
||||
@@ -808,8 +808,8 @@ testDuplicateMessage t = do
|
||||
|
||||
testSkippedMessages :: HasCallStack => ATransport -> IO ()
|
||||
testSkippedMessages t = do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
(aliceId, bobId) <- withSmpServerStoreLogOn t testPort $ \_ -> do
|
||||
(aliceId, bobId) <- runRight $ makeConnection alice bob
|
||||
runRight_ $ do
|
||||
@@ -835,8 +835,8 @@ testSkippedMessages t = do
|
||||
|
||||
disconnectAgentClient alice
|
||||
|
||||
alice2 <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
|
||||
bob2 <- getSMPAgentClient' 4 agentCfg initAgentServers testDB2
|
||||
|
||||
withSmpServerStoreLogOn t testPort $ \_ -> do
|
||||
runRight_ $ do
|
||||
@@ -899,7 +899,7 @@ setupDesynchronizedRatchet alice bob = do
|
||||
-- importing database backup after progressing ratchet de-synchronizes ratchet
|
||||
liftIO $ renameFile (testDB2 <> ".bak") testDB2
|
||||
|
||||
bob2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
bob2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
|
||||
|
||||
runRight_ $ do
|
||||
subscribeConnection bob2 aliceId
|
||||
@@ -959,8 +959,8 @@ serverUpP = \case
|
||||
|
||||
testRatchetSyncClientRestart :: HasCallStack => ATransport -> IO ()
|
||||
testRatchetSyncClientRestart t = do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
(aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn t testPort $ \_ ->
|
||||
setupDesynchronizedRatchet alice bob
|
||||
("", "", DOWN _ _) <- nGet alice
|
||||
@@ -968,7 +968,7 @@ testRatchetSyncClientRestart t = do
|
||||
ConnectionStats {ratchetSyncState} <- runRight $ synchronizeRatchet bob2 aliceId False
|
||||
liftIO $ ratchetSyncState `shouldBe` RSStarted
|
||||
disconnectAgentClient bob2
|
||||
bob3 <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
bob3 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
|
||||
withSmpServerStoreMsgLogOn t testPort $ \_ -> do
|
||||
runRight_ $ do
|
||||
("", "", UP _ _) <- nGet alice
|
||||
@@ -984,8 +984,8 @@ testRatchetSyncClientRestart t = do
|
||||
|
||||
testRatchetSyncSuspendForeground :: HasCallStack => ATransport -> IO ()
|
||||
testRatchetSyncSuspendForeground t = do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
(aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn t testPort $ \_ ->
|
||||
setupDesynchronizedRatchet alice bob
|
||||
|
||||
@@ -1018,8 +1018,8 @@ testRatchetSyncSuspendForeground t = do
|
||||
|
||||
testRatchetSyncSimultaneous :: HasCallStack => ATransport -> IO ()
|
||||
testRatchetSyncSimultaneous t = do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
(aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn t testPort $ \_ ->
|
||||
setupDesynchronizedRatchet alice bob
|
||||
|
||||
@@ -1101,7 +1101,7 @@ testInactiveNoSubs :: ATransport -> IO ()
|
||||
testInactiveNoSubs t = do
|
||||
let cfg' = cfg {inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
|
||||
withSmpServerConfigOn t cfg' testPort $ \_ -> do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
runRight_ . void $ createConnection alice 1 True SCMInvitation Nothing SMOnlyCreate -- do not subscribe to pass noSubscriptions check
|
||||
Just (_, _, APC SAENone (CONNECT _ _)) <- timeout 2000000 $ atomically (readTBQueue $ subQ alice)
|
||||
Just (_, _, APC SAENone (DISCONNECT _ _)) <- timeout 5000000 $ atomically (readTBQueue $ subQ alice)
|
||||
@@ -1111,7 +1111,7 @@ testInactiveWithSubs :: ATransport -> IO ()
|
||||
testInactiveWithSubs t = do
|
||||
let cfg' = cfg {inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
|
||||
withSmpServerConfigOn t cfg' testPort $ \_ -> do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
runRight_ . void $ createConnection alice 1 True SCMInvitation Nothing SMSubscribe
|
||||
Nothing <- 800000 `timeout` get alice
|
||||
liftIO $ threadDelay 1200000
|
||||
@@ -1123,7 +1123,7 @@ testActiveClientNotDisconnected :: ATransport -> IO ()
|
||||
testActiveClientNotDisconnected t = do
|
||||
let cfg' = cfg {inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
|
||||
withSmpServerConfigOn t cfg' testPort $ \_ -> do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
ts <- getSystemTime
|
||||
runRight_ $ do
|
||||
(connId, _cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
|
||||
@@ -1214,8 +1214,8 @@ testSuspendingAgentTimeout t = withAgentClients2 $ \a b -> do
|
||||
|
||||
testBatchedSubscriptions :: Int -> Int -> ATransport -> IO ()
|
||||
testBatchedSubscriptions nCreate nDel t = do
|
||||
a <- getSMPAgentClient' agentCfg initAgentServers2 testDB
|
||||
b <- getSMPAgentClient' agentCfg initAgentServers2 testDB2
|
||||
a <- getSMPAgentClient' 1 agentCfg initAgentServers2 testDB
|
||||
b <- getSMPAgentClient' 2 agentCfg initAgentServers2 testDB2
|
||||
conns <- runServers $ do
|
||||
conns <- replicateM (nCreate :: Int) $ makeConnection a b
|
||||
forM_ conns $ \(aId, bId) -> exchangeGreetings a bId b aId
|
||||
@@ -1324,11 +1324,11 @@ testAsyncCommands =
|
||||
|
||||
testAsyncCommandsRestore :: ATransport -> IO ()
|
||||
testAsyncCommandsRestore t = do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
bobId <- runRight $ createConnectionAsync alice 1 "1" True SCMInvitation SMSubscribe
|
||||
liftIO $ noMessages alice "alice doesn't receive INV because server is down"
|
||||
disconnectAgentClient alice
|
||||
alice' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
alice' <- liftIO $ getSMPAgentClient' 2 agentCfg initAgentServers testDB
|
||||
withSmpServerStoreLogOn t testPort $ \_ -> do
|
||||
runRight_ $ do
|
||||
subscribeConnection alice' bobId
|
||||
@@ -1378,7 +1378,7 @@ testAcceptContactAsync =
|
||||
|
||||
testDeleteConnectionAsync :: ATransport -> IO ()
|
||||
testDeleteConnectionAsync t = do
|
||||
a <- getSMPAgentClient' agentCfg {initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3} initAgentServers testDB
|
||||
a <- getSMPAgentClient' 1 agentCfg {initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3} initAgentServers testDB
|
||||
connIds <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ do
|
||||
(bId1, _inv) <- createConnection a 1 True SCMInvitation Nothing SMSubscribe
|
||||
(bId2, _inv) <- createConnection a 1 True SCMInvitation Nothing SMSubscribe
|
||||
@@ -1398,8 +1398,8 @@ testDeleteConnectionAsync t = do
|
||||
testJoinConnectionAsyncReplyError :: HasCallStack => ATransport -> IO ()
|
||||
testJoinConnectionAsyncReplyError t = do
|
||||
let initAgentServersSrv2 = initAgentServers {smp = userServers [noAuthSrv testSMPServer2]}
|
||||
a <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
b <- getSMPAgentClient' agentCfg initAgentServersSrv2 testDB2
|
||||
a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
b <- getSMPAgentClient' 2 agentCfg initAgentServersSrv2 testDB2
|
||||
(aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ do
|
||||
bId <- createConnectionAsync a 1 "1" True SCMInvitation SMSubscribe
|
||||
("1", bId', INV (ACR _ qInfo)) <- get a
|
||||
@@ -1491,8 +1491,8 @@ testUsersNoServer t = withAgentClientsCfg2 aCfg agentCfg $ \a b -> do
|
||||
|
||||
testSwitchConnection :: InitialAgentServers -> IO ()
|
||||
testSwitchConnection servers = do
|
||||
a <- getSMPAgentClient' agentCfg servers testDB
|
||||
b <- getSMPAgentClient' agentCfg {initialClientId = 1} servers testDB2
|
||||
a <- getSMPAgentClient' 1 agentCfg servers testDB
|
||||
b <- getSMPAgentClient' 2 agentCfg servers testDB2
|
||||
runRight_ $ do
|
||||
(aId, bId) <- makeConnection a b
|
||||
exchangeGreetingsMsgId 4 a bId b aId
|
||||
@@ -1575,12 +1575,12 @@ testSwitchAsync servers = do
|
||||
testFullSwitch a bId b aId 16
|
||||
where
|
||||
withA :: (AgentClient -> IO a) -> IO a
|
||||
withA = withAgent agentCfg servers testDB
|
||||
withA = withAgent 1 agentCfg servers testDB
|
||||
withB :: (AgentClient -> IO a) -> IO a
|
||||
withB = withAgent agentCfg {initialClientId = 1} servers testDB2
|
||||
withB = withAgent 2 agentCfg servers testDB2
|
||||
|
||||
withAgent :: AgentConfig -> InitialAgentServers -> FilePath -> (AgentClient -> IO a) -> IO a
|
||||
withAgent cfg' servers dbPath = bracket (getSMPAgentClient' cfg' servers dbPath) disconnectAgentClient
|
||||
withAgent :: Int -> AgentConfig -> InitialAgentServers -> FilePath -> (AgentClient -> IO a) -> IO a
|
||||
withAgent clientId cfg' servers dbPath = bracket (getSMPAgentClient' clientId cfg' servers dbPath) disconnectAgentClient
|
||||
|
||||
sessionSubscribe :: (forall a. (AgentClient -> IO a) -> IO a) -> [ConnId] -> (AgentClient -> ExceptT AgentErrorType IO ()) -> IO ()
|
||||
sessionSubscribe withC connIds a =
|
||||
@@ -1593,8 +1593,8 @@ sessionSubscribe withC connIds a =
|
||||
|
||||
testSwitchDelete :: InitialAgentServers -> IO ()
|
||||
testSwitchDelete servers = do
|
||||
a <- getSMPAgentClient' agentCfg servers testDB
|
||||
b <- getSMPAgentClient' agentCfg {initialClientId = 1} servers testDB2
|
||||
a <- getSMPAgentClient' 1 agentCfg servers testDB
|
||||
b <- getSMPAgentClient' 2 agentCfg servers testDB2
|
||||
runRight_ $ do
|
||||
(aId, bId) <- makeConnection a b
|
||||
exchangeGreetingsMsgId 4 a bId b aId
|
||||
@@ -1656,9 +1656,9 @@ testAbortSwitchStarted servers = do
|
||||
testFullSwitch a bId b aId 18
|
||||
where
|
||||
withA :: (AgentClient -> IO a) -> IO a
|
||||
withA = withAgent agentCfg servers testDB
|
||||
withA = withAgent 1 agentCfg servers testDB
|
||||
withB :: (AgentClient -> IO a) -> IO a
|
||||
withB = withAgent agentCfg {initialClientId = 1} servers testDB2
|
||||
withB = withAgent 2 agentCfg servers testDB2
|
||||
|
||||
testAbortSwitchStartedReinitiate :: HasCallStack => InitialAgentServers -> IO ()
|
||||
testAbortSwitchStartedReinitiate servers = do
|
||||
@@ -1707,9 +1707,9 @@ testAbortSwitchStartedReinitiate servers = do
|
||||
testFullSwitch a bId b aId 18
|
||||
where
|
||||
withA :: (AgentClient -> IO a) -> IO a
|
||||
withA = withAgent agentCfg servers testDB
|
||||
withA = withAgent 1 agentCfg servers testDB
|
||||
withB :: (AgentClient -> IO a) -> IO a
|
||||
withB = withAgent agentCfg {initialClientId = 1} servers testDB2
|
||||
withB = withAgent 2 agentCfg servers testDB2
|
||||
|
||||
switchPhaseRcvP :: ConnId -> SwitchPhase -> [Maybe RcvSwitchStatus] -> ATransmission 'Agent -> Bool
|
||||
switchPhaseRcvP cId sphase swchStatuses = switchPhaseP cId QDRcv sphase (\stats -> rcvSwchStatuses' stats == swchStatuses)
|
||||
@@ -1761,9 +1761,9 @@ testCannotAbortSwitchSecured servers = do
|
||||
testFullSwitch a bId b aId 16
|
||||
where
|
||||
withA :: (AgentClient -> IO a) -> IO a
|
||||
withA = withAgent agentCfg servers testDB
|
||||
withA = withAgent 1 agentCfg servers testDB
|
||||
withB :: (AgentClient -> IO a) -> IO a
|
||||
withB = withAgent agentCfg {initialClientId = 1} servers testDB2
|
||||
withB = withAgent 2 agentCfg servers testDB2
|
||||
|
||||
testSwitch2Connections :: HasCallStack => InitialAgentServers -> IO ()
|
||||
testSwitch2Connections servers = do
|
||||
@@ -1819,9 +1819,9 @@ testSwitch2Connections servers = do
|
||||
testFullSwitch a bId2 b aId2 16
|
||||
where
|
||||
withA :: (AgentClient -> IO a) -> IO a
|
||||
withA = withAgent agentCfg servers testDB
|
||||
withA = withAgent 1 agentCfg servers testDB
|
||||
withB :: (AgentClient -> IO a) -> IO a
|
||||
withB = withAgent agentCfg {initialClientId = 1} servers testDB2
|
||||
withB = withAgent 2 agentCfg servers testDB2
|
||||
|
||||
testSwitch2ConnectionsAbort1 :: HasCallStack => InitialAgentServers -> IO ()
|
||||
testSwitch2ConnectionsAbort1 servers = do
|
||||
@@ -1872,14 +1872,14 @@ testSwitch2ConnectionsAbort1 servers = do
|
||||
testFullSwitch a bId2 b aId2 14
|
||||
where
|
||||
withA :: (AgentClient -> IO a) -> IO a
|
||||
withA = withAgent agentCfg servers testDB
|
||||
withA = withAgent 1 agentCfg servers testDB
|
||||
withB :: (AgentClient -> IO a) -> IO a
|
||||
withB = withAgent agentCfg {initialClientId = 1} servers testDB2
|
||||
withB = withAgent 2 agentCfg servers testDB2
|
||||
|
||||
testCreateQueueAuth :: HasCallStack => (Maybe BasicAuth, Version) -> (Maybe BasicAuth, Version) -> IO Int
|
||||
testCreateQueueAuth clnt1 clnt2 = do
|
||||
a <- getClient clnt1 testDB
|
||||
b <- getClient clnt2 testDB2
|
||||
a <- getClient 1 clnt1 testDB
|
||||
b <- getClient 2 clnt2 testDB2
|
||||
r <- runRight $ do
|
||||
tryError (createConnection a 1 True SCMInvitation Nothing SMSubscribe) >>= \case
|
||||
Left (SMP AUTH) -> pure 0
|
||||
@@ -1900,15 +1900,15 @@ testCreateQueueAuth clnt1 clnt2 = do
|
||||
disconnectAgentClient b
|
||||
pure r
|
||||
where
|
||||
getClient (clntAuth, clntVersion) db =
|
||||
getClient clientId (clntAuth, clntVersion) db =
|
||||
let servers = initAgentServers {smp = userServers [ProtoServerWithAuth testSMPServer clntAuth]}
|
||||
smpCfg = (defaultClientConfig :: ProtocolClientConfig) {serverVRange = mkVersionRange 4 clntVersion}
|
||||
in getSMPAgentClient' agentCfg {smpCfg} servers db
|
||||
in getSMPAgentClient' clientId agentCfg {smpCfg} servers db
|
||||
|
||||
testSMPServerConnectionTest :: ATransport -> Maybe BasicAuth -> SMPServerWithAuth -> IO (Maybe ProtocolTestFailure)
|
||||
testSMPServerConnectionTest t newQueueBasicAuth srv =
|
||||
withSmpServerConfigOn t cfg {newQueueBasicAuth} testPort2 $ \_ -> do
|
||||
a <- getSMPAgentClient' agentCfg initAgentServers testDB -- initially passed server is not running
|
||||
a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB -- initially passed server is not running
|
||||
runRight $ testProtocolServer a 1 srv
|
||||
|
||||
testRatchetAdHash :: HasCallStack => IO ()
|
||||
@@ -1941,8 +1941,8 @@ testDeliveryReceipts =
|
||||
|
||||
testDeliveryReceiptsVersion :: HasCallStack => ATransport -> IO ()
|
||||
testDeliveryReceiptsVersion t = do
|
||||
a <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
|
||||
b <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
|
||||
a <- getSMPAgentClient' 1 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB
|
||||
b <- getSMPAgentClient' 2 agentCfg {smpAgentVRange = mkVersionRange 1 3} initAgentServers testDB2
|
||||
withSmpServerStoreMsgLogOn t testPort $ \_ -> do
|
||||
(aId, bId) <- runRight $ do
|
||||
(aId, bId) <- makeConnection a b
|
||||
@@ -1962,8 +1962,8 @@ testDeliveryReceiptsVersion t = do
|
||||
|
||||
disconnectAgentClient a
|
||||
disconnectAgentClient b
|
||||
a' <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB
|
||||
b' <- getSMPAgentClient' agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB2
|
||||
a' <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB
|
||||
b' <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = mkVersionRange 1 4} initAgentServers testDB2
|
||||
|
||||
runRight_ $ do
|
||||
subscribeConnection a' bId
|
||||
@@ -2103,10 +2103,10 @@ testTwoUsers = withAgentClients2 $ \a b -> do
|
||||
hasClients :: HasCallStack => AgentClient -> Int -> ExceptT AgentErrorType IO ()
|
||||
hasClients c n = liftIO $ M.size <$> readTVarIO (smpClients c) `shouldReturn` n
|
||||
|
||||
getSMPAgentClient' :: AgentConfig -> InitialAgentServers -> FilePath -> IO AgentClient
|
||||
getSMPAgentClient' cfg' initServers dbPath = do
|
||||
getSMPAgentClient' :: Int -> AgentConfig -> InitialAgentServers -> FilePath -> IO AgentClient
|
||||
getSMPAgentClient' clientId cfg' initServers dbPath = do
|
||||
Right st <- liftIO $ createAgentStore dbPath "" False MCError
|
||||
getSMPAgentClient cfg' initServers st False
|
||||
getSMPAgentClient_ clientId cfg' initServers st False
|
||||
|
||||
testServerMultipleIdentities :: HasCallStack => IO ()
|
||||
testServerMultipleIdentities =
|
||||
@@ -2122,7 +2122,7 @@ testServerMultipleIdentities =
|
||||
-- this saves queue with second server identity
|
||||
Left (BROKER _ NETWORK) <- runExceptT $ joinConnection bob 1 True secondIdentityCReq "bob's connInfo" SMSubscribe
|
||||
disconnectAgentClient bob
|
||||
bob' <- liftIO $ getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
bob' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB2
|
||||
subscribeConnection bob' aliceId
|
||||
exchangeGreetingsMsgId 6 alice bobId bob' aliceId
|
||||
where
|
||||
|
||||
@@ -24,7 +24,7 @@ import NtfClient
|
||||
import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2)
|
||||
import SMPClient (cfg, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, xit')
|
||||
import Simplex.Messaging.Agent
|
||||
import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers)
|
||||
import Simplex.Messaging.Agent.Env.SQLite (InitialAgentServers)
|
||||
import Simplex.Messaging.Agent.Protocol
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Encoding.String
|
||||
@@ -98,7 +98,7 @@ notificationTests t =
|
||||
|
||||
testNotificationToken :: APNSMockServer -> IO ()
|
||||
testNotificationToken APNSMockServer {apnsQ} = do
|
||||
a <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
runRight_ $ do
|
||||
let tkn = DeviceToken PPApnsTest "abcd"
|
||||
NTRegistered <- registerNtfToken a tkn NMPeriodic
|
||||
@@ -126,7 +126,7 @@ testNtfTokenRepeatRegistration :: APNSMockServer -> IO ()
|
||||
testNtfTokenRepeatRegistration APNSMockServer {apnsQ} = do
|
||||
-- setLogLevel LogError -- LogDebug
|
||||
-- withGlobalLogging logCfg $ do
|
||||
a <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
runRight_ $ do
|
||||
let tkn = DeviceToken PPApnsTest "abcd"
|
||||
NTRegistered <- registerNtfToken a tkn NMPeriodic
|
||||
@@ -150,8 +150,8 @@ testNtfTokenSecondRegistration :: APNSMockServer -> IO ()
|
||||
testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
|
||||
-- setLogLevel LogError -- LogDebug
|
||||
-- withGlobalLogging logCfg $ do
|
||||
a <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
a' <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
a' <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
runRight_ $ do
|
||||
let tkn = DeviceToken PPApnsTest "abcd"
|
||||
NTRegistered <- registerNtfToken a tkn NMPeriodic
|
||||
@@ -185,7 +185,7 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
|
||||
|
||||
testNtfTokenServerRestart :: ATransport -> APNSMockServer -> IO ()
|
||||
testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
|
||||
a <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
let tkn = DeviceToken PPApnsTest "abcd"
|
||||
ntfData <- withNtfServer t . runRight $ do
|
||||
NTRegistered <- registerNtfToken a tkn NMPeriodic
|
||||
@@ -196,7 +196,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
|
||||
-- the new agent is created as otherwise when running the tests in CI the old agent was keeping the connection to the server
|
||||
threadDelay 1000000
|
||||
disconnectAgentClient a
|
||||
a' <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
a' <- getSMPAgentClient' 2 agentCfg initAgentServers testDB
|
||||
-- server stopped before token is verified, so now the attempt to verify it will return AUTH error but re-register token,
|
||||
-- so that repeat verification happens without restarting the clients, when notification arrives
|
||||
withNtfServer t . runRight_ $ do
|
||||
@@ -214,8 +214,8 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
|
||||
|
||||
testNotificationSubscriptionExistingConnection :: APNSMockServer -> IO ()
|
||||
testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
(bobId, aliceId, nonce, message) <- runRight $ do
|
||||
-- establish connection
|
||||
(bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
|
||||
@@ -247,7 +247,7 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
|
||||
Left (CMD PROHIBITED) <- runExceptT $ getNotificationMessage alice nonce message
|
||||
|
||||
-- aliceNtf client doesn't have subscription and is allowed to get notification message
|
||||
aliceNtf <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
aliceNtf <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
|
||||
runRight_ $ do
|
||||
(_, [SMPMsgMeta {msgFlags = MsgFlags True}]) <- getNotificationMessage aliceNtf nonce message
|
||||
pure ()
|
||||
@@ -272,8 +272,8 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
|
||||
|
||||
testNotificationSubscriptionNewConnection :: APNSMockServer -> IO ()
|
||||
testNotificationSubscriptionNewConnection APNSMockServer {apnsQ} = do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
runRight_ $ do
|
||||
-- alice registers notification token
|
||||
DeviceToken {} <- registerTestToken alice "abcd" NMInstant apnsQ
|
||||
@@ -330,8 +330,8 @@ registerTestToken a token mode apnsQ = do
|
||||
|
||||
testChangeNotificationsMode :: APNSMockServer -> IO ()
|
||||
testChangeNotificationsMode APNSMockServer {apnsQ} = do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
runRight_ $ do
|
||||
-- establish connection
|
||||
(bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
|
||||
@@ -396,8 +396,8 @@ testChangeNotificationsMode APNSMockServer {apnsQ} = do
|
||||
|
||||
testChangeToken :: APNSMockServer -> IO ()
|
||||
testChangeToken APNSMockServer {apnsQ} = do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
(aliceId, bobId) <- runRight $ do
|
||||
-- establish connection
|
||||
(bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
|
||||
@@ -419,7 +419,7 @@ testChangeToken APNSMockServer {apnsQ} = do
|
||||
pure (aliceId, bobId)
|
||||
disconnectAgentClient alice
|
||||
|
||||
alice1 <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
alice1 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
|
||||
runRight_ $ do
|
||||
subscribeConnection alice1 bobId
|
||||
-- change notification token
|
||||
@@ -441,8 +441,8 @@ testChangeToken APNSMockServer {apnsQ} = do
|
||||
|
||||
testNotificationsStoreLog :: ATransport -> APNSMockServer -> IO ()
|
||||
testNotificationsStoreLog t APNSMockServer {apnsQ} = do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
(aliceId, bobId) <- withNtfServerStoreLog t $ \threadId -> runRight $ do
|
||||
(aliceId, bobId) <- makeConnection alice bob
|
||||
_ <- registerTestToken alice "abcd" NMInstant apnsQ
|
||||
@@ -469,8 +469,8 @@ testNotificationsStoreLog t APNSMockServer {apnsQ} = do
|
||||
|
||||
testNotificationsSMPRestart :: ATransport -> APNSMockServer -> IO ()
|
||||
testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
|
||||
alice <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
(aliceId, bobId) <- withSmpServerStoreLogOn t testPort $ \threadId -> runRight $ do
|
||||
(aliceId, bobId) <- makeConnection alice bob
|
||||
_ <- registerTestToken alice "abcd" NMInstant apnsQ
|
||||
@@ -501,8 +501,8 @@ testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
|
||||
|
||||
testNotificationsSMPRestartBatch :: Int -> ATransport -> APNSMockServer -> IO ()
|
||||
testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do
|
||||
a <- getSMPAgentClient' agentCfg initAgentServers2 testDB
|
||||
b <- getSMPAgentClient' agentCfg initAgentServers2 testDB2
|
||||
a <- getSMPAgentClient' 1 agentCfg initAgentServers2 testDB
|
||||
b <- getSMPAgentClient' 2 agentCfg initAgentServers2 testDB2
|
||||
threadDelay 1000000
|
||||
conns <- runServers $ do
|
||||
conns <- replicateM (n :: Int) $ makeConnection a b
|
||||
@@ -550,8 +550,8 @@ testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do
|
||||
|
||||
testSwitchNotifications :: InitialAgentServers -> APNSMockServer -> IO ()
|
||||
testSwitchNotifications servers APNSMockServer {apnsQ} = do
|
||||
a <- getSMPAgentClient' agentCfg servers testDB
|
||||
b <- getSMPAgentClient' agentCfg {initialClientId = 1} servers testDB2
|
||||
a <- getSMPAgentClient' 1 agentCfg servers testDB
|
||||
b <- getSMPAgentClient' 2 agentCfg servers testDB2
|
||||
runRight_ $ do
|
||||
(aId, bId) <- makeConnection a b
|
||||
exchangeGreetingsMsgId 4 a bId b aId
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
{-# LANGUAGE ConstraintKinds #-}
|
||||
{-# LANGUAGE DuplicateRecordFields #-}
|
||||
{-# LANGUAGE GADTs #-}
|
||||
{-# LANGUAGE NamedFieldPuns #-}
|
||||
{-# LANGUAGE NumericUnderscores #-}
|
||||
{-# LANGUAGE OverloadedLists #-}
|
||||
{-# LANGUAGE OverloadedStrings #-}
|
||||
@@ -194,8 +195,8 @@ agentCfg =
|
||||
{ tcpPort = agentTestPort,
|
||||
tbqSize = 4,
|
||||
-- database = testDB,
|
||||
smpCfg = defaultClientConfig {qSize = 1, defaultTransport = (testPort, transport @TLS)},
|
||||
ntfCfg = defaultClientConfig {qSize = 1, defaultTransport = (ntfTestPort, transport @TLS)},
|
||||
smpCfg = defaultClientConfig {qSize = 1, defaultTransport = (testPort, transport @TLS), networkConfig},
|
||||
ntfCfg = defaultClientConfig {qSize = 1, defaultTransport = (ntfTestPort, transport @TLS), networkConfig},
|
||||
reconnectInterval = defaultReconnectInterval {initialInterval = 50_000},
|
||||
xftpNotifyErrsOnRetry = False,
|
||||
ntfWorkerDelay = 100,
|
||||
@@ -204,6 +205,8 @@ agentCfg =
|
||||
privateKeyFile = "tests/fixtures/server.key",
|
||||
certificateFile = "tests/fixtures/server.crt"
|
||||
}
|
||||
where
|
||||
networkConfig = defaultNetworkConfig {tcpConnectTimeout = 3_000_000, tcpTimeout = 2_000_000}
|
||||
|
||||
type AgentTestMonad m = (MonadUnliftIO m, MonadRandom m, MonadFail m)
|
||||
|
||||
|
||||
+38
-38
@@ -94,18 +94,18 @@ testXFTPAgentSendReceive :: HasCallStack => IO ()
|
||||
testXFTPAgentSendReceive = withXFTPServer $ do
|
||||
filePath <- createRandomFile
|
||||
-- send file, delete snd file internally
|
||||
sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
(rfd1, rfd2) <- runRight $ do
|
||||
(sfId, _, rfd1, rfd2) <- testSend sndr filePath
|
||||
xftpDeleteSndFileInternal sndr sfId
|
||||
pure (rfd1, rfd2)
|
||||
|
||||
-- receive file, delete rcv file
|
||||
testReceiveDelete rfd1 filePath
|
||||
testReceiveDelete rfd2 filePath
|
||||
testReceiveDelete 2 rfd1 filePath
|
||||
testReceiveDelete 3 rfd2 filePath
|
||||
where
|
||||
testReceiveDelete rfd originalFilePath = do
|
||||
rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
testReceiveDelete clientId rfd originalFilePath = do
|
||||
rcp <- getSMPAgentClient' clientId agentCfg initAgentServers testDB2
|
||||
runRight_ $ do
|
||||
rfId <- testReceive rcp rfd originalFilePath
|
||||
xftpDeleteRcvFile rcp rfId
|
||||
@@ -118,17 +118,17 @@ testXFTPAgentSendReceiveEncrypted = withXFTPServer $ do
|
||||
s <- LB.readFile filePath
|
||||
file <- atomically $ CryptoFile (senderFiles </> "encrypted_testfile") . Just <$> CF.randomArgs g
|
||||
runRight_ $ CF.writeFile file s
|
||||
sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
(rfd1, rfd2) <- runRight $ do
|
||||
(sfId, _, rfd1, rfd2) <- testSendCF sndr file
|
||||
xftpDeleteSndFileInternal sndr sfId
|
||||
pure (rfd1, rfd2)
|
||||
-- receive file, delete rcv file
|
||||
testReceiveDelete rfd1 filePath g
|
||||
testReceiveDelete rfd2 filePath g
|
||||
testReceiveDelete 2 rfd1 filePath g
|
||||
testReceiveDelete 3 rfd2 filePath g
|
||||
where
|
||||
testReceiveDelete rfd originalFilePath g = do
|
||||
rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
testReceiveDelete clientId rfd originalFilePath g = do
|
||||
rcp <- getSMPAgentClient' clientId agentCfg initAgentServers testDB2
|
||||
cfArgs <- atomically $ Just <$> CF.randomArgs g
|
||||
runRight_ $ do
|
||||
rfId <- testReceiveCF rcp rfd cfArgs originalFilePath
|
||||
@@ -188,13 +188,13 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
|
||||
|
||||
rfd <- withXFTPServerStoreLogOn $ \_ -> do
|
||||
-- send file
|
||||
sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
runRight $ do
|
||||
(_, _, rfd, _) <- testSend sndr filePath
|
||||
pure rfd
|
||||
|
||||
-- receive file - should not succeed with server down
|
||||
rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
rfId <- runRight $ do
|
||||
xftpStartWorkers rcp (Just recipientFiles)
|
||||
rfId <- xftpReceiveFile rcp 1 rfd Nothing
|
||||
@@ -208,7 +208,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
|
||||
|
||||
withXFTPServerStoreLogOn $ \_ -> do
|
||||
-- receive file - should start downloading with server up
|
||||
rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
rcp' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
|
||||
runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
|
||||
("", rfId', RFPROG _ _) <- rfGet rcp'
|
||||
liftIO $ rfId' `shouldBe` rfId
|
||||
@@ -218,7 +218,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
|
||||
|
||||
withXFTPServerStoreLogOn $ \_ -> do
|
||||
-- receive file - should continue downloading with server up
|
||||
rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
rcp' <- getSMPAgentClient' 4 agentCfg initAgentServers testDB2
|
||||
runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
|
||||
rfProgress rcp' $ mb 18
|
||||
("", rfId', RFDONE path) <- rfGet rcp'
|
||||
@@ -236,13 +236,13 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
|
||||
|
||||
rfd <- withXFTPServerStoreLogOn $ \_ -> do
|
||||
-- send file
|
||||
sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
runRight $ do
|
||||
(_, _, rfd, _) <- testSend sndr filePath
|
||||
pure rfd
|
||||
|
||||
-- receive file - should not succeed with server down
|
||||
rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
rfId <- runRight $ do
|
||||
xftpStartWorkers rcp (Just recipientFiles)
|
||||
rfId <- xftpReceiveFile rcp 1 rfd Nothing
|
||||
@@ -256,7 +256,7 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
|
||||
|
||||
withXFTPServerThreadOn $ \_ -> do
|
||||
-- receive file - should fail with AUTH error
|
||||
rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
rcp' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
|
||||
runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
|
||||
("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp'
|
||||
rfId' `shouldBe` rfId
|
||||
@@ -269,7 +269,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
|
||||
filePath <- createRandomFile
|
||||
|
||||
-- send file - should not succeed with server down
|
||||
sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
sfId <- runRight $ do
|
||||
xftpStartWorkers sndr (Just senderFiles)
|
||||
sfId <- xftpSendFile sndr 1 (CF.plain filePath) 2
|
||||
@@ -286,7 +286,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
|
||||
|
||||
withXFTPServerStoreLogOn $ \_ -> do
|
||||
-- send file - should start uploading with server up
|
||||
sndr' <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr' <- getSMPAgentClient' 2 agentCfg initAgentServers testDB
|
||||
runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
|
||||
("", sfId', SFPROG _ _) <- sfGet sndr'
|
||||
liftIO $ sfId' `shouldBe` sfId
|
||||
@@ -296,7 +296,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
|
||||
|
||||
withXFTPServerStoreLogOn $ \_ -> do
|
||||
-- send file - should continue uploading with server up
|
||||
sndr' <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
|
||||
runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
|
||||
sfProgress sndr' $ mb 18
|
||||
("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr'
|
||||
@@ -308,7 +308,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
|
||||
doesFileExist encPath `shouldReturn` False
|
||||
|
||||
-- receive file
|
||||
rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
rcp <- getSMPAgentClient' 4 agentCfg initAgentServers testDB2
|
||||
runRight_ . void $
|
||||
testReceive rcp rfd1 filePath
|
||||
|
||||
@@ -318,7 +318,7 @@ testXFTPAgentSendCleanup = withGlobalLogging logCfgNoLogs $ do
|
||||
|
||||
sfId <- withXFTPServerStoreLogOn $ \_ -> do
|
||||
-- send file
|
||||
sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
sfId <- runRight $ do
|
||||
xftpStartWorkers sndr (Just senderFiles)
|
||||
sfId <- xftpSendFile sndr 1 (CF.plain filePath) 2
|
||||
@@ -339,7 +339,7 @@ testXFTPAgentSendCleanup = withGlobalLogging logCfgNoLogs $ do
|
||||
|
||||
withXFTPServerThreadOn $ \_ -> do
|
||||
-- send file - should fail with AUTH error
|
||||
sndr' <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr' <- getSMPAgentClient' 2 agentCfg initAgentServers testDB
|
||||
runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
|
||||
("", sfId', SFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- sfGet sndr'
|
||||
sfId' `shouldBe` sfId
|
||||
@@ -354,11 +354,11 @@ testXFTPAgentDelete = withGlobalLogging logCfgNoLogs $
|
||||
filePath <- createRandomFile
|
||||
|
||||
-- send file
|
||||
sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
(sfId, sndDescr, rfd1, rfd2) <- runRight $ testSend sndr filePath
|
||||
|
||||
-- receive file
|
||||
rcp1 <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
rcp1 <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
runRight_ . void $
|
||||
testReceive rcp1 rfd1 filePath
|
||||
|
||||
@@ -376,7 +376,7 @@ testXFTPAgentDelete = withGlobalLogging logCfgNoLogs $
|
||||
length <$> listDirectory xftpServerFiles `shouldReturn` 0
|
||||
|
||||
-- receive file - should fail with AUTH error
|
||||
rcp2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
rcp2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2
|
||||
runRight $ do
|
||||
xftpStartWorkers rcp2 (Just recipientFiles)
|
||||
rfId <- xftpReceiveFile rcp2 1 rfd2 Nothing
|
||||
@@ -389,11 +389,11 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
|
||||
|
||||
(sfId, sndDescr, rfd2) <- withXFTPServerStoreLogOn $ \_ -> do
|
||||
-- send file
|
||||
sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
(sfId, sndDescr, rfd1, rfd2) <- runRight $ testSend sndr filePath
|
||||
|
||||
-- receive file
|
||||
rcp1 <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
rcp1 <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
runRight_ . void $
|
||||
testReceive rcp1 rfd1 filePath
|
||||
disconnectAgentClient rcp1
|
||||
@@ -401,7 +401,7 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
|
||||
pure (sfId, sndDescr, rfd2)
|
||||
|
||||
-- delete file - should not succeed with server down
|
||||
sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
|
||||
runRight $ do
|
||||
xftpStartWorkers sndr (Just senderFiles)
|
||||
xftpDeleteSndFileRemote sndr 1 sfId sndDescr
|
||||
@@ -413,14 +413,14 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
|
||||
|
||||
withXFTPServerStoreLogOn $ \_ -> do
|
||||
-- delete file - should succeed with server up
|
||||
sndr' <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr' <- getSMPAgentClient' 4 agentCfg initAgentServers testDB
|
||||
runRight_ $ xftpStartWorkers sndr' (Just senderFiles)
|
||||
|
||||
threadDelay 1000000
|
||||
length <$> listDirectory xftpServerFiles `shouldReturn` 0
|
||||
|
||||
-- receive file - should fail with AUTH error
|
||||
rcp2 <- getSMPAgentClient' agentCfg initAgentServers testDB3
|
||||
rcp2 <- getSMPAgentClient' 5 agentCfg initAgentServers testDB3
|
||||
runRight $ do
|
||||
xftpStartWorkers rcp2 (Just recipientFiles)
|
||||
rfId <- xftpReceiveFile rcp2 1 rfd2 Nothing
|
||||
@@ -433,11 +433,11 @@ testXFTPAgentDeleteOnServer = withGlobalLogging logCfgNoLogs $
|
||||
filePath1 <- createRandomFile' "testfile1"
|
||||
|
||||
-- send file 1
|
||||
sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
(_, _, rfd1_1, rfd1_2) <- runRight $ testSend sndr filePath1
|
||||
|
||||
-- receive file 1 successfully
|
||||
rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
runRight_ . void $
|
||||
testReceive rcp rfd1_1 filePath1
|
||||
|
||||
@@ -471,11 +471,11 @@ testXFTPAgentExpiredOnServer = withGlobalLogging logCfgNoLogs $ do
|
||||
filePath1 <- createRandomFile' "testfile1"
|
||||
|
||||
-- send file 1
|
||||
sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
(_, _, rfd1_1, rfd1_2) <- runRight $ testSend sndr filePath1
|
||||
|
||||
-- receive file 1 successfully
|
||||
rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
runRight_ . void $
|
||||
testReceive rcp rfd1_1 filePath1
|
||||
|
||||
@@ -509,7 +509,7 @@ testXFTPAgentRequestAdditionalRecipientIDs = withXFTPServer $ do
|
||||
filePath <- createRandomFile
|
||||
|
||||
-- send file
|
||||
sndr <- getSMPAgentClient' agentCfg initAgentServers testDB
|
||||
sndr <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
|
||||
rfds <- runRight $ do
|
||||
xftpStartWorkers sndr (Just senderFiles)
|
||||
sfId <- xftpSendFile sndr 1 (CF.plain filePath) 500
|
||||
@@ -522,7 +522,7 @@ testXFTPAgentRequestAdditionalRecipientIDs = withXFTPServer $ do
|
||||
|
||||
-- receive file using different descriptions
|
||||
-- ! revise number of recipients and indexes if xftpMaxRecipientsPerRequest is changed
|
||||
rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
|
||||
rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
|
||||
runRight_ $ do
|
||||
void $ testReceive rcp (head rfds) filePath
|
||||
void $ testReceive rcp (rfds !! 99) filePath
|
||||
@@ -532,5 +532,5 @@ testXFTPAgentRequestAdditionalRecipientIDs = withXFTPServer $ do
|
||||
testXFTPServerTest :: HasCallStack => Maybe BasicAuth -> XFTPServerWithAuth -> IO (Maybe ProtocolTestFailure)
|
||||
testXFTPServerTest newFileBasicAuth srv =
|
||||
withXFTPServerCfg testXFTPServerConfig {newFileBasicAuth, xftpPort = xftpTestPort2} $ \_ -> do
|
||||
a <- getSMPAgentClient' agentCfg initAgentServers testDB -- initially passed server is not running
|
||||
a <- getSMPAgentClient' 1 agentCfg initAgentServers testDB -- initially passed server is not running
|
||||
runRight $ testProtocolServer a 1 srv
|
||||
|
||||
Reference in New Issue
Block a user