+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
from __future__ import absolute_import
from builtins import hex
from builtins import str
c = Collection('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count.txt\n', keep_client=keep)
writer = c.open("count.txt", "ab+")
self.assertEqual(writer.read(20), b"0123456789")
- writer.seek(0, os.SEEK_SET)
+ writer.seek(0, os.SEEK_SET)
writer.write("hello")
- self.assertEqual(writer.read(20), b"0123456789hello")
+ self.assertEqual(writer.read(), b"")
+ writer.seek(-5, os.SEEK_CUR)
+ self.assertEqual(writer.read(3), b"hel")
+ self.assertEqual(writer.read(), b"lo")
writer.seek(0, os.SEEK_SET)
+ self.assertEqual(writer.read(), b"0123456789hello")
+ writer.seek(0)
writer.write("world")
- self.assertEqual(writer.read(20), b"0123456789helloworld")
+ self.assertEqual(writer.read(), b"")
+ writer.seek(0)
+ self.assertEqual(writer.read(), b"0123456789helloworld")
self.assertEqual(". 781e5e245d69b566979b86e28d23f2c7+10 fc5e038d38a57032085441e7fe7010b0+10 0:20:count.txt\n", c.portable_manifest_text())
mockkeep = mock.MagicMock()
with arvados.arvfile._BlockManager(mockkeep) as blockmanager:
bufferblock = blockmanager.alloc_bufferblock()
- bufferblock.owner = mock.MagicMock()
+ bufferblock.owner = mock.MagicMock(spec=arvados.arvfile.ArvadosFile)
def flush(sync=None):
blockmanager.commit_bufferblock(bufferblock, sync)
bufferblock.owner.flush.side_effect = flush
mockkeep.put.side_effect = arvados.errors.KeepWriteError("fail")
with arvados.arvfile._BlockManager(mockkeep) as blockmanager:
bufferblock = blockmanager.alloc_bufferblock()
- bufferblock.owner = mock.MagicMock()
+ bufferblock.owner = mock.MagicMock(spec=arvados.arvfile.ArvadosFile)
def flush(sync=None):
blockmanager.commit_bufferblock(bufferblock, sync)
bufferblock.owner.flush.side_effect = flush