<s> import sys <EOL> import os <EOL> import os . path <EOL> import hashlib <EOL> import struct <EOL> import tempfile <EOL> import shutil <EOL> import pickle <EOL> import unittest <EOL> this_dir = os . path . dirname ( os . path . abspath ( __file__ ) ) <EOL> sys . path . append ( os . path . dirname ( this_dir ) ) <EOL> from paxos import durable <EOL> class DObj ( object ) : <EOL> def __init__ ( self ) : <EOL> self . state = '<STR_LIT>' <EOL> class DurableReadTester ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> tmpfs_dir = '<STR_LIT>' if os . path . exists ( '<STR_LIT>' ) else None <EOL> self . tdir = tempfile . mkdtemp ( dir = tmpfs_dir ) <EOL> self . fds = list ( ) <EOL> def tearDown ( self ) : <EOL> shutil . rmtree ( self . tdir ) <EOL> for fd in self . fds : <EOL> os . close ( fd ) <EOL> def newfd ( self , data = None ) : <EOL> bin_flag = <NUM_LIT:0> if not hasattr ( os , '<STR_LIT>' ) else os . O_BINARY <EOL> fd = os . open ( os . path . join ( self . tdir , str ( len ( self . fds ) ) ) , os . O_CREAT | os . O_RDWR | bin_flag ) <EOL> self . fds . append ( fd ) <EOL> if data is not None : <EOL> os . write ( fd , data ) <EOL> return fd <EOL> def test_read_zero_length ( self ) : <EOL> self . assertRaises ( durable . FileTruncated , durable . read , self . newfd ( ) ) <EOL> def test_read_header_too_small ( self ) : <EOL> self . assertRaises ( durable . FileTruncated , durable . read , self . newfd ( '<STR_LIT>' * <NUM_LIT> ) ) <EOL> def test_read_no_pickle_data ( self ) : <EOL> data = '<STR_LIT>' * <NUM_LIT> + struct . pack ( '<STR_LIT>' , <NUM_LIT:5> ) <EOL> self . assertRaises ( durable . FileTruncated , durable . read , self . newfd ( data ) ) <EOL> def test_read_bad_hash_mismatch ( self ) : <EOL> data = '<STR_LIT>' * <NUM_LIT> + struct . pack ( '<STR_LIT>' , <NUM_LIT:5> ) + '<STR_LIT:x>' * <NUM_LIT:5> <EOL> self . assertRaises ( durable . HashMismatch , durable . read , self . newfd ( data ) ) <EOL> def test_read_ok ( self ) : <EOL> pdata = '<STR_LIT:x>' * <NUM_LIT:5> <EOL> p = pickle . dumps ( pdata , pickle . HIGHEST_PROTOCOL ) <EOL> data = '<STR_LIT>' * <NUM_LIT:8> + struct . pack ( '<STR_LIT>' , len ( p ) ) + p <EOL> data = hashlib . md5 ( data ) . digest ( ) + data <EOL> self . assertEqual ( durable . read ( self . newfd ( data ) ) , ( <NUM_LIT:0> , pdata ) ) <EOL> class DurableObjectHandlerTester ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> tmpfs_dir = '<STR_LIT>' if os . path . exists ( '<STR_LIT>' ) else None <EOL> self . o = DObj ( ) <EOL> self . tdir = tempfile . mkdtemp ( dir = tmpfs_dir ) <EOL> self . doh = durable . DurableObjectHandler ( self . tdir , '<STR_LIT>' ) <EOL> self . dohs = [ self . doh , ] <EOL> def tearDown ( self ) : <EOL> for doh in self . dohs : <EOL> doh . close ( ) <EOL> shutil . rmtree ( self . tdir ) <EOL> def newdoh ( self , obj_id = None ) : <EOL> if obj_id is None : <EOL> obj_id = '<STR_LIT:id>' + str ( len ( self . dohs ) ) <EOL> doh = durable . DurableObjectHandler ( self . tdir , obj_id ) <EOL> self . dohs . append ( doh ) <EOL> return doh <EOL> def test_bad_directory ( self ) : <EOL> self . assertRaises ( Exception , durable . DurableObjectHandler , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_no_save ( self ) : <EOL> self . doh . close ( ) <EOL> d = self . newdoh ( '<STR_LIT>' ) <EOL> self . assertEquals ( d . recovered , None ) <EOL> self . assertEquals ( d . serial , <NUM_LIT:1> ) <EOL> def test_one_save ( self ) : <EOL> self . doh . save ( self . o ) <EOL> self . doh . close ( ) <EOL> d = self . newdoh ( '<STR_LIT>' ) <EOL> self . assertTrue ( os . stat ( self . doh . fn_a ) . st_size > <NUM_LIT:0> ) <EOL> self . assertTrue ( os . stat ( self . doh . fn_b ) . st_size == <NUM_LIT:0> ) <EOL> self . assertTrue ( isinstance ( d . recovered , DObj ) ) <EOL> self . assertEquals ( d . recovered . state , '<STR_LIT>' ) <EOL> def test_two_save ( self ) : <EOL> self . doh . save ( self . o ) <EOL> self . o . state = '<STR_LIT>' <EOL> self . doh . save ( self . o ) <EOL> self . doh . close ( ) <EOL> d = self . newdoh ( '<STR_LIT>' ) <EOL> self . assertTrue ( os . stat ( self . doh . fn_a ) . st_size > <NUM_LIT:0> ) <EOL> self . assertTrue ( os . stat ( self . doh . fn_b ) . st_size > <NUM_LIT:0> ) <EOL> self . assertTrue ( isinstance ( d . recovered , DObj ) ) <EOL> self . assertEquals ( d . recovered . state , '<STR_LIT>' ) <EOL> def test_three_save ( self ) : <EOL> self . doh . save ( self . o ) <EOL> self . o . state = '<STR_LIT>' <EOL> self . doh . save ( self . o ) <EOL> self . o . state = '<STR_LIT>' <EOL> self . doh . save ( self . o ) <EOL> self . doh . close ( ) <EOL> d = self . newdoh ( '<STR_LIT>' ) <EOL> self . assertTrue ( isinstance ( d . recovered , DObj ) ) <EOL> self . assertEquals ( d . recovered . state , '<STR_LIT>' ) <EOL> def test_new_object_corrupted ( self ) : <EOL> self . test_two_save ( ) <EOL> with open ( self . doh . fn_b , '<STR_LIT:wb>' ) as f : <EOL> f . write ( '<STR_LIT>' ) <EOL> f . flush ( ) <EOL> d = self . newdoh ( '<STR_LIT>' ) <EOL> self . assertTrue ( isinstance ( d . recovered , DObj ) ) <EOL> self . assertEquals ( d . recovered . state , '<STR_LIT>' ) <EOL> def test_old_object_corrupted ( self ) : <EOL> self . test_two_save ( ) <EOL> with open ( self . doh . fn_a , '<STR_LIT:wb>' ) as f : <EOL> f . write ( '<STR_LIT>' ) <EOL> f . flush ( ) <EOL> d = self . newdoh ( '<STR_LIT>' ) <EOL> self . assertTrue ( isinstance ( d . recovered , DObj ) ) <EOL> self . assertEquals ( d . recovered . state , '<STR_LIT>' ) <EOL> def test_unrecoverable_corruption ( self ) : <EOL> self . test_two_save ( ) <EOL> with open ( self . doh . fn_a , '<STR_LIT:wb>' ) as f : <EOL> f . write ( '<STR_LIT>' ) <EOL> f . flush ( ) <EOL> with open ( self . doh . fn_b , '<STR_LIT:wb>' ) as f : <EOL> f . write ( '<STR_LIT>' ) <EOL> f . flush ( ) <EOL> def diehorribly ( ) : <EOL> self . newdoh ( '<STR_LIT>' ) <EOL> self . assertRaises ( durable . UnrecoverableFailure , diehorribly ) </s>
