| 1 | import os |
|---|
| 2 | import tempfile |
|---|
| 3 | import shutil |
|---|
| 4 | import unittest |
|---|
| 5 | from StringIO import StringIO |
|---|
| 6 | from hurry.file import HurryFile |
|---|
| 7 | from hurry.file.interfaces import IFileRetrieval |
|---|
| 8 | from zope.component import ( |
|---|
| 9 | getUtility, provideUtility, queryUtility, provideAdapter) |
|---|
| 10 | from zope.component.hooks import setSite |
|---|
| 11 | from zope.interface.verify import verifyClass, verifyObject |
|---|
| 12 | from waeup.sirp.app import University |
|---|
| 13 | from waeup.sirp.testing import FunctionalLayer, FunctionalTestCase |
|---|
| 14 | from waeup.sirp.imagestorage import ( |
|---|
| 15 | FileStoreNameChooser, ExtFileStore, DefaultFileStoreHandler, |
|---|
| 16 | DefaultStorage) |
|---|
| 17 | from waeup.sirp.interfaces import ( |
|---|
| 18 | IFileStoreNameChooser, IExtFileStore, IFileStoreHandler,) |
|---|
| 19 | |
|---|
| 20 | class HelperFuncsTests(unittest.TestCase): |
|---|
| 21 | |
|---|
| 22 | def setUp(self): |
|---|
| 23 | self.workdir = tempfile.mkdtemp() |
|---|
| 24 | |
|---|
| 25 | def tearDown(self): |
|---|
| 26 | shutil.rmtree(self.workdir) |
|---|
| 27 | |
|---|
| 28 | class FileStoreNameChooserTests(FunctionalTestCase): |
|---|
| 29 | |
|---|
| 30 | layer = FunctionalLayer |
|---|
| 31 | |
|---|
| 32 | def test_iface(self): |
|---|
| 33 | # we provide the interfaces we promise to do |
|---|
| 34 | obj = FileStoreNameChooser(None) |
|---|
| 35 | verifyClass(IFileStoreNameChooser, FileStoreNameChooser) |
|---|
| 36 | verifyObject(IFileStoreNameChooser, obj) |
|---|
| 37 | return |
|---|
| 38 | |
|---|
| 39 | def test_accessible_as_adapter(self): |
|---|
| 40 | # we can get a file name chooser via adapter |
|---|
| 41 | chooser = IFileStoreNameChooser(object()) |
|---|
| 42 | self.assertTrue( |
|---|
| 43 | isinstance(chooser, FileStoreNameChooser)) |
|---|
| 44 | return |
|---|
| 45 | |
|---|
| 46 | def test_check_name(self): |
|---|
| 47 | # default file name choosers accept any string |
|---|
| 48 | chooser = FileStoreNameChooser(object()) |
|---|
| 49 | self.assertEqual(chooser.checkName('Hi there!'), True) |
|---|
| 50 | self.assertEqual(chooser.checkName(None), False) |
|---|
| 51 | return |
|---|
| 52 | |
|---|
| 53 | def test_choose_name(self): |
|---|
| 54 | # we get a simple string if we do not pass in a valid string |
|---|
| 55 | chooser = FileStoreNameChooser(object()) |
|---|
| 56 | self.assertEqual(chooser.chooseName('myname'), 'myname') |
|---|
| 57 | self.assertEqual(chooser.chooseName(None), u'unknown_file') |
|---|
| 58 | return |
|---|
| 59 | |
|---|
| 60 | class ExtFileStoreTests(unittest.TestCase): |
|---|
| 61 | # Test external file store (non-functional mode) |
|---|
| 62 | |
|---|
| 63 | def setUp(self): |
|---|
| 64 | self.workdir = tempfile.mkdtemp() |
|---|
| 65 | self.root = None |
|---|
| 66 | return |
|---|
| 67 | |
|---|
| 68 | def tearDown(self): |
|---|
| 69 | shutil.rmtree(self.workdir) |
|---|
| 70 | if self.root is not None: |
|---|
| 71 | shutil.rmtree(self.root) |
|---|
| 72 | return |
|---|
| 73 | |
|---|
| 74 | def test_iface(self): |
|---|
| 75 | obj = ExtFileStore(None) |
|---|
| 76 | verifyClass(IExtFileStore, ExtFileStore) |
|---|
| 77 | verifyObject(IExtFileStore, obj) |
|---|
| 78 | return |
|---|
| 79 | |
|---|
| 80 | def test_root_setup_wo_site(self): |
|---|
| 81 | # if no site is available we can use a temporary root |
|---|
| 82 | fs = ExtFileStore() |
|---|
| 83 | self.root = fs.root |
|---|
| 84 | self.assertTrue(isinstance(self.root, basestring)) |
|---|
| 85 | self.assertTrue(os.path.exists(self.root)) |
|---|
| 86 | return |
|---|
| 87 | |
|---|
| 88 | def test_create_instance(self): |
|---|
| 89 | storage1 = ExtFileStore() |
|---|
| 90 | storage2 = ExtFileStore(root=self.workdir) |
|---|
| 91 | self.root = storage1.root |
|---|
| 92 | self.assertTrue(storage1.root is not None) |
|---|
| 93 | self.assertTrue(storage1.root != storage2.root) |
|---|
| 94 | self.assertEqual(storage2.root, self.workdir) |
|---|
| 95 | return |
|---|
| 96 | |
|---|
| 97 | def test_create_file(self): |
|---|
| 98 | # We can store files |
|---|
| 99 | storage = ExtFileStore(root=self.workdir) |
|---|
| 100 | dummy_file = StringIO('sample file') |
|---|
| 101 | image_file = storage.createFile('mysample', dummy_file) |
|---|
| 102 | self.assertTrue('mysample' in os.listdir(storage.root)) |
|---|
| 103 | self.assertEqual('mysample', image_file.data) |
|---|
| 104 | return |
|---|
| 105 | |
|---|
| 106 | def test_get_file(self): |
|---|
| 107 | # We can get files after having them stored |
|---|
| 108 | storage = ExtFileStore(root=self.workdir) |
|---|
| 109 | dummy_file = StringIO('sample file') |
|---|
| 110 | image_file = storage.createFile('mysample', dummy_file) |
|---|
| 111 | result = storage.getFile(image_file.data) |
|---|
| 112 | self.assertEqual(result.read(), 'sample file') |
|---|
| 113 | return |
|---|
| 114 | |
|---|
| 115 | def test_extract_marker(self): |
|---|
| 116 | # file stores support extracting markers from filenames |
|---|
| 117 | storage = ExtFileStore(root=self.workdir) |
|---|
| 118 | result1 = storage.extractMarker(None) |
|---|
| 119 | result2 = storage.extractMarker('') |
|---|
| 120 | result3 = storage.extractMarker('no-marker') |
|---|
| 121 | result4 = storage.extractMarker('no-marker.txt') |
|---|
| 122 | result5 = storage.extractMarker('__MARKER__foo.jpg') |
|---|
| 123 | result6 = storage.extractMarker('__MaRkEr__foo.jpg') |
|---|
| 124 | result7 = storage.extractMarker('__THE_MARKER__foo.jpg') |
|---|
| 125 | result8 = storage.extractMarker('__A_MARK__my__foo.jpg') |
|---|
| 126 | |
|---|
| 127 | self.assertEqual(result1, ('', '', '', '')) |
|---|
| 128 | self.assertEqual(result2, ('', '', '', '')) |
|---|
| 129 | self.assertEqual(result3, ('', 'no-marker', 'no-marker', '')) |
|---|
| 130 | self.assertEqual(result4, ('', 'no-marker.txt', 'no-marker', '.txt')) |
|---|
| 131 | self.assertEqual(result5, ('marker', 'foo.jpg', 'foo', '.jpg')) |
|---|
| 132 | self.assertEqual(result6, ('marker', 'foo.jpg', 'foo', '.jpg')) |
|---|
| 133 | self.assertEqual(result7, ('the_marker', 'foo.jpg', 'foo', '.jpg')) |
|---|
| 134 | self.assertEqual(result8, ('a_mark', 'my__foo.jpg', 'my__foo', '.jpg')) |
|---|
| 135 | return |
|---|
| 136 | |
|---|
| 137 | class DefaultFileStoreHandlerTests(unittest.TestCase): |
|---|
| 138 | |
|---|
| 139 | def test_iface(self): |
|---|
| 140 | obj = DefaultFileStoreHandler() |
|---|
| 141 | verifyClass(IFileStoreHandler, DefaultFileStoreHandler) |
|---|
| 142 | verifyObject(IFileStoreHandler, obj) |
|---|
| 143 | return |
|---|
| 144 | |
|---|
| 145 | class CustomizedFileHandler(object): |
|---|
| 146 | def pathFromFileID(self, store, root, file_id): |
|---|
| 147 | return os.path.join(root, file_id[12:]) |
|---|
| 148 | |
|---|
| 149 | def createFile(self, store, root, file_id, filename, f): |
|---|
| 150 | path = self.pathFromFileID(store, root, file_id) |
|---|
| 151 | return f, path, HurryFile(filename, file_id) |
|---|
| 152 | |
|---|
| 153 | class CustomContext(object): |
|---|
| 154 | pass |
|---|
| 155 | |
|---|
| 156 | class CustomContextFileChooser(object): |
|---|
| 157 | def __init__(self, context): |
|---|
| 158 | self.context = context |
|---|
| 159 | |
|---|
| 160 | def chooseName(self, name=None, attr=None): |
|---|
| 161 | # this name chooser returns different file ids depending on |
|---|
| 162 | # the `attr` parameter, a simple string. |
|---|
| 163 | if attr=='img': |
|---|
| 164 | return '__mymarker__mysample.jpg' |
|---|
| 165 | elif attr=='doc': |
|---|
| 166 | return '__mymarker__mysample.doc' |
|---|
| 167 | return '__mymarker__mysample.txt' |
|---|
| 168 | |
|---|
| 169 | class FunctionalExtFileStoreTests(FunctionalTestCase): |
|---|
| 170 | |
|---|
| 171 | layer = FunctionalLayer |
|---|
| 172 | |
|---|
| 173 | def setUp(self): |
|---|
| 174 | super(FunctionalExtFileStoreTests, self).setUp() |
|---|
| 175 | self.workdir = tempfile.mkdtemp() |
|---|
| 176 | self.samplefile = os.path.join(self.workdir, 'sample') |
|---|
| 177 | self.otherfile = os.path.join(self.workdir, 'other') |
|---|
| 178 | open(self.samplefile, 'wb').write('Hi there!') |
|---|
| 179 | open(self.otherfile, 'wb').write('Hi from other!') |
|---|
| 180 | self.fd = open(self.samplefile, 'r') |
|---|
| 181 | self.fd2 = open(self.otherfile, 'r') |
|---|
| 182 | self.getRootFolder()['app'] = University() |
|---|
| 183 | self.app = self.getRootFolder()['app'] |
|---|
| 184 | self.app['datacenter'].setStoragePath(self.workdir) |
|---|
| 185 | # register a custom filename mangler |
|---|
| 186 | provideUtility( |
|---|
| 187 | CustomizedFileHandler(), IFileStoreHandler, name=u'mymarker') |
|---|
| 188 | # register a file chooser adapter for CustomContext |
|---|
| 189 | provideAdapter( |
|---|
| 190 | CustomContextFileChooser, |
|---|
| 191 | (CustomContext,), IFileStoreNameChooser) |
|---|
| 192 | return |
|---|
| 193 | |
|---|
| 194 | |
|---|
| 195 | def tearDown(self): |
|---|
| 196 | super(FunctionalExtFileStoreTests, self).tearDown() |
|---|
| 197 | self.fd.close() |
|---|
| 198 | self.fd2.close() |
|---|
| 199 | shutil.rmtree(self.workdir) |
|---|
| 200 | return |
|---|
| 201 | |
|---|
| 202 | def test_root_setup_w_site(self): |
|---|
| 203 | # if a site is available we use it to determine the root dir |
|---|
| 204 | fs = ExtFileStore() |
|---|
| 205 | setSite(self.app) |
|---|
| 206 | self.root = fs.root |
|---|
| 207 | expected_root = os.path.join( |
|---|
| 208 | self.app['datacenter'].storage, 'media') |
|---|
| 209 | self.assertTrue(isinstance(self.root, basestring)) |
|---|
| 210 | self.assertEqual(self.root, expected_root) |
|---|
| 211 | return |
|---|
| 212 | |
|---|
| 213 | def test_get_utility(self): |
|---|
| 214 | # we can get an ExtFileStore by global utility lookup |
|---|
| 215 | fs1 = getUtility(IExtFileStore) |
|---|
| 216 | fs2 = getUtility(IExtFileStore) |
|---|
| 217 | self.assertTrue(isinstance(fs1, ExtFileStore)) |
|---|
| 218 | self.assertTrue(fs1 is fs2) |
|---|
| 219 | return |
|---|
| 220 | |
|---|
| 221 | def test_default_handler_create_file(self): |
|---|
| 222 | # we can use the default handler to store files |
|---|
| 223 | fs = ExtFileStore() |
|---|
| 224 | result = fs.createFile('sample.txt', StringIO('sample text')) |
|---|
| 225 | self.assertEqual(result.data, 'sample.txt') |
|---|
| 226 | self.assertTrue('sample.txt' in os.listdir(fs.root)) |
|---|
| 227 | return |
|---|
| 228 | |
|---|
| 229 | def test_default_handler_get_file(self): |
|---|
| 230 | # we can get files stored by the default handler |
|---|
| 231 | fs = ExtFileStore() |
|---|
| 232 | fs.createFile('sample.txt', StringIO('sample text')) |
|---|
| 233 | result1 = fs.getFile('sample.txt') |
|---|
| 234 | result2 = fs.getFile('not-existent') |
|---|
| 235 | self.assertEqual(result1.read(), 'sample text') |
|---|
| 236 | self.assertTrue(result2 is None) |
|---|
| 237 | return |
|---|
| 238 | |
|---|
| 239 | def test_customized_handler_create_file(self): |
|---|
| 240 | # we can use registered filename handlers |
|---|
| 241 | fs = ExtFileStore() |
|---|
| 242 | result = fs.createFile( |
|---|
| 243 | '__MYMARKER__sample.txt', StringIO('sample text')) |
|---|
| 244 | self.assertEqual(result.data, '__MYMARKER__sample.txt') |
|---|
| 245 | self.assertTrue('sample.txt' in os.listdir(fs.root)) |
|---|
| 246 | return |
|---|
| 247 | |
|---|
| 248 | def test_customized_handler_get_file(self): |
|---|
| 249 | # we consider registered filename handlers when asking for |
|---|
| 250 | # stored files. |
|---|
| 251 | fs = ExtFileStore() |
|---|
| 252 | fs.createFile('__MYMARKER__sample.txt', StringIO('sample text')) |
|---|
| 253 | result1 = fs.getFile('__MYMARKER__sample.txt') |
|---|
| 254 | result2 = fs.getFile('__MYMARKER__not-existent') |
|---|
| 255 | result3 = fs.getFile('not-existent') |
|---|
| 256 | self.assertEqual(result1.read(), 'sample text') |
|---|
| 257 | self.assertTrue(result2 is None) |
|---|
| 258 | self.assertTrue(result3 is None) |
|---|
| 259 | return |
|---|
| 260 | |
|---|
| 261 | def test_get_file_by_context(self): |
|---|
| 262 | # if we register a file name chooser, we can also get a file |
|---|
| 263 | # by context |
|---|
| 264 | fs = ExtFileStore() |
|---|
| 265 | context = CustomContext() |
|---|
| 266 | file_id = IFileStoreNameChooser(context).chooseName() |
|---|
| 267 | fs = ExtFileStore() |
|---|
| 268 | fs.createFile(file_id, StringIO('my sample')) |
|---|
| 269 | result = fs.getFileByContext(context) |
|---|
| 270 | self.assertEqual( |
|---|
| 271 | file_id, '__mymarker__mysample.txt') |
|---|
| 272 | self.assertEqual( |
|---|
| 273 | result.read(), 'my sample') |
|---|
| 274 | return |
|---|
| 275 | |
|---|
| 276 | def test_get_default_handler(self): |
|---|
| 277 | # we can get a default handler |
|---|
| 278 | result = queryUtility(IFileStoreHandler) |
|---|
| 279 | self.assertTrue( |
|---|
| 280 | isinstance(result, DefaultFileStoreHandler)) |
|---|
| 281 | return |
|---|
| 282 | |
|---|
| 283 | def test_get_default_file_retrieval(self): |
|---|
| 284 | # we get a file store when requesting a file retrieval |
|---|
| 285 | result = queryUtility(IFileRetrieval) |
|---|
| 286 | self.assertTrue( |
|---|
| 287 | isinstance(result, DefaultStorage)) |
|---|
