← Back to branch summary

~jderose/dmedia/transfer-manager

« back to all changes in this revision

Viewing changes to dmedia/tests/test_transfers.py

  • Committer: Jason Gerard DeRose
  • Date: 2011-04-19 03:19:36 UTC
  • Revision ID: jderose@novacut.com-20110419031936-3fbeh9l3kxwi6u6f
More doodles

Show diffs side-by-side

added added

removed removed

Lines of Context:
dmedia/tests/test_transfers.py
26
26
from unittest import TestCase
27
27
from os import urandom
28
28
from base64 import b32encode
29
 
 
30
 
from dmedia import transfers
 
29
from multiprocessing import current_process
 
30
import time
 
31
 
 
32
from dmedia import transfers, schema, filestore
 
33
 
 
34
from .helpers import DummyQueue, mov_hash, mov_size, mov_leaves
 
35
from .couch import CouchCase
31
36
 
32
37
 
33
38
 
225
230
            str(cm.exception),
226
231
            'TransferBackend.upload()'
227
232
        )
 
233
 
 
234
 
 
235
class TestTransferWorker(CouchCase):
 
236
    klass = transfers.TransferWorker
 
237
 
 
238
    def setUp(self):
 
239
        super(TestTransferWorker, self).setUp()
 
240
        self.q = DummyQueue()
 
241
        self.pid = current_process().pid
 
242
 
 
243
    def new(self):
 
244
        self.store_id = schema.random_id()
 
245
        key = ('upload', mov_hash, self.store_id)
 
246
        return self.klass(self.env, self.q, key, (mov_hash, self.store_id))
 
247
 
 
248
    def test_init(self):
 
249
        inst = self.new()
 
250
        self.assertIsInstance(inst.filestore, filestore.FileStore)
 
251
        self.assertEqual(inst.filestore.parent, self.env['filestore']['path'])
 
252
        self.assertEqual(inst.filestore_id, self.env['filestore']['_id'])
 
253
 
 
254
    def test_init_file(self):
 
255
        inst = self.new()
 
256
        doc = schema.create_file(mov_size, mov_leaves, self.store_id)
 
257
        self.assertEqual(doc['_id'], mov_hash)
 
258
        inst.db.save(doc)
 
259
        self.assertIsNone(inst.init_file(mov_hash))
 
260
        self.assertEqual(inst.file_id, mov_hash)
 
261
        self.assertEqual(
 
262
            inst.file.pop('_attachments')['leaves']['data'],
 
263
            doc.pop('_attachments')['leaves']['data']
 
264
        )
 
265
        self.assertEqual(inst.file, doc)
 
266
        self.assertEqual(inst.file_size, mov_size)
 
267
 
 
268
    def test_init_remote(self):
 
269
        inst = self.new()
 
270
        doc = {
 
271
            '_id': self.store_id,
 
272
            'ver': 0,
 
273
            'type': 'dmedia/store',
 
274
            'time': time.time(),
 
275
            'plugin': 's3',
 
276
            'copies': 3,
 
277
            'bucket': 'stuff',
 
278
        }
 
279
        inst.db.save(doc)
 
280
        self.assertIsNone(inst.init_remote(self.store_id))
 
281
        self.assertEqual(inst.remote_id, self.store_id)
 
282
        self.assertEqual(inst.remote, doc)