14946: Updates SSO's install documentation so it doesn't ask for ruby 2.5
[arvados.git] / sdk / python / tests / test_arvfile.py
index a8009130551fe3b0ecf0d74aa3ebfe834eb59ad6..a760255dd6da8e01470dacafa6bcfafeddc50058 100644 (file)
@@ -1,3 +1,7 @@
+# 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
@@ -213,26 +217,41 @@ class ArvadosFileWriterTestCase(unittest.TestCase):
         keep = ArvadosFileWriterTestCase.MockKeep({
             "781e5e245d69b566979b86e28d23f2c7+10": b"0123456789",
         })
-        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.write("hello")
-        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(), 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())
+        for (mode, convert) in (
+                ('a+', lambda data: data.decode(encoding='utf-8')),
+                ('at+', lambda data: data.decode(encoding='utf-8')),
+                ('ab+', lambda data: data)):
+            c = Collection('. 781e5e245d69b566979b86e28d23f2c7+10 0:10:count.txt\n', keep_client=keep)
+            writer = c.open("count.txt", mode)
+            self.assertEqual(writer.read(20), convert(b"0123456789"))
+
+            writer.seek(0, os.SEEK_SET)
+            writer.write(convert(b"hello"))
+            self.assertEqual(writer.read(), convert(b""))
+            if 'b' in mode:
+                writer.seek(-5, os.SEEK_CUR)
+                self.assertEqual(writer.read(3), convert(b"hel"))
+                self.assertEqual(writer.read(), convert(b"lo"))
+            else:
+                with self.assertRaises(IOError):
+                    writer.seek(-5, os.SEEK_CUR)
+                with self.assertRaises(IOError):
+                    writer.seek(-3, os.SEEK_END)
+            writer.seek(0, os.SEEK_SET)
+            writer.read(7)
+            self.assertEqual(7, writer.tell())
+            self.assertEqual(7, writer.seek(7, os.SEEK_SET))
+
+            writer.seek(0, os.SEEK_SET)
+            self.assertEqual(writer.read(), convert(b"0123456789hello"))
+
+            writer.seek(0)
+            writer.write(convert(b"world"))
+            self.assertEqual(writer.read(), convert(b""))
+            writer.seek(0)
+            self.assertEqual(writer.read(), convert(b"0123456789helloworld"))
+
+            self.assertEqual(". 781e5e245d69b566979b86e28d23f2c7+10 fc5e038d38a57032085441e7fe7010b0+10 0:20:count.txt\n", c.portable_manifest_text())
 
     def test_write_at_beginning(self):
         keep = ArvadosFileWriterTestCase.MockKeep({
@@ -832,7 +851,7 @@ class BlockManagerTest(unittest.TestCase):
         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
@@ -863,7 +882,7 @@ class BlockManagerTest(unittest.TestCase):
         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