Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_write_file_like(self):
one = motor.MotorGridIn(self.db.fs)
yield one.write(b"hello world")
yield one.close()
two = motor.MotorGridOut(self.db.fs, one._id)
three = motor.MotorGridIn(self.db.fs)
yield three.write(two)
yield three.close()
four = motor.MotorGridOut(self.db.fs, three._id)
self.assertEqual(b"hello world", (yield four.read()))
def test_attributes(self):
f = motor.MotorGridIn(
self.db.fs,
filename="test",
foo="bar",
content_type="text")
yield f.close()
g = motor.MotorGridOut(self.db.fs, f._id)
attr_names = (
'_id',
'filename',
'name',
'name',
'content_type',
'length',
'chunk_size',
'upload_date',
'aliases',
'metadata',
'md5')
for attr_name in attr_names:
self.assertRaises(InvalidOperation, getattr, g, attr_name)
def test_grid_out_file_document(self):
one = motor.MotorGridIn(self.db.fs)
yield one.write(b"foo bar")
yield one.close()
file_document = yield self.db.fs.files.find_one()
two = motor.MotorGridOut(
self.db.fs, file_document=file_document)
self.assertEqual(b"foo bar", (yield two.read()))
file_document = yield self.db.fs.files.find_one()
three = motor.MotorGridOut(self.db.fs, 5, file_document)
self.assertEqual(b"foo bar", (yield three.read()))
gridout = motor.MotorGridOut(self.db.fs, file_document={})
with self.assertRaises(NoFile):
yield gridout.open()
def test_grid_out_default_opts(self):
self.assertRaises(TypeError, motor.MotorGridOut, "foo")
gout = motor.MotorGridOut(self.db.fs, 5)
with self.assertRaises(NoFile):
yield gout.open()
a = motor.MotorGridIn(self.db.fs)
yield a.close()
b = yield motor.MotorGridOut(self.db.fs, a._id).open()
self.assertEqual(a._id, b._id)
self.assertEqual(0, b.length)
self.assertEqual(None, b.content_type)
self.assertEqual(255 * 1024, b.chunk_size)
self.assertTrue(isinstance(b.upload_date, datetime.datetime))
self.assertEqual(None, b.aliases)
self.assertEqual(None, b.metadata)
self.assertEqual("d41d8cd98f00b204e9800998ecf8427e", b.md5)
def test_grid_out_custom_opts(self):
one = motor.MotorGridIn(
self.db.fs, _id=5, filename="my_file",
contentType="text/html", chunkSize=1000, aliases=["foo"],
metadata={"foo": 1, "bar": 2}, bar=3, baz="hello")
yield one.write(b"hello world")
yield one.close()
two = yield motor.MotorGridOut(self.db.fs, 5).open()
self.assertEqual(5, two._id)
self.assertEqual(11, two.length)
self.assertEqual("text/html", two.content_type)
self.assertEqual(1000, two.chunk_size)
self.assertTrue(isinstance(two.upload_date, datetime.datetime))
self.assertEqual(["foo"], two.aliases)
self.assertEqual({"foo": 1, "bar": 2}, two.metadata)
self.assertEqual(3, two.bar)
self.assertEqual("5eb63bbbe01eeed093cb22bb8f5acdc3", two.md5)
def test_gridout_open_exc_info(self):
if sys.version_info < (3, ):
raise SkipTest("Requires Python 3")
g = motor.MotorGridOut(self.db.fs, "_id that doesn't exist")
try:
yield g.open()
except NoFile:
_, _, tb = sys.exc_info()
# The call tree should include PyMongo code we ran on a thread.
formatted = '\n'.join(traceback.format_tb(tb))
self.assertTrue('_ensure_file' in formatted)
def test_grid_out_file_document(self):
one = motor.MotorGridIn(self.db.fs)
yield one.write(b"foo bar")
yield one.close()
file_document = yield self.db.fs.files.find_one()
two = motor.MotorGridOut(
self.db.fs, file_document=file_document)
self.assertEqual(b"foo bar", (yield two.read()))
file_document = yield self.db.fs.files.find_one()
three = motor.MotorGridOut(self.db.fs, 5, file_document)
self.assertEqual(b"foo bar", (yield three.read()))
gridout = motor.MotorGridOut(self.db.fs, file_document={})
with self.assertRaises(NoFile):
yield gridout.open()
def test_grid_out_default_opts(self):
self.assertRaises(TypeError, motor.MotorGridOut, "foo")
gout = motor.MotorGridOut(self.db.fs, 5)
with self.assertRaises(NoFile):
yield gout.open()
a = motor.MotorGridIn(self.db.fs)
yield a.close()
b = yield motor.MotorGridOut(self.db.fs, a._id).open()
self.assertEqual(a._id, b._id)
self.assertEqual(0, b.length)
self.assertEqual(None, b.content_type)
self.assertEqual(255 * 1024, b.chunk_size)
self.assertTrue(isinstance(b.upload_date, datetime.datetime))
self.assertEqual(None, b.aliases)
self.assertEqual(None, b.metadata)
self.assertEqual("d41d8cd98f00b204e9800998ecf8427e", b.md5)
# Send the initial aggregate as PyMongo expects.
motor_obj._lazy_init()
return ChangeStream(motor_obj)
if isinstance(motor_obj, motor.motor_tornado.MotorLatentCommandCursor):
return CommandCursor(motor_obj)
if isinstance(motor_obj, motor.motor_tornado.MotorCommandCursor):
return CommandCursor(motor_obj)
if isinstance(motor_obj, _MotorRawBatchCommandCursor):
return CommandCursor(motor_obj)
if isinstance(motor_obj, motor.motor_tornado.MotorCursor):
return Cursor(motor_obj)
if isinstance(motor_obj, _MotorRawBatchCursor):
return Cursor(motor_obj)
if isinstance(motor_obj, motor.MotorGridIn):
return GridIn(None, delegate=motor_obj)
if isinstance(motor_obj, motor.MotorGridOut):
return GridOut(None, delegate=motor_obj)
if isinstance(motor_obj, motor.motor_tornado.MotorGridOutCursor):
return GridOutCursor(motor_obj)
else:
return motor_obj