+class CollectionMethods(run_test_server.TestCaseWithServers):
+
+ def test_keys_values_items_support_indexing(self):
+ c = Collection()
+ with c.open('foo', 'wb') as f:
+ f.write(b'foo')
+ with c.open('bar', 'wb') as f:
+ f.write(b'bar')
+ self.assertEqual(2, len(c.keys()))
+ if sys.version_info < (3, 0):
+ # keys() supports indexing only for python2 callers.
+ fn0 = c.keys()[0]
+ fn1 = c.keys()[1]
+ else:
+ fn0, fn1 = c.keys()
+ self.assertEqual(2, len(c.values()))
+ f0 = c.values()[0]
+ f1 = c.values()[1]
+ self.assertEqual(2, len(c.items()))
+ self.assertEqual(fn0, c.items()[0][0])
+ self.assertEqual(fn1, c.items()[1][0])
+
+ def test_get_properties(self):
+ c = Collection()
+ self.assertEqual(c.get_properties(), {})
+ c.save_new(properties={"foo":"bar"})
+ self.assertEqual(c.get_properties(), {"foo":"bar"})
+
+ def test_get_trash_at(self):
+ c = Collection()
+ self.assertEqual(c.get_trash_at(), None)
+ c.save_new(trash_at=datetime.datetime(2111, 1, 1, 11, 11, 11, 111111))
+ self.assertEqual(c.get_trash_at(), ciso8601.parse_datetime('2111-01-01T11:11:11.111111000Z'))
+
+
+class CollectionOpenModes(run_test_server.TestCaseWithServers):
+
+ def test_open_binary_modes(self):
+ c = Collection()
+ for mode in ['wb', 'wb+', 'ab', 'ab+']:
+ with c.open('foo', mode) as f:
+ f.write(b'foo')
+
+ def test_open_invalid_modes(self):
+ c = Collection()
+ for mode in ['+r', 'aa', '++', 'r+b', 'beer', '', None]:
+ with self.assertRaises(Exception):
+ c.open('foo', mode)
+
+ def test_open_text_modes(self):
+ c = Collection()
+ with c.open('foo', 'wb') as f:
+ f.write('foo')
+ for mode in ['r', 'rt', 'r+', 'rt+', 'w', 'wt', 'a', 'at']:
+ with c.open('foo', mode) as f:
+ if mode[0] == 'r' and '+' not in mode:
+ self.assertEqual('foo', f.read(3))
+ else:
+ f.write('bar')
+ f.seek(0, os.SEEK_SET)
+ self.assertEqual('bar', f.read(3))
+
+
+class TextModes(run_test_server.TestCaseWithServers):
+
+ def setUp(self):
+ arvados.config.KEEP_BLOCK_SIZE = 4
+ if sys.version_info < (3, 0):
+ import unicodedata
+ self.sailboat = unicodedata.lookup('SAILBOAT')
+ self.snowman = unicodedata.lookup('SNOWMAN')
+ else:
+ self.sailboat = '\N{SAILBOAT}'
+ self.snowman = '\N{SNOWMAN}'
+
+ def tearDown(self):
+ arvados.config.KEEP_BLOCK_SIZE = 2 ** 26
+
+ def test_read_sailboat_across_block_boundary(self):
+ c = Collection()
+ f = c.open('sailboats', 'wb')
+ data = self.sailboat.encode('utf-8')
+ f.write(data)
+ f.write(data[:1])
+ f.write(data[1:])
+ f.write(b'\n')
+ f.close()
+ self.assertRegex(c.portable_manifest_text(), r'\+4 .*\+3 ')
+
+ f = c.open('sailboats', 'r')
+ string = f.readline()
+ self.assertEqual(string, self.sailboat+self.sailboat+'\n')
+ f.close()
+
+ def test_write_snowman_across_block_boundary(self):
+ c = Collection()
+ f = c.open('snowmany', 'w')
+ data = self.snowman
+ f.write(data+data+'\n'+data+'\n')
+ f.close()
+ self.assertRegex(c.portable_manifest_text(), r'\+4 .*\+4 .*\+3 ')
+
+ f = c.open('snowmany', 'r')
+ self.assertEqual(f.readline(), self.snowman+self.snowman+'\n')
+ self.assertEqual(f.readline(), self.snowman+'\n')
+ f.close()
+
+