Split kind tests into component-owned probes
This commit is contained in:
parent
b3c0f6d58a
commit
691ad92e30
1 changed files with 4269 additions and 3108 deletions
386
test/StreamingProbe.hs
Normal file
386
test/StreamingProbe.hs
Normal file
|
|
@ -0,0 +1,386 @@
|
|||
module Main (main) where
|
||||
|
||||
import Control.Concurrent (threadDelay)
|
||||
import Control.Concurrent.Async (concurrently)
|
||||
import Control.Exception (SomeException, try)
|
||||
import Control.Monad (replicateM_, unless)
|
||||
import Crypto.Hash
|
||||
( Digest
|
||||
, SHA256
|
||||
, hashFinalize
|
||||
, hashInit
|
||||
, hashUpdate
|
||||
)
|
||||
import qualified Data.ByteArray as BA
|
||||
import qualified Data.ByteString as BS
|
||||
import qualified Data.ByteString.Base16 as B16
|
||||
import qualified Data.ByteString.Char8 as B8
|
||||
import System.Directory (getFileSize)
|
||||
import System.Environment (getArgs)
|
||||
import System.Exit (ExitCode (..))
|
||||
import System.IO
|
||||
( IOMode (..)
|
||||
, hClose
|
||||
, withBinaryFile
|
||||
)
|
||||
import System.IO.Temp
|
||||
( withSystemTempDirectory
|
||||
, withSystemTempFile
|
||||
)
|
||||
import System.Process (readProcessWithExitCode)
|
||||
import Text.Read (readMaybe)
|
||||
import Nuthatch.Probe.Assert
|
||||
import Nuthatch.Probe.Curl
|
||||
import Nuthatch.Probe.Oidc
|
||||
import Nuthatch.Probe.ProductClient
|
||||
|
||||
main :: IO ()
|
||||
main =
|
||||
withSystemTempDirectory "nuthatch-streaming-probe" $ \directory -> do
|
||||
context <- newProbeContext directory
|
||||
user <- requireEnv "NUTHATCH_OIDC_USER_A"
|
||||
session <- login context False user "streaming-a"
|
||||
getArgs >>= \case
|
||||
["large"] -> verifyLarge context session directory
|
||||
["throttle"] -> verifyThrottle context session directory
|
||||
["refill"] -> verifyRefill context session directory
|
||||
["unavailable"] -> verifyUnavailable context session
|
||||
_ -> fail "expected large, throttle, refill, or unavailable mode"
|
||||
|
||||
verifyLarge :: ProbeContext -> Session -> FilePath -> IO ()
|
||||
verifyLarge context session directory = do
|
||||
largeBytes <- readEnvInteger "NUTHATCH_STREAMING_LARGE_BYTES"
|
||||
overLimit <- readEnvInteger "NUTHATCH_STREAMING_OVER_LIMIT_BYTES"
|
||||
let largePath = directory <> "/large.bin"
|
||||
downloadPath = directory <> "/large-download.bin"
|
||||
writeZeros largePath largeBytes
|
||||
largeDigest <- hashFile largePath
|
||||
uploaded <- uploadFileDeclared
|
||||
context session "streaming-large.bin"
|
||||
"70000000-0000-4000-8000-000000000001"
|
||||
largeBytes largeDigest largePath []
|
||||
requireSucceeded "large streaming upload" uploaded
|
||||
downloaded <- curlToFile downloadPath
|
||||
[ "--cookie", sessionCookies session
|
||||
, probeBase context
|
||||
<> "/api/download?root=personal&path=streaming-large.bin"
|
||||
]
|
||||
assertEqual "large streaming download status" 200 (curlStatus downloaded)
|
||||
hashFile downloadPath >>= assertEqual "large download digest" largeDigest
|
||||
let tiny = "x"
|
||||
mapM_
|
||||
(\label -> do
|
||||
replay <- uploadBytesDeclared
|
||||
context session "streaming-large.bin"
|
||||
"70000000-0000-4000-8000-000000000007"
|
||||
largeBytes largeDigest tiny []
|
||||
requireSucceeded ("prior/desired observation " <> label) replay)
|
||||
["first", "replay"]
|
||||
sizeMismatch <- uploadBytesDeclared
|
||||
context session "size-mismatch.bin"
|
||||
"70000000-0000-4000-8000-000000000002"
|
||||
2 (sha256Hex tiny) tiny []
|
||||
requireError "size mismatch" 400 "size_mismatch" sizeMismatch
|
||||
identityMismatch <- uploadBytesDeclared
|
||||
context session "identity-mismatch.bin"
|
||||
"70000000-0000-4000-8000-000000000003"
|
||||
1 (BS.replicate 64 48) tiny []
|
||||
requireError "content identity mismatch"
|
||||
400 "content_identity_mismatch" identityMismatch
|
||||
tooLarge <- uploadFileDeclared
|
||||
context session "too-large.bin"
|
||||
"70000000-0000-4000-8000-000000000004"
|
||||
overLimit (sha256Hex tiny) largePath
|
||||
["--max-time", "5", "--limit-rate", "64k"]
|
||||
requireError "object size limit" 413 "object_too_large" tooLarge
|
||||
aborted <- interruptedUpload
|
||||
context session "aborted-upload.bin"
|
||||
"70000000-0000-4000-8000-000000000005"
|
||||
largeBytes largeDigest largePath
|
||||
assert "aborted upload unexpectedly completed" (not aborted)
|
||||
missing <- download context session "personal" "aborted-upload.bin"
|
||||
assertEqual "aborted upload visibility" 404 (curlStatus missing)
|
||||
replayAfterAbort <- uploadBytesDeclared
|
||||
context session "aborted-upload.bin"
|
||||
"70000000-0000-4000-8000-000000000005"
|
||||
largeBytes largeDigest tiny []
|
||||
assertEqual "aborted upload replay status" 200 (curlStatus replayAfterAbort)
|
||||
assert "aborted upload replay lost active or failed identity" $
|
||||
any (`BS.isInfixOf` curlBody replayAfterAbort)
|
||||
["\"result\":\"pending\"", "\"result\":\"failed\""]
|
||||
abortedOverwrite <- interruptedUpload
|
||||
context session "streaming-large.bin"
|
||||
"70000000-0000-4000-8000-000000000006"
|
||||
largeBytes (BS.replicate 64 48) largePath
|
||||
assert "aborted overwrite unexpectedly completed" (not abortedOverwrite)
|
||||
preservedPath <- pure (directory <> "/preserved.bin")
|
||||
preserved <- curlToFile preservedPath
|
||||
[ "--cookie", sessionCookies session
|
||||
, probeBase context
|
||||
<> "/api/download?root=personal&path=streaming-large.bin"
|
||||
]
|
||||
assertEqual "preserved object status" 200 (curlStatus preserved)
|
||||
hashFile preservedPath >>= assertEqual "preserved object digest" largeDigest
|
||||
let partialPath = directory <> "/partial-download.bin"
|
||||
partial <- interruptedDownload context session partialPath
|
||||
assert "aborted download unexpectedly completed" (not partial)
|
||||
partialBytes <- getFileSize partialPath
|
||||
assert "aborted download emitted no bytes" (partialBytes > 0)
|
||||
assert "aborted download emitted the complete object" $
|
||||
partialBytes < largeBytes
|
||||
ready <- curl [probeBase context <> "/readyz"]
|
||||
requireStatus "readiness after aborts" 200 ready
|
||||
assertContains "readiness after aborts" "\"ready\":true" (curlBody ready)
|
||||
putStrLn "Large streaming and abort behavior passed"
|
||||
|
||||
verifyThrottle :: ProbeContext -> Session -> FilePath -> IO ()
|
||||
verifyThrottle context session directory = do
|
||||
userB <- requireEnv "NUTHATCH_OIDC_USER_B"
|
||||
sessionB <- login context False userB "streaming-b"
|
||||
let jointPath = directory <> "/joint-upload.bin"
|
||||
throttlePath = directory <> "/throttle-upload.bin"
|
||||
writeZeros jointPath 65536
|
||||
jointDigest <- hashFile jointPath
|
||||
joint <- uploadFileDeclared
|
||||
context session "joint-throttle.bin"
|
||||
"70000000-0000-4000-8000-000000000013"
|
||||
65536 jointDigest jointPath []
|
||||
requireSucceeded "joint throttle setup" joint
|
||||
writeZeros throttlePath 1048576
|
||||
uploadDigest <- hashFile throttlePath
|
||||
throttled <- uploadFileDeclared
|
||||
context sessionB "throttled-upload.bin"
|
||||
"71000000-0000-4000-8000-000000000001"
|
||||
1048576 uploadDigest throttlePath []
|
||||
requireError "incremental upload throttle"
|
||||
429 "bandwidth_exceeded" throttled
|
||||
prestart <- uploadBytesDeclared
|
||||
context sessionB "prestart.bin"
|
||||
"71000000-0000-4000-8000-000000000002"
|
||||
1 uploadDigest "x" []
|
||||
assertEqual "pre-start upload throttle" 429 (curlStatus prestart)
|
||||
let firstPath = directory <> "/parallel-a"
|
||||
secondPath = directory <> "/parallel-b"
|
||||
downloadTo path =
|
||||
try (curlToFile path
|
||||
[ "--cookie", sessionCookies session
|
||||
, probeBase context
|
||||
<> "/api/download?root=personal&path=streaming-large.bin"
|
||||
]) :: IO (Either SomeException CurlResponse)
|
||||
(first, second) <- concurrently (downloadTo firstPath) (downloadTo secondPath)
|
||||
assert "first throttled download completed" (isLeft first)
|
||||
assert "second throttled download completed" (isLeft second)
|
||||
firstBytes <- getFileSize firstPath
|
||||
secondBytes <- getFileSize secondPath
|
||||
burst <- readEnvInteger "NUTHATCH_STREAMING_THROTTLE_BURST_BYTES"
|
||||
let moved = firstBytes + secondBytes
|
||||
jointLimit = burst - 65536 + 4096
|
||||
assert "parallel throttled downloads emitted no bytes" (moved > 0)
|
||||
assert "parallel downloads exceeded their joint token budget" $
|
||||
moved <= jointLimit
|
||||
exhausted <- download context session "personal" "streaming-large.bin"
|
||||
requireStatus "post-exhaustion download" 429 exhausted
|
||||
putStrLn "Shared incremental throttle behavior passed"
|
||||
|
||||
verifyRefill :: ProbeContext -> Session -> FilePath -> IO ()
|
||||
verifyRefill context session directory = do
|
||||
let bodyPath = directory <> "/refill-upload.bin"
|
||||
writeZeros bodyPath 65536
|
||||
digest <- hashFile bodyPath
|
||||
initial <- uploadFileDeclared
|
||||
context session "refill-initial.bin"
|
||||
"72000000-0000-4000-8000-000000000001"
|
||||
65536 digest bodyPath []
|
||||
requireSucceeded "initial token use" initial
|
||||
exhausted <- uploadFileDeclared
|
||||
context session "refill-exhausted.bin"
|
||||
"72000000-0000-4000-8000-000000000003"
|
||||
65536 digest bodyPath []
|
||||
requireStatus "exhausted token bucket" 429 exhausted
|
||||
threadDelay 2000000
|
||||
refilled <- uploadFileDeclared
|
||||
context session "refill-refilled.bin"
|
||||
"72000000-0000-4000-8000-000000000002"
|
||||
65536 digest bodyPath []
|
||||
requireSucceeded "refilled token bucket" refilled
|
||||
putStrLn "Token-bucket refill behavior passed"
|
||||
|
||||
verifyUnavailable :: ProbeContext -> Session -> IO ()
|
||||
verifyUnavailable context session = do
|
||||
listing <- browse context session "?root=personal"
|
||||
requireStatus "browse without Redis" 200 listing
|
||||
assertContains "browse without Redis" "\"path\":" (curlBody listing)
|
||||
let path =
|
||||
if "cross-user.txt" `BS.isInfixOf` curlBody listing
|
||||
then "cross-user.txt"
|
||||
else "streaming-large.bin"
|
||||
query =
|
||||
if path == "cross-user.txt" then "CROSS-USER" else "STREAMING-LARGE"
|
||||
searched <- browse context session ("?root=personal&q=" <> query)
|
||||
assertContains "search without Redis" (B8.pack path) (curlBody searched)
|
||||
unavailableUpload <- upload context session "personal" "unavailable.bin"
|
||||
"72000000-0000-4000-8000-000000000001" "x"
|
||||
requireError "upload without Redis"
|
||||
503 "throttle_unavailable" unavailableUpload
|
||||
unavailableDownload <-
|
||||
retryTransport 4 (download context session "personal" path)
|
||||
requireError "download without Redis"
|
||||
503 "throttle_unavailable" unavailableDownload
|
||||
putStrLn "Redis-unavailable transfer behavior passed"
|
||||
|
||||
retryTransport :: Int -> IO value -> IO value
|
||||
retryTransport attempts action = do
|
||||
result <- try action
|
||||
case result of
|
||||
Right value -> pure value
|
||||
Left exception
|
||||
| attempts > 1 -> do
|
||||
threadDelay 100000
|
||||
retryTransport (attempts - 1) action
|
||||
| otherwise -> fail (show (exception :: SomeException))
|
||||
|
||||
uploadFileDeclared
|
||||
:: ProbeContext
|
||||
-> Session
|
||||
-> String
|
||||
-> String
|
||||
-> Integer
|
||||
-> BS.ByteString
|
||||
-> FilePath
|
||||
-> [String]
|
||||
-> IO CurlResponse
|
||||
uploadFileDeclared context session path operationId size digest bodyPath extra =
|
||||
curl
|
||||
( extra
|
||||
<> uploadArguments context session path operationId size digest
|
||||
<> ["--data-binary", "@" <> bodyPath]
|
||||
)
|
||||
|
||||
uploadBytesDeclared
|
||||
:: ProbeContext
|
||||
-> Session
|
||||
-> String
|
||||
-> String
|
||||
-> Integer
|
||||
-> BS.ByteString
|
||||
-> BS.ByteString
|
||||
-> [String]
|
||||
-> IO CurlResponse
|
||||
uploadBytesDeclared context session path operationId size digest body extra =
|
||||
withSystemTempFile "nuthatch-streaming-body" $ \bodyPath handle -> do
|
||||
BS.hPut handle body
|
||||
hClose handle
|
||||
uploadFileDeclared
|
||||
context session path operationId size digest bodyPath extra
|
||||
|
||||
uploadArguments
|
||||
:: ProbeContext
|
||||
-> Session
|
||||
-> String
|
||||
-> String
|
||||
-> Integer
|
||||
-> BS.ByteString
|
||||
-> [String]
|
||||
uploadArguments context session path operationId size digest =
|
||||
[ "--cookie", sessionCookies session
|
||||
, "--header", "Content-Type: application/octet-stream"
|
||||
, "--header", "X-Declared-Size: " <> show size
|
||||
, "--header", "X-Content-SHA256: " <> B8.unpack digest
|
||||
, "--header", "X-CSRF-Token: " <> B8.unpack (sessionCsrf session)
|
||||
, probeBase context
|
||||
<> "/api/upload?root=personal&path="
|
||||
<> path
|
||||
<> "&operation_id="
|
||||
<> operationId
|
||||
]
|
||||
|
||||
interruptedUpload
|
||||
:: ProbeContext
|
||||
-> Session
|
||||
-> String
|
||||
-> String
|
||||
-> Integer
|
||||
-> BS.ByteString
|
||||
-> FilePath
|
||||
-> IO Bool
|
||||
interruptedUpload context session path operationId size digest bodyPath = do
|
||||
let arguments =
|
||||
[ "--insecure"
|
||||
, "--silent"
|
||||
, "--show-error"
|
||||
, "--max-time", "1"
|
||||
, "--limit-rate", "64k"
|
||||
, "--request", "POST"
|
||||
, "--upload-file", bodyPath
|
||||
]
|
||||
<> uploadArguments context session path operationId size digest
|
||||
(exitCode, _, _) <- readProcessWithExitCode "curl" arguments ""
|
||||
pure (exitCode == ExitSuccess)
|
||||
|
||||
interruptedDownload
|
||||
:: ProbeContext
|
||||
-> Session
|
||||
-> FilePath
|
||||
-> IO Bool
|
||||
interruptedDownload context session outputPath = do
|
||||
let arguments =
|
||||
[ "--insecure"
|
||||
, "--silent"
|
||||
, "--show-error"
|
||||
, "--max-time", "1"
|
||||
, "--limit-rate", "64k"
|
||||
, "--cookie", sessionCookies session
|
||||
, "--output", outputPath
|
||||
, probeBase context
|
||||
<> "/api/download?root=personal&path=streaming-large.bin"
|
||||
]
|
||||
(exitCode, _, _) <- readProcessWithExitCode "curl" arguments ""
|
||||
pure (exitCode == ExitSuccess)
|
||||
|
||||
writeZeros :: FilePath -> Integer -> IO ()
|
||||
writeZeros path byteCount =
|
||||
withBinaryFile path WriteMode $ \handle -> do
|
||||
let chunk = BS.replicate 1048576 0
|
||||
(fullChunks, remainder) = byteCount `divMod` 1048576
|
||||
replicateM_ (fromIntegral fullChunks) (BS.hPut handle chunk)
|
||||
unless (remainder == 0) $
|
||||
BS.hPut handle (BS.take (fromIntegral remainder) chunk)
|
||||
|
||||
hashFile :: FilePath -> IO BS.ByteString
|
||||
hashFile path =
|
||||
withBinaryFile path ReadMode $ \handle ->
|
||||
go handle hashInit
|
||||
where
|
||||
go handle context = do
|
||||
chunk <- BS.hGetSome handle 1048576
|
||||
if BS.null chunk
|
||||
then
|
||||
pure
|
||||
(B16.encode
|
||||
(BA.convert (hashFinalize context :: Digest SHA256)))
|
||||
else do
|
||||
let updated = hashUpdate context chunk
|
||||
updated `seq` go handle updated
|
||||
|
||||
readEnvInteger :: String -> IO Integer
|
||||
readEnvInteger name = do
|
||||
value <- requireEnv name
|
||||
maybe (fail ("invalid " <> name)) pure (readMaybe value)
|
||||
|
||||
requireSucceeded :: String -> CurlResponse -> IO ()
|
||||
requireSucceeded label response = do
|
||||
requireStatus label 200 response
|
||||
assertContains label "\"result\":\"succeeded\"" (curlBody response)
|
||||
|
||||
requireError :: String -> Int -> BS.ByteString -> CurlResponse -> IO ()
|
||||
requireError label status code response = do
|
||||
requireStatus label status response
|
||||
assertContains label ("\"error\":\"" <> code <> "\"") (curlBody response)
|
||||
|
||||
requireStatus :: String -> Int -> CurlResponse -> IO ()
|
||||
requireStatus label expected response =
|
||||
assertEqual (label <> " status") expected (curlStatus response)
|
||||
|
||||
isLeft :: Either left right -> Bool
|
||||
isLeft (Left _) = True
|
||||
isLeft (Right _) = False
|
||||
Loading…
Reference in a new issue