Ignore:
Timestamp:
8 Nov 2011, 18:22:43 (13 years ago)
Author:
uli
Message:

Remove old blob storage and fix related tests.

File:
1 edited

Legend:

Unmodified
Added
Removed
  • main/waeup.sirp/branches/ulif-extimgstore/src/waeup/sirp/tests/test_imagestorage.py

    r7032 r7034  
    1616from waeup.sirp.image import createWAeUPImageFile
    1717from waeup.sirp.imagestorage import (
    18     md5digest, Basket, ImageStorage, ImageStorageFileRetrieval,
    1918    FileStoreNameChooser, ExtFileStore, DefaultFileStoreHandler,
    2019    DefaultStorage)
     
    2928    def tearDown(self):
    3029        shutil.rmtree(self.workdir)
    31 
    32     def test_md5digest(self):
    33         samplefile = os.path.join(self.workdir, 'sample')
    34         open(samplefile, 'wb').write('blah')
    35         fp = open(samplefile, 'r')
    36         digest = md5digest(fp)
    37         self.assertEqual(digest, '6f1ed002ab5595859014ebf0951522d9')
    38 
    39 class BasketTests(FunctionalTestCase):
    40 
    41     layer = FunctionalLayer
    42 
    43     def setUp(self):
    44         super(BasketTests, self).setUp()
    45         self.workdir = tempfile.mkdtemp()
    46         self.samplefile = os.path.join(self.workdir, 'sample')
    47         self.otherfile = os.path.join(self.workdir, 'other')
    48         open(self.samplefile, 'wb').write('Hi there!')
    49         open(self.otherfile, 'wb').write('Hi from other!')
    50         self.basket = Basket()
    51         self.fd = open(self.samplefile, 'r')
    52         self.fd2 = open(self.otherfile, 'r')
    53         self.stderr = StringIO()
    54         self.old_stderr = sys.stderr
    55 
    56     def tearDown(self):
    57         sys.stderr = self.old_stderr
    58         super(BasketTests, self).tearDown()
    59         self.fd.close()
    60         shutil.rmtree(self.workdir)
    61         self.basket._del() # Remove subojects explicitly
    62         del self.basket
    63         return
    64 
    65     def test_ifaces(self):
    66         pass
    67 
    68     def test_curr_id_empty(self):
    69         curr_id = self.basket.curr_id
    70         self.assertEqual(curr_id, '1')
    71 
    72     def test_getInternalId_empty(self):
    73         basket_id = self.basket.getInternalId(self.fd)
    74         self.assertTrue(basket_id is None)
    75 
    76     def test_storeFile_single(self):
    77         basket_id = self.basket.storeFile(self.fd, 'sample')
    78         self.assertEqual(basket_id, '1')
    79         contents = self.basket['1'].open('r').read()
    80         self.assertEqual(contents, 'Hi there!')
    81 
    82     def test_storeFile_double(self):
    83         basket_id1 = self.basket.storeFile(self.fd, 'sample')
    84         basket_id2 = self.basket.storeFile(self.fd, 'sample')
    85         self.assertTrue(basket_id1 == basket_id2 == '1')
    86         contents = self.basket['1'].open('r').read()
    87         self.assertEqual(contents, 'Hi there!')
    88 
    89     def test_storeFile_multiple(self):
    90         basket_id1 = self.basket.storeFile(self.fd, 'sample')
    91         basket_id2 = self.basket.storeFile(self.fd2, 'sample')
    92         self.assertEqual(basket_id1, '1')
    93         self.assertEqual(basket_id2, '2')
    94         contents1 = self.basket['1'].open('r').read()
    95         contents2 = self.basket['2'].open('r').read()
    96         self.assertEqual(contents1, 'Hi there!')
    97         self.assertEqual(contents2, 'Hi from other!')
    98 
    99     def test_retrieveFile(self):
    100         basket_id = self.basket.storeFile(self.fd, 'sample')
    101         fd = self.basket.retrieveFile(basket_id)
    102         result = fd.read()
    103         self.assertEqual(result, 'Hi there!')
    104 
    105     def test_retrieveFile_not_existent(self):
    106         result = self.basket.retrieveFile('not-a-valid-basket-id')
    107         self.assertTrue(result is None)
    108 
    109     def test_detect_zero_length_blobs(self):
    110         # Ensure we get a warning when an empty Blob is found
    111         self.basket.storeFile(self.fd, 'sample')
    112         self.basket['1'].open('w').write('')
    113         self.fd.seek(0)
    114         sys.stderr = self.stderr         # Redirect stderr
    115         self.basket.storeFile(self.fd, 'sample')
    116         sys.stderr = self.old_stderr     # Restore stderr
    117         self.stderr.seek(0)
    118         self.assertTrue(
    119             "EMPTY BLOB DETECTED" in self.stderr.read())
    120 
    121     def test_refill_zero_length_blobs(self):
    122         # When we detect an empty Blob, it will be reused
    123         self.basket.storeFile(self.fd, 'sample')
    124         self.basket['1'].open('w').write('')
    125         self.fd.seek(0)
    126         sys.stderr = self.stderr         # Redirect stderr
    127         self.basket.storeFile(self.fd, 'sample')
    128         sys.stderr = self.old_stderr     # Restore stderr
    129         contents = self.basket['1'].open('r').read()
    130         self.assertEqual(contents, 'Hi there!')
    131 
    132 
    133 class ImageStorageTests(FunctionalTestCase):
    134 
    135     layer = FunctionalLayer
    136 
    137     def setUp(self):
    138         super(ImageStorageTests, self).setUp()
    139         self.workdir = tempfile.mkdtemp()
    140         self.samplefile = os.path.join(self.workdir, 'sample')
    141         self.otherfile = os.path.join(self.workdir, 'other')
    142         open(self.samplefile, 'wb').write('Hi there!')
    143         open(self.otherfile, 'wb').write('Hi from other!')
    144         self.storage = ImageStorage()
    145         self.fd = open(self.samplefile, 'r')
    146         self.fd2 = open(self.otherfile, 'r')
    147 
    148     def tearDown(self):
    149         super(ImageStorageTests, self).tearDown()
    150         self.fd.close()
    151         self.fd2.close()
    152         shutil.rmtree(self.workdir)
    153         self.storage._del() # Remove subojects explicitly
    154         del self.storage
    155         return
    156 
    157     def test_ifaces(self):
    158         pass
    159 
    160     def test_storeFile(self):
    161         full_id = self.storage.storeFile(self.fd, 'sample.txt')
    162         self.assertEqual(full_id, '396199333edbf40ad43e62a1c1397793-1')
    163 
    164     def test_storeFile_duplicate(self):
    165         full_id1 = self.storage.storeFile(self.fd, 'sample1.txt')
    166         full_id2 = self.storage.storeFile(self.fd, 'sample2.txt')
    167         full_id3 = self.storage.storeFile(self.fd, 'sample1.txt')
    168         self.assertEqual(full_id1, '396199333edbf40ad43e62a1c1397793-1')
    169         self.assertEqual(full_id2, '396199333edbf40ad43e62a1c1397793-1')
    170         self.assertEqual(full_id3, '396199333edbf40ad43e62a1c1397793-1')
    171         contents = self.storage.retrieveFile(
    172             '396199333edbf40ad43e62a1c1397793-1').read()
    173         self.assertEqual(contents, 'Hi there!')
    174 
    175     def test_storeFile_multiple(self):
    176         full_id1 = self.storage.storeFile(self.fd, 'sample1.txt')
    177         full_id2 = self.storage.storeFile(self.fd2, 'sample1.txt')
    178         self.assertEqual(full_id1, '396199333edbf40ad43e62a1c1397793-1')
    179         self.assertEqual(full_id2, '6936fcf8d564f1c5be5a017e650c5e8f-1')
    180 
    181     def test_retrieveFile_not_existent(self):
    182         result = self.storage.retrieveFile('not-existent')
    183         self.assertTrue(result is None)
    184 
    185     def test_retrieveFile_illegal_marker(self):
    186         result1 = self.storage.retrieveFile('really-not-existent')
    187         result2 = self.storage.retrieveFile('notexistent')
    188         self.assertTrue(result1 is result2 is None)
    189 
    190     def test_retrieveFile(self):
    191         full_id = self.storage.storeFile(self.fd, 'sample.txt')
    192         result = self.storage.retrieveFile(full_id)
    193         content = result.read()
    194         self.assertEqual(content, 'Hi there!')
    195 
    196     def test_retrieveFile_multiple(self):
    197         full_id1 = self.storage.storeFile(self.fd, 'sample.txt')
    198         full_id2 = self.storage.storeFile(self.fd2, 'other.txt')
    199         result1 = self.storage.retrieveFile(full_id1)
    200         result2 = self.storage.retrieveFile(full_id2)
    201         content1 = result1.read()
    202         content2 = result2.read()
    203         self.assertEqual(content1, 'Hi there!')
    204         self.assertEqual(content2, 'Hi from other!')
    205 
    206 class ImageStorageFileRetrievalTests(FunctionalTestCase):
    207 
    208     layer = FunctionalLayer
    209 
    210     def setUp(self):
    211         super(ImageStorageFileRetrievalTests, self).setUp()
    212         self.workdir = tempfile.mkdtemp()
    213         self.samplefile = os.path.join(self.workdir, 'sample')
    214         self.otherfile = os.path.join(self.workdir, 'other')
    215         open(self.samplefile, 'wb').write('Hi there!')
    216         open(self.otherfile, 'wb').write('Hi from other!')
    217         self.storage = ImageStorage()
    218         self.fd = open(self.samplefile, 'r')
    219         self.fd2 = open(self.otherfile, 'r')
    220         # Set up a single image storage in a site
    221         self.getRootFolder()['app'] = University()
    222         self.app = self.getRootFolder()['app']
    223         if not 'images' in self.app.keys():
    224             self.app['images'] = ImageStorage()
    225         self.storage = self.app['images']
    226         return
    227 
    228 
    229     def tearDown(self):
    230         super(ImageStorageFileRetrievalTests, self).tearDown()
    231         self.fd.close()
    232         self.fd2.close()
    233         shutil.rmtree(self.workdir)
    234         self.storage._del() # Remove subojects explicitly
    235         return
    236 
    237     def test_ifaces(self):
    238         retrieval = ImageStorageFileRetrieval()
    239         assert verifyClass(IFileRetrieval, ImageStorageFileRetrieval)
    240         assert verifyObject(IFileRetrieval, retrieval)
    241         return
    242 
    243     def test_getImageStorage_nosite(self):
    244         retrieval = ImageStorageFileRetrieval()
    245         storage = retrieval.getImageStorage()
    246         self.assertTrue(storage is None)
    247         return
    248 
    249     def test_getImageStorage(self):
    250         setSite(self.app)
    251         retrieval = ImageStorageFileRetrieval()
    252         storage = retrieval.getImageStorage()
    253         self.assertTrue(storage is self.storage)
    254         return
    255 
    256     def test_isImageStorageEnabled_nosite(self):
    257         retrieval = ImageStorageFileRetrieval()
    258         self.assertTrue(retrieval.isImageStorageEnabled() is False)
    259         return
    260 
    261     def test_isImageStorageEnabled(self):
    262         setSite(self.app)
    263         retrieval = ImageStorageFileRetrieval()
    264         self.assertTrue(retrieval.isImageStorageEnabled() is True)
    265         return
    266 
    267     def test_getFile_nosite(self):
    268         retrieval = ImageStorageFileRetrieval()
    269         f = retrieval.getFile('Hi there!')
    270         self.assertEqual(f.read(), 'Hi there!')
    271         return
    272 
    273     def test_getFile(self):
    274         setSite(self.app)
    275         retrieval = ImageStorageFileRetrieval()
    276         waeup_image = retrieval.createFile('sample.txt', self.fd)
    277         full_id = waeup_image.data
    278         result = retrieval.getFile(full_id)
    279         self.assertEqual(result.read(), 'Hi there!')
    280         return
    281 
    282     def test_createFile_nosite(self):
    283         retrieval = ImageStorageFileRetrieval()
    284         waeup_image = retrieval.createFile('sample.txt', self.fd)
    285         self.assertEqual(waeup_image.data, 'Hi there!')
    286         return
    287 
    288     def test_createFile(self):
    289         # Ensure we can create WAeUPImageFiles when in site
    290         setSite(self.app)
    291         retrieval = ImageStorageFileRetrieval()
    292         waeup_image = retrieval.createFile('sample.txt', self.fd)
    293         full_id = waeup_image.data
    294         self.assertEqual(full_id, '396199333edbf40ad43e62a1c1397793-1')
    295         return
    296 
    297     def DISABLEDtest_waeupimagefile(self):
    298         # Make sure WAeUPImageFile can use our file retrieval
    299         setSite(self.app)
    300         myfile = createWAeUPImageFile('sample.jpg', self.fd)
    301         contents = myfile.file.read()
    302         self.assertEqual(contents, 'Hi there!')
    303 
    304     def DISABLEDtest_waeupimagefile_raw(self):
    305         # Make sure we can retrieve a file also if it was initialized
    306         # with no image storage available
    307         myfile = createWAeUPImageFile('sample.jpg', self.fd)
    308         setSite(self.app)
    309         contents = myfile.file.read()
    310         self.assertEqual(contents, 'Hi there!')
    31130
    31231class FileStoreNameChooserTests(FunctionalTestCase):
     
    429148class CustomizedFileHandler(object):
    430149    def pathFromFileID(self, store, root, file_id):
    431         """Turn filename into path to store.
    432         """
    433150        return os.path.join(root, file_id[12:])
    434151
     
    440157
    441158    layer = FunctionalLayer
    442 
    443159
    444160    def setUp(self):
     
    449165        open(self.samplefile, 'wb').write('Hi there!')
    450166        open(self.otherfile, 'wb').write('Hi from other!')
    451         self.storage = ImageStorage()
    452167        self.fd = open(self.samplefile, 'r')
    453168        self.fd2 = open(self.otherfile, 'r')
Note: See TracChangeset for help on using the changeset viewer.