IOByteStreamTests.hs (8576B)
1 {-# LANGUAGE ScopedTypeVariables #-} 2 module IOByteStreamTests (ioByteStreamTests) where 3 4 import qualified IO.ByteStream as B 5 import qualified IO.Resources as R 6 import Control.Concurrent 7 import Control.Exception 8 import Control.Monad (forM_) 9 import Crypto.Hash (hash, Digest, SHA256) 10 import qualified Data.ByteString as BS 11 import Data.IORef 12 import System.Directory (listDirectory) 13 import System.FilePath ((</>)) 14 import System.IO (hClose) 15 import System.IO.Temp (withSystemTempDirectory) 16 import Test.Tasty 17 import Test.Tasty.HUnit 18 19 sha :: BS.ByteString -> String 20 sha bytes = show (hash bytes :: Digest SHA256) 21 category wanted value = case value of 22 Left p -> R.category p @?= wanted 23 Right _ -> assertFailure ("expected " ++ wanted) 24 reader runtime bytes = do 25 remaining <- newIORef bytes 26 Right token <- R.acquire runtime R.rootScope (pure (R.Reader (\n -> 27 atomicModifyIORef' remaining (\rest -> (BS.drop n rest, BS.take n rest))) (pure ()))) 28 pure token 29 30 ioByteStreamTests = localOption (mkTimeout 30000000) $ testGroup "IO native byte streams" 31 [ testCase "authenticated binary snapshots and partial writes across chunk boundaries" $ 32 withSystemTempDirectory "tricu-bytes" $ \directory -> forM_ [0,1,4095,4096,4097,917828,1048576] $ \size -> do 33 let bytes = BS.pack (take size (cycle [0..255])) 34 runtime <- R.newRuntime 35 input <- reader runtime bytes 36 Right (snapshot,count) <- B.snapshot runtime R.rootScope input directory (toInteger size) (sha bytes) 37 count @?= toInteger size 38 listDirectory directory >>= (@?= []) 39 chunks <- newIORef [] 40 Right output <- R.acquire runtime R.rootScope (pure (R.Writer (\chunk -> do 41 let part = BS.take 997 chunk 42 modifyIORef' chunks (part:) 43 pure (BS.length part)) (pure ()))) 44 B.transfer runtime R.rootScope snapshot output count >>= (@?= Right ()) 45 BS.concat . reverse <$> readIORef chunks >>= (@?= bytes) 46 R.readSome runtime R.rootScope snapshot 1 >>= (@?= Right BS.empty) 47 R.seekFile runtime R.rootScope snapshot 0 >>= (@?= Right ()) 48 R.fileSize runtime R.rootScope snapshot >>= (@?= Right count) 49 R.writeSome runtime R.rootScope snapshot "x" >>= category "wrong-kind" 50 R.resizeFile runtime R.rootScope snapshot 0 >>= category "unsupported" 51 R.closeScope runtime R.rootScope >>= (@?= []) 52 , testCase "digest and limit failures remove the spool; EOF at the limit is required" $ 53 withSystemTempDirectory "tricu-bytes" $ \directory -> do 54 runtime <- R.newRuntime 55 forM_ [("abc",3,sha "xyz","integrity-failed"),("abcd",3,sha "abc","invalid-bound"),("",0,sha "x","integrity-failed")] $ \(bytes,limit,digest,wanted) -> do 56 input <- reader runtime bytes 57 counts <- R.liveCounts runtime 58 B.snapshot runtime R.rootScope input directory limit digest >>= category wanted 59 R.liveCounts runtime >>= (@?= counts) 60 listDirectory directory >>= (@?= []) 61 input <- reader runtime "abc" 62 B.snapshot runtime R.rootScope input directory (-1) (sha "abc") >>= category "invalid-bound" 63 B.snapshot runtime R.rootScope input directory 3 (replicate 64 'Z') >>= category "invalid-argument" 64 R.readSome runtime R.rootScope input 3 >>= (@?= Right "abc") 65 R.closeScope runtime R.rootScope >>= (@?= []) 66 , testCase "source mutation after snapshot cannot change transferred bytes" $ 67 withSystemTempDirectory "tricu-bytes" $ \directory -> do 68 runtime <- R.newRuntime 69 -- Injectable source bytes model in-place changes, independent of host 70 -- Handle locking of a simultaneously opened file. 71 source <- newIORef "original" 72 cursor <- newIORef 0 73 Right input <- R.acquire runtime R.rootScope (pure (R.Reader (\n -> do 74 pos <- readIORef cursor 75 bytes <- BS.take n . BS.drop pos <$> readIORef source 76 modifyIORef' cursor (+ BS.length bytes) 77 pure bytes) (pure ()))) 78 Right (snapshot,count) <- B.snapshot runtime R.rootScope input directory 100 (sha "original") 79 writeIORef source "modified" 80 Right output <- R.openWriter runtime R.rootScope (directory </> "output") 81 B.transfer runtime R.rootScope snapshot output count >>= (@?= Right ()) 82 R.closeScope runtime R.rootScope >>= (@?= []) 83 BS.readFile (directory </> "output") >>= (@?= "original") 84 , testCase "mutation during copying fails authentication" $ 85 withSystemTempDirectory "tricu-bytes" $ \directory -> do 86 runtime <- R.newRuntime 87 let original = BS.replicate 8192 65 88 calls <- newIORef (0 :: Int) 89 Right input <- R.acquire runtime R.rootScope (pure (R.Reader (\n -> do 90 call <- atomicModifyIORef' calls (\x -> (x+1,x)) 91 pure (if call < 2 then BS.replicate n (if call == 0 then 65 else 66) else BS.empty)) (pure ()))) 92 B.snapshot runtime R.rootScope input directory 8192 (sha original) >>= category "integrity-failed" 93 listDirectory directory >>= (@?= []) 94 R.closeScope runtime R.rootScope >>= (@?= []) 95 , testCase "snapshot cancellation keeps scoped cleanup and releases source lease" $ 96 withSystemTempDirectory "tricu-bytes" $ \directory -> do 97 runtime <- R.newRuntime 98 entered <- newEmptyMVar; block <- newEmptyMVar; done <- newEmptyMVar 99 Right input <- R.acquire runtime R.rootScope (pure (R.Reader (\_ -> putMVar entered () >> takeMVar block) (pure ()))) 100 released <- newIORef (0 :: Int) 101 worker <- forkIO $ (B.snapshotWith (\h -> modifyIORef' released (+1) >> hClose h) 102 runtime R.rootScope input directory 4096 (sha "") >> pure ()) 103 `catch` (\(_ :: AsyncException) -> pure ()) `finally` putMVar done () 104 takeMVar entered 105 listDirectory directory >>= (@?= []) 106 killThread worker; takeMVar done 107 R.closeScope runtime R.rootScope >>= (@?= []) 108 readIORef released >>= (@?= 1) 109 , testCase "cleanup failure preserves primary integrity diagnostic" $ 110 withSystemTempDirectory "tricu-bytes" $ \directory -> do 111 runtime <- R.newRuntime 112 input <- reader runtime "bad" 113 result <- B.snapshotWith (\h -> hClose h >> ioError (userError "injected close")) 114 runtime R.rootScope input directory 3 (sha "yes") 115 category "integrity-failed" result 116 case result of Left p -> assertBool "cleanup recorded" (length (R.detail p) > length ("SHA-256 mismatch" :: String)); _ -> pure () 117 R.closeScope runtime R.rootScope >>= (@?= []) 118 , testCase "transfer failure and cancellation release leases without retrying" $ 119 withSystemTempDirectory "tricu-bytes" $ \directory -> do 120 runtime <- R.newRuntime 121 input <- reader runtime "abcdef" 122 writes <- newIORef (0 :: Int) 123 Right broken <- R.acquire runtime R.rootScope (pure (R.Writer (\_ -> do 124 modifyIORef' writes (+1) 125 ioError (userError "peer disconnected")) (pure ()))) 126 B.transfer runtime R.rootScope input broken 6 >>= category "io-error" 127 readIORef writes >>= (@?= 1) 128 R.closeResource runtime R.rootScope broken >>= (@?= Right ()) 129 second <- reader runtime "abcdef" 130 entered <- newEmptyMVar; block <- newEmptyMVar; done <- newEmptyMVar 131 Right blocked <- R.acquire runtime R.rootScope (pure (R.Writer (\_ -> putMVar entered () >> takeMVar block) (pure ()))) 132 worker <- forkIO $ (B.transfer runtime R.rootScope second blocked 6 >> pure ()) 133 `catch` (\(_ :: AsyncException) -> pure ()) `finally` putMVar done () 134 takeMVar entered 135 R.readSome runtime R.rootScope second 1 >>= category "busy" 136 R.writeSome runtime R.rootScope blocked "x" >>= category "busy" 137 killThread worker; takeMVar done 138 R.closeResource runtime R.rootScope second >>= (@?= Right ()) 139 R.closeResource runtime R.rootScope blocked >>= (@?= Right ()) 140 R.closeScope runtime R.rootScope >>= (@?= []) 141 , testCase "transfer rejects short sources and wrong-kind targets" $ do 142 runtime <- R.newRuntime 143 input <- reader runtime "x" 144 B.transfer runtime R.rootScope input input 1 >>= category "wrong-kind" 145 chunks <- newIORef [] 146 Right output <- R.acquire runtime R.rootScope (pure (R.Writer (\bytes -> modifyIORef' chunks (bytes:) >> pure (BS.length bytes)) (pure ()))) 147 B.transfer runtime R.rootScope input output 2 >>= category "invalid-data" 148 readIORef chunks >>= (@?= ["x"]) 149 R.closeScope runtime R.rootScope >>= (@?= []) 150 ]