mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-12 02:22:45 +00:00
add tests for establishing connection asynchronously; retrieve correct key for activation on restart (#169)
This commit is contained in:
@@ -38,6 +38,7 @@ module Simplex.Messaging.Agent
|
||||
AgentMonad,
|
||||
AgentErrorMonad,
|
||||
getSMPAgentClient,
|
||||
disconnectAgentClient, -- used in tests
|
||||
createConnection,
|
||||
joinConnection,
|
||||
allowConnection,
|
||||
@@ -359,27 +360,28 @@ subscribeConnection' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
|
||||
subscribeConnection' c connId =
|
||||
withStore (`getConn` connId) >>= \case
|
||||
SomeConn _ (DuplexConnection _ rq sq) -> case status (sq :: SndQueue) of
|
||||
Confirmed -> withVerifyKey sq $ \sndKey -> do
|
||||
secureQueue c rq sndKey
|
||||
Confirmed -> withVerifyKey sq $ \verifyKey -> do
|
||||
conf <- withStore (`getAcceptedConfirmation` connId)
|
||||
secureQueue c rq $ senderKey (conf :: AcceptedConfirmation)
|
||||
withStore $ \st -> setRcvQueueStatus st rq Secured
|
||||
activateSecuredQueue rq sq sndKey
|
||||
activateSecuredQueue rq sq verifyKey
|
||||
Secured -> withVerifyKey sq $ activateSecuredQueue rq sq
|
||||
Active -> subscribeQueue c rq connId
|
||||
_ -> throwError $ INTERNAL "unexpected queue status"
|
||||
SomeConn _ (SndConnection _ sq) -> case status (sq :: SndQueue) of
|
||||
Confirmed -> withVerifyKey sq $ \sndKey ->
|
||||
activateQueueJoining c connId sq sndKey resumeInterval
|
||||
Confirmed -> withVerifyKey sq $ \verifyKey ->
|
||||
activateQueueJoining c connId sq verifyKey resumeInterval
|
||||
Active -> throwError $ CONN SIMPLEX
|
||||
_ -> throwError $ INTERNAL "unexpected queue status"
|
||||
SomeConn _ (RcvConnection _ rq) -> subscribeQueue c rq connId
|
||||
where
|
||||
withVerifyKey :: SndQueue -> (C.PublicKey -> m ()) -> m ()
|
||||
withVerifyKey sq action =
|
||||
let err = throwError $ INTERNAL "missing send queue public key"
|
||||
in maybe err action . C.publicKey $ sndPrivateKey sq
|
||||
let err = throwError $ INTERNAL "missing signing key public counterpart"
|
||||
in maybe err action . C.publicKey $ signKey sq
|
||||
activateSecuredQueue :: RcvQueue -> SndQueue -> C.PublicKey -> m ()
|
||||
activateSecuredQueue rq sq sndKey = do
|
||||
activateQueueInitiating c connId sq sndKey resumeInterval
|
||||
activateSecuredQueue rq sq verifyKey = do
|
||||
activateQueueInitiating c connId sq verifyKey resumeInterval
|
||||
subscribeQueue c rq connId
|
||||
|
||||
-- | Send message to the connection (SEND command) in Reader monad
|
||||
|
||||
+3
-50
@@ -5,32 +5,26 @@
|
||||
{-# LANGUAGE OverloadedStrings #-}
|
||||
{-# LANGUAGE PatternSynonyms #-}
|
||||
{-# LANGUAGE PostfixOperators #-}
|
||||
{-# LANGUAGE RankNTypes #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
|
||||
|
||||
module AgentTests where
|
||||
module AgentTests (agentTests) where
|
||||
|
||||
import AgentTests.FunctionalAPITests (functionalAPITests)
|
||||
import AgentTests.SQLiteTests (storeTests)
|
||||
import Control.Concurrent
|
||||
import Control.Monad.Except (catchError, runExceptT)
|
||||
import Control.Monad.IO.Unlift
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import SMPAgentClient
|
||||
import SMPClient (withSmpServer)
|
||||
import Simplex.Messaging.Agent
|
||||
import Simplex.Messaging.Agent.Env.SQLite (dbFile)
|
||||
import Simplex.Messaging.Agent.Protocol
|
||||
import Simplex.Messaging.Agent.Store (InternalId (..))
|
||||
import Simplex.Messaging.Protocol (ErrorType (..), MsgBody)
|
||||
import Simplex.Messaging.Transport (ATransport (..), TProxy (..), Transport (..))
|
||||
import System.Timeout
|
||||
import Test.Hspec
|
||||
import UnliftIO.STM
|
||||
|
||||
agentTests :: ATransport -> Spec
|
||||
agentTests (ATransport t) = do
|
||||
describe "Functional API" $ functionalAPITests (ATransport t)
|
||||
describe "SQLite store" storeTests
|
||||
describe "SMP agent protocol syntax" $ syntaxTests t
|
||||
describe "Establishing duplex connection" do
|
||||
@@ -46,8 +40,6 @@ agentTests (ATransport t) = do
|
||||
smpAgentTest2_2_2 $ testDuplexConnection t
|
||||
it "should connect via 2 servers and 2 agents (random IDs)" $
|
||||
smpAgentTest2_2_2 $ testDuplexConnRandomIds t
|
||||
it "should connect via one server using SMP agent clients" $
|
||||
withSmpServer (ATransport t) testAgentClient
|
||||
describe "Connection subscriptions" do
|
||||
it "should connect via one server and one agent" $
|
||||
smpAgentTest3_1_1 $ testSubscription t
|
||||
@@ -125,45 +117,6 @@ testDuplexConnection _ alice bob = do
|
||||
alice #: ("6", "bob", "DEL") #> ("6", "bob", OK)
|
||||
alice #:# "nothing else should be delivered to alice"
|
||||
|
||||
testAgentClient :: IO ()
|
||||
testAgentClient = do
|
||||
alice <- getSMPAgentClient cfg
|
||||
bob <- getSMPAgentClient cfg {dbFile = testDB2}
|
||||
Right () <- runExceptT $ do
|
||||
(bobId, qInfo) <- createConnection alice
|
||||
aliceId <- joinConnection bob qInfo "bob's connInfo"
|
||||
("", _, CONF confId "bob's connInfo") <- get alice
|
||||
allowConnection alice bobId confId "alice's connInfo"
|
||||
get alice ##> ("", bobId, CON)
|
||||
get bob ##> ("", aliceId, INFO "alice's connInfo")
|
||||
get bob ##> ("", aliceId, CON)
|
||||
InternalId 1 <- sendMessage alice bobId "hello"
|
||||
InternalId 2 <- sendMessage alice bobId "how are you?"
|
||||
get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
|
||||
get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False
|
||||
InternalId 3 <- sendMessage bob aliceId "hello too"
|
||||
InternalId 4 <- sendMessage bob aliceId "message 1"
|
||||
get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
|
||||
get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False
|
||||
suspendConnection alice bobId
|
||||
InternalId 0 <- sendMessage bob aliceId "message 2" `catchError` \(SMP AUTH) -> pure $ InternalId 0
|
||||
deleteConnection alice bobId
|
||||
liftIO $ noMessages alice "nothing else should be delivered to alice"
|
||||
pure ()
|
||||
where
|
||||
(##>) :: MonadIO m => m (ATransmission 'Agent) -> ATransmission 'Agent -> m ()
|
||||
a ##> t = a >>= \t' -> liftIO (t' `shouldBe` t)
|
||||
(=##>) :: MonadIO m => m (ATransmission 'Agent) -> (ATransmission 'Agent -> Bool) -> m ()
|
||||
a =##> p = a >>= \t -> liftIO (t `shouldSatisfy` p)
|
||||
noMessages :: AgentClient -> String -> Expectation
|
||||
noMessages c err = tryGet `shouldReturn` ()
|
||||
where
|
||||
tryGet =
|
||||
10000 `timeout` get c >>= \case
|
||||
Just _ -> error err
|
||||
_ -> return ()
|
||||
get c = atomically (readTBQueue $ subQ c)
|
||||
|
||||
testDuplexConnRandomIds :: Transport c => TProxy c -> c -> c -> IO ()
|
||||
testDuplexConnRandomIds _ alice bob = do
|
||||
("1", bobConn, Right (INV qInfo)) <- alice #: ("1", "", "NEW")
|
||||
|
||||
@@ -0,0 +1,151 @@
|
||||
{-# LANGUAGE DataKinds #-}
|
||||
{-# LANGUAGE GADTs #-}
|
||||
{-# LANGUAGE LambdaCase #-}
|
||||
{-# LANGUAGE OverloadedStrings #-}
|
||||
{-# LANGUAGE PatternSynonyms #-}
|
||||
{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
|
||||
|
||||
module AgentTests.FunctionalAPITests (functionalAPITests) where
|
||||
|
||||
import Control.Monad.Except (ExceptT, catchError, runExceptT)
|
||||
import Control.Monad.IO.Unlift
|
||||
import SMPAgentClient
|
||||
import SMPClient (withSmpServer)
|
||||
import Simplex.Messaging.Agent
|
||||
import Simplex.Messaging.Agent.Env.SQLite (dbFile)
|
||||
import Simplex.Messaging.Agent.Protocol
|
||||
import Simplex.Messaging.Agent.Store (InternalId (..))
|
||||
import Simplex.Messaging.Protocol (ErrorType (..), MsgBody)
|
||||
import Simplex.Messaging.Transport (ATransport (..))
|
||||
import System.Timeout
|
||||
import Test.Hspec
|
||||
import UnliftIO.STM
|
||||
|
||||
(##>) :: MonadIO m => m (ATransmission 'Agent) -> ATransmission 'Agent -> m ()
|
||||
a ##> t = a >>= \t' -> liftIO (t' `shouldBe` t)
|
||||
|
||||
(=##>) :: MonadIO m => m (ATransmission 'Agent) -> (ATransmission 'Agent -> Bool) -> m ()
|
||||
a =##> p = a >>= \t -> liftIO (t `shouldSatisfy` p)
|
||||
|
||||
get :: MonadIO m => AgentClient -> m (ATransmission 'Agent)
|
||||
get c = atomically (readTBQueue $ subQ c)
|
||||
|
||||
pattern Msg :: MsgBody -> ACommand 'Agent
|
||||
pattern Msg msgBody <- MSG MsgMeta {integrity = MsgOk} msgBody
|
||||
|
||||
functionalAPITests :: ATransport -> Spec
|
||||
functionalAPITests t = do
|
||||
describe "Establishing duplex connection" $
|
||||
it "should connect via one server using SMP agent clients" $
|
||||
withSmpServer t testAgentClient
|
||||
describe "Establishing connection asynchronously" $ do
|
||||
it "should connect with initiating client going offline" $
|
||||
withSmpServer t testAsyncInitiatingOffline
|
||||
it "should connect with joining client going offline before its queue activation" $
|
||||
withSmpServer t testAsyncJoiningOfflineBeforeActivation
|
||||
-- TODO a valid test case but not trivial to implement, probably requires some agent rework
|
||||
xit "should connect with joining client going offline after its queue activation" $
|
||||
withSmpServer t testAsyncJoiningOfflineAfterActivation
|
||||
it "should connect with both clients going offline" $
|
||||
withSmpServer t testAsyncBothOffline
|
||||
|
||||
testAgentClient :: IO ()
|
||||
testAgentClient = do
|
||||
alice <- getSMPAgentClient cfg
|
||||
bob <- getSMPAgentClient cfg {dbFile = testDB2}
|
||||
Right () <- runExceptT $ do
|
||||
(bobId, qInfo) <- createConnection alice
|
||||
aliceId <- joinConnection bob qInfo "bob's connInfo"
|
||||
("", _, CONF confId "bob's connInfo") <- get alice
|
||||
allowConnection alice bobId confId "alice's connInfo"
|
||||
get alice ##> ("", bobId, CON)
|
||||
get bob ##> ("", aliceId, INFO "alice's connInfo")
|
||||
get bob ##> ("", aliceId, CON)
|
||||
InternalId 1 <- sendMessage alice bobId "hello"
|
||||
InternalId 2 <- sendMessage alice bobId "how are you?"
|
||||
get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
|
||||
get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False
|
||||
InternalId 3 <- sendMessage bob aliceId "hello too"
|
||||
InternalId 4 <- sendMessage bob aliceId "message 1"
|
||||
get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
|
||||
get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False
|
||||
suspendConnection alice bobId
|
||||
InternalId 0 <- sendMessage bob aliceId "message 2" `catchError` \(SMP AUTH) -> pure $ InternalId 0
|
||||
deleteConnection alice bobId
|
||||
liftIO $ noMessages alice "nothing else should be delivered to alice"
|
||||
pure ()
|
||||
where
|
||||
noMessages :: AgentClient -> String -> Expectation
|
||||
noMessages c err = tryGet `shouldReturn` ()
|
||||
where
|
||||
tryGet =
|
||||
10000 `timeout` get c >>= \case
|
||||
Just _ -> error err
|
||||
_ -> return ()
|
||||
|
||||
testAsyncInitiatingOffline :: IO ()
|
||||
testAsyncInitiatingOffline = do
|
||||
alice <- getSMPAgentClient cfg
|
||||
bob <- getSMPAgentClient cfg {dbFile = testDB2}
|
||||
Right () <- runExceptT $ do
|
||||
(bobId, qInfo) <- createConnection alice
|
||||
disconnectAgentClient alice
|
||||
aliceId <- joinConnection bob qInfo "bob's connInfo"
|
||||
alice' <- liftIO $ getSMPAgentClient cfg
|
||||
subscribeConnection alice' bobId
|
||||
("", _, CONF confId "bob's connInfo") <- get alice'
|
||||
allowConnection alice' bobId confId "alice's connInfo"
|
||||
get alice' ##> ("", bobId, CON)
|
||||
get bob ##> ("", aliceId, INFO "alice's connInfo")
|
||||
get bob ##> ("", aliceId, CON)
|
||||
exchangeGreetings alice' bobId bob aliceId
|
||||
pure ()
|
||||
|
||||
testAsyncJoiningOfflineBeforeActivation :: IO ()
|
||||
testAsyncJoiningOfflineBeforeActivation = do
|
||||
alice <- getSMPAgentClient cfg
|
||||
bob <- getSMPAgentClient cfg {dbFile = testDB2}
|
||||
Right () <- runExceptT $ do
|
||||
(bobId, qInfo) <- createConnection alice
|
||||
aliceId <- joinConnection bob qInfo "bob's connInfo"
|
||||
disconnectAgentClient bob
|
||||
("", _, CONF confId "bob's connInfo") <- get alice
|
||||
allowConnection alice bobId confId "alice's connInfo"
|
||||
bob' <- liftIO $ getSMPAgentClient cfg {dbFile = testDB2}
|
||||
subscribeConnection bob' aliceId
|
||||
get alice ##> ("", bobId, CON)
|
||||
get bob' ##> ("", aliceId, INFO "alice's connInfo")
|
||||
get bob' ##> ("", aliceId, CON)
|
||||
exchangeGreetings alice bobId bob' aliceId
|
||||
pure ()
|
||||
|
||||
testAsyncJoiningOfflineAfterActivation :: IO ()
|
||||
testAsyncJoiningOfflineAfterActivation = error "not implemented"
|
||||
|
||||
testAsyncBothOffline :: IO ()
|
||||
testAsyncBothOffline = do
|
||||
alice <- getSMPAgentClient cfg
|
||||
bob <- getSMPAgentClient cfg {dbFile = testDB2}
|
||||
Right () <- runExceptT $ do
|
||||
(bobId, qInfo) <- createConnection alice
|
||||
disconnectAgentClient alice
|
||||
aliceId <- joinConnection bob qInfo "bob's connInfo"
|
||||
disconnectAgentClient bob
|
||||
alice' <- liftIO $ getSMPAgentClient cfg
|
||||
subscribeConnection alice' bobId
|
||||
("", _, CONF confId "bob's connInfo") <- get alice'
|
||||
allowConnection alice' bobId confId "alice's connInfo"
|
||||
bob' <- liftIO $ getSMPAgentClient cfg {dbFile = testDB2}
|
||||
subscribeConnection bob' aliceId
|
||||
get alice' ##> ("", bobId, CON)
|
||||
get bob' ##> ("", aliceId, INFO "alice's connInfo")
|
||||
get bob' ##> ("", aliceId, CON)
|
||||
exchangeGreetings alice' bobId bob' aliceId
|
||||
pure ()
|
||||
|
||||
exchangeGreetings :: AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO ()
|
||||
exchangeGreetings alice bobId bob aliceId = do
|
||||
InternalId 1 <- sendMessage alice bobId "hello"
|
||||
get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
|
||||
InternalId 2 <- sendMessage bob aliceId "hello too"
|
||||
get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{-# LANGUAGE TypeApplications #-}
|
||||
|
||||
import AgentTests
|
||||
import AgentTests (agentTests)
|
||||
import ProtocolErrorTests
|
||||
import ServerTests
|
||||
import Simplex.Messaging.Transport (TCP, Transport (..))
|
||||
|
||||
Reference in New Issue
Block a user