tests/test_gio.py

   1 # -*- Mode: Python -*-
   2
   3 import os
   4 import unittest
   5
   6 import glib
   7 import gio
   8
   9
  10 class TestFile(unittest.TestCase):
  11     def setUp(self):
  12         self._f = open("file.txt", "w+")
  13         self.file = gio.File("file.txt")
  14
  15     def tearDown(self):
  16         self._f.close()
  17         if os.path.exists('file.txt'):
  18             os.unlink("file.txt")
  19
  20     def testReadAsync(self):
  21         self._f.write("testing")
  22         self._f.seek(0)
  23
  24         def callback(file, result):
  25             try:
  26                 stream = file.read_finish(result)
  27                 self.failUnless(isinstance(stream, gio.InputStream))
  28                 self.assertEquals(stream.read(), "testing")
  29             finally:
  30                 loop.quit()
  31
  32         self.file.read_async(callback)
  33
  34         loop = glib.MainLoop()
  35         loop.run()
  36
  37     def testAppendToAsync(self):
  38         self._f.write("append_to ")
  39         self._f.close()
  40
  41         def callback(file, result):
  42             try:
  43                 stream = file.append_to_finish(result)
  44                 self.failUnless(isinstance(stream, gio.OutputStream))
  45                 w = stream.write("testing")
  46                 cont, leng, etag = self.file.load_contents()
  47                 self.assertEqual(cont, "append_to testing")
  48             finally:
  49                 loop.quit()
  50
  51         self.file.append_to_async(callback, gio.FILE_CREATE_NONE,
  52                                   glib.PRIORITY_HIGH)
  53
  54         loop = glib.MainLoop()
  55         loop.run()
  56
  57     def testAppendToAsyncNoargs(self):
  58         self._f.write("append_to ")
  59         self._f.close()
  60
  61         def callback(file, result):
  62             try:
  63                 stream = file.append_to_finish(result)
  64                 self.failUnless(isinstance(stream, gio.OutputStream))
  65                 w = stream.write("testing")
  66                 cont, leng, etag = self.file.load_contents()
  67                 self.assertEqual(cont, "append_to testing")
  68             finally:
  69                 loop.quit()
  70
  71         self.file.append_to_async(callback)
  72
  73         loop = glib.MainLoop()
  74         loop.run()
  75
  76     def testCreateAsync(self):
  77         def callback(file, result):
  78             try:
  79                 stream = file.create_finish(result)
  80                 self.failUnless(isinstance(stream, gio.OutputStream))
  81                 w = stream.write("testing")
  82                 cont, leng, etag = file.load_contents()
  83                 self.assertEqual(cont, "testing")
  84             finally:
  85                 if os.path.exists('temp.txt'):
  86                     os.unlink("temp.txt")
  87                 loop.quit()
  88
  89         gfile = gio.File("temp.txt")
  90         gfile.create_async(callback, gio.FILE_CREATE_NONE,
  91                            glib.PRIORITY_HIGH)
  92
  93         loop = glib.MainLoop()
  94         loop.run()
  95
  96     def testCreateReadWriteAsync(self):
  97         def callback(file, result):
  98             try:
  99                 iostream = file.create_readwrite_finish(result)
 100                 self.failUnless(isinstance(iostream, gio.FileIOStream))
 101
 102                 ostream = iostream.get_output_stream()
 103                 self.failUnless(isinstance(ostream, gio.OutputStream))
 104
 105                 w = ostream.write("testing")
 106                 cont, leng, etag = file.load_contents()
 107                 self.assertEqual(cont, "testing")
 108             finally:
 109                 if os.path.exists('temp.txt'):
 110                     os.unlink("temp.txt")
 111                 loop.quit()
 112
 113         gfile = gio.File("temp.txt")
 114         gfile.create_readwrite_async(callback, gio.FILE_CREATE_NONE,
 115                                      glib.PRIORITY_HIGH)
 116
 117         loop = glib.MainLoop()
 118         loop.run()
 119
 120     def testCreateAsyncNoargs(self):
 121         def callback(file, result):
 122             try:
 123                 stream = file.create_finish(result)
 124                 self.failUnless(isinstance(stream, gio.OutputStream))
 125                 w = stream.write("testing")
 126                 cont, leng, etag = file.load_contents()
 127                 self.assertEqual(cont, "testing")
 128             finally:
 129                 if os.path.exists('temp.txt'):
 130                     os.unlink("temp.txt")
 131                 loop.quit()
 132
 133         gfile = gio.File("temp.txt")
 134         gfile.create_async(callback)
 135
 136         loop = glib.MainLoop()
 137         loop.run()
 138
 139     def testReplaceAsync(self):
 140         self._f.write("testing")
 141         self._f.close()
 142
 143         def callback(file, result):
 144             try:
 145                 stream = file.replace_finish(result)
 146                 self.failUnless(isinstance(stream, gio.OutputStream))
 147                 stream.write("some new string")
 148                 stream.close()
 149                 cont, leng, etag = file.load_contents()
 150                 self.assertEqual(cont, "some new string")
 151             finally:
 152                 loop.quit()
 153
 154
 155         self.file.replace_async(callback, None, True, gio.FILE_CREATE_NONE,
 156                                 glib.PRIORITY_HIGH)
 157
 158         loop = glib.MainLoop()
 159         loop.run()
 160
 161     def testReplaceAsyncNoargs(self):
 162         self._f.write("testing")
 163         self._f.close()
 164
 165         def callback(file, result):
 166             try:
 167                 stream = file.replace_finish(result)
 168                 self.failUnless(isinstance(stream, gio.OutputStream))
 169                 stream.write("some new string")
 170                 stream.close()
 171                 cont, leng, etag = file.load_contents()
 172                 self.assertEqual(cont, "some new string")
 173             finally:
 174                 loop.quit()
 175
 176
 177         self.file.replace_async(callback)
 178
 179         loop = glib.MainLoop()
 180         loop.run()
 181
 182     def testReadAsyncError(self):
 183         self.assertRaises(TypeError, self.file.read_async)
 184         self.assertRaises(TypeError, self.file.read_async, "foo", "bar")
 185         self.assertRaises(TypeError, self.file.read_async, "foo",
 186                           priority="bar")
 187         self.assertRaises(TypeError, self.file.read_async, "foo",
 188                           priority="bar")
 189         self.assertRaises(TypeError, self.file.read_async, "foo",
 190                           priority=1, cancellable="bar")
 191         self.assertRaises(TypeError, self.file.read_async, "foo", 1, "bar")
 192
 193     def testConstructor(self):
 194         for gfile in [gio.File("/"),
 195                       gio.File("file:///"),
 196                       gio.File(uri="file:///"),
 197                       gio.File(path="/"),
 198                       gio.File(u"/"),
 199                       gio.File(path=u"/")]:
 200             self.failUnless(isinstance(gfile, gio.File))
 201             self.assertEquals(gfile.get_path(), "/")
 202             self.assertEquals(gfile.get_uri(), "file:///")
 203
 204     def testConstructorError(self):
 205         self.assertRaises(TypeError, gio.File)
 206         self.assertRaises(TypeError, gio.File, 1)
 207         self.assertRaises(TypeError, gio.File, "foo", "bar")
 208         self.assertRaises(TypeError, gio.File, foo="bar")
 209         self.assertRaises(TypeError, gio.File, uri=1)
 210         self.assertRaises(TypeError, gio.File, path=1)
 211
 212     def testLoadContents(self):
 213         self._f.write("testing load_contents")
 214         self._f.seek(0)
 215         c = gio.Cancellable()
 216         cont, leng, etag = self.file.load_contents(cancellable=c)
 217         self.assertEqual(cont, "testing load_contents")
 218         self.assertEqual(leng, 21)
 219         self.assertNotEqual(etag, '')
 220
 221     def testLoadContentsAsync(self):
 222         self._f.write("testing load_contents_async")
 223         self._f.seek(0)
 224
 225         def callback(contents, result):
 226             try:
 227                 cont, leng, etag = contents.load_contents_finish(result)
 228                 self.assertEqual(cont, "testing load_contents_async")
 229                 self.assertEqual(leng, 27)
 230                 self.assertNotEqual(etag, '')
 231             finally:
 232                 loop.quit()
 233
 234         canc = gio.Cancellable()
 235         self.file.load_contents_async(callback, cancellable=canc)
 236
 237         loop = glib.MainLoop()
 238         loop.run()
 239
 240     def testQueryInfoAsync(self):
 241         def callback(file, result):
 242             try:
 243                 info = file.query_info_finish(result)
 244                 self.failUnless(isinstance(info, gio.FileInfo))
 245                 self.failUnless(info.get_name(), "file.txt")
 246             finally:
 247                 loop.quit()
 248
 249         self.file.query_info_async("standard", callback)
 250
 251         loop = glib.MainLoop()
 252         loop.run()
 253
 254     def testMountMountable(self):
 255         gfile = gio.File('localtest:')
 256         def unmount_done(mount, result):
 257             try:
 258                 retval = mount.unmount_finish(result)
 259                 self.failUnless(retval)
 260             finally:
 261                 loop.quit()
 262
 263         def mount_enclosing_volume_done(gfile, result):
 264             try:
 265                 try:
 266                     retval = gfile.mount_enclosing_volume_finish(result)
 267                 except gio.Error, e:
 268                     # If we run the tests too fast
 269                     if e.code == gio.ERROR_ALREADY_MOUNTED:
 270                         print ('WARNING: testfile is already mounted, '
 271                         'skipping test')
 272                         loop.quit()
 273                         return
 274                     raise
 275                 self.failUnless(retval)
 276             finally:
 277                 try:
 278                     mount = gfile.find_enclosing_mount()
 279                 except gio.Error:
 280                     loop.quit()
 281                     return
 282                 mount.unmount(unmount_done)
 283
 284         mount_operation = gio.MountOperation()
 285         gfile.mount_enclosing_volume(mount_operation,
 286                                      mount_enclosing_volume_done)
 287
 288         loop = glib.MainLoop()
 289         loop.run()
 290
 291     def testCopy(self):
 292         if os.path.exists('copy.txt'):
 293             os.unlink("copy.txt")
 294
 295         source = gio.File('file.txt')
 296         destination = gio.File('copy.txt')
 297         try:
 298             retval = source.copy(destination)
 299             self.failUnless(retval)
 300
 301             self.failUnless(os.path.exists('copy.txt'))
 302             self.assertEqual(open('file.txt').read(),
 303                              open('copy.txt').read())
 304         finally:
 305             os.unlink("copy.txt")
 306
 307         self.called = False
 308         def callback(current, total):
 309             self.called = True
 310         source = gio.File('file.txt')
 311         destination = gio.File('copy.txt')
 312         try:
 313             retval = source.copy(destination, callback)
 314             self.failUnless(retval)
 315
 316             self.failUnless(os.path.exists('copy.txt'))
 317             self.assertEqual(open('file.txt').read(),
 318                              open('copy.txt').read())
 319             self.failUnless(self.called)
 320         finally:
 321             os.unlink("copy.txt")
 322
 323     def test_copy_async(self):
 324         if os.path.exists('copy.txt'):
 325             os.unlink("copy.txt")
 326
 327         source = gio.File('file.txt')
 328         destination = gio.File('copy.txt')
 329
 330         def copied(source_, result):
 331             try:
 332                 self.assert_(source_ is source)
 333                 self.failUnless(source_.copy_finish(result))
 334             finally:
 335                 loop.quit()
 336
 337         def progress(current, total):
 338             self.assert_(isinstance(current, long))
 339             self.assert_(isinstance(total, long))
 340             self.assert_(0 <= current <= total)
 341
 342         try:
 343             loop = glib.MainLoop()
 344             source.copy_async(destination, copied, progress_callback = progress)
 345             loop.run()
 346
 347             self.failUnless(os.path.exists('copy.txt'))
 348             self.assertEqual(open('file.txt').read(),
 349                              open('copy.txt').read())
 350         finally:
 351             os.unlink("copy.txt")
 352
 353     # See bug 546591.
 354     def test_copy_progress(self):
 355         source = gio.File('file.txt')
 356         destination = gio.File('copy.txt')
 357
 358         def progress(current, total):
 359             self.assert_(isinstance(current, long))
 360             self.assert_(isinstance(total, long))
 361             self.assert_(0 <= current <= total)
 362
 363         try:
 364             retval = source.copy(destination,
 365                                  flags=gio.FILE_COPY_OVERWRITE,
 366                                  progress_callback=progress)
 367             self.failUnless(retval)
 368
 369             self.failUnless(os.path.exists('copy.txt'))
 370             self.assertEqual(open('file.txt').read(),
 371                              open('copy.txt').read())
 372         finally:
 373             os.unlink("copy.txt")
 374
 375     def testMove(self):
 376         if os.path.exists('move.txt'):
 377             os.unlink("move.txt")
 378
 379         source = gio.File('file.txt')
 380         destination = gio.File('move.txt')
 381         retval = source.move(destination)
 382         self.failUnless(retval)
 383
 384         self.failIf(os.path.exists('file.txt'))
 385         self.failUnless(os.path.exists('move.txt'))
 386
 387         self.called = False
 388         def callback(current, total):
 389             self.called = True
 390         source = gio.File('move.txt')
 391         destination = gio.File('move-2.txt')
 392         try:
 393             retval = source.move(destination, callback)
 394             self.failUnless(retval)
 395
 396             self.failIf(os.path.exists('move.txt'))
 397             self.failUnless(os.path.exists('move-2.txt'))
 398             self.failUnless(self.called)
 399         finally:
 400             os.unlink("move-2.txt")
 401
 402     def testInfoList(self):
 403         infolist = self.file.query_settable_attributes()
 404         for info in infolist:
 405             if info.name == "time::modified":
 406                 self.assertEqual(info.type, gio.FILE_ATTRIBUTE_TYPE_UINT64)
 407                 self.assertEqual(info.name, "time::modified")
 408                 self.assertEqual(info.flags,
 409                                  gio.FILE_ATTRIBUTE_INFO_COPY_WHEN_MOVED |
 410                                  gio.FILE_ATTRIBUTE_INFO_COPY_WITH_FILE)
 411
 412     def testQueryWritableNamespaces(self):
 413         infolist = self.file.query_writable_namespaces()
 414         for info in infolist:
 415             if info.name == "xattr":
 416                 self.assertEqual(info.type, gio.FILE_ATTRIBUTE_TYPE_STRING)
 417
 418     def testSetAttribute(self):
 419         self._f.write("testing attributes")
 420         self._f.seek(0)
 421         infolist = self.file.query_settable_attributes()
 422
 423         self.assertNotEqual(len(infolist), 0)
 424
 425         for info in infolist:
 426             if info.name == "time::modified-usec":
 427                 ret = self.file.set_attribute("time::modified-usec",
 428                                               gio.FILE_ATTRIBUTE_TYPE_UINT32,
 429                                               10, gio.FILE_QUERY_INFO_NONE)
 430                 self.assertEqual(ret, True)
 431
 432     def testSetAttributesAsync(self):
 433         def callback(gfile, result):
 434             try:
 435                 info = gfile.set_attributes_finish(result)
 436                 usec = info.get_attribute_uint32("time::modified-usec")
 437                 self.assertEqual(usec, 10)
 438             finally:
 439                 loop.quit()
 440
 441         info = gio.FileInfo()
 442         info.set_attribute_uint32("time::modified-usec", 10)
 443
 444         canc = gio.Cancellable()
 445         self.file.set_attributes_async(info, callback)
 446
 447         loop = glib.MainLoop()
 448         loop.run()
 449
 450     def testReplaceContents(self):
 451         self.file.replace_contents("testing replace_contents")
 452         cont, leng, etag = self.file.load_contents()
 453         self.assertEqual(cont, "testing replace_contents")
 454
 455         caught = False
 456         try:
 457             self.file.replace_contents("this won't work", etag="wrong")
 458         except gio.Error, e:
 459             self.assertEqual(e.code, gio.ERROR_WRONG_ETAG)
 460             caught = True
 461         self.failUnless(caught)
 462
 463         self.file.replace_contents("testing replace_contents again", etag=etag)
 464         cont, leng, etag = self.file.load_contents()
 465         self.assertEqual(cont, "testing replace_contents again")
 466
 467         self.file.replace_contents("testing replace_contents yet again", etag=None)
 468         cont, leng, etag = self.file.load_contents()
 469         self.assertEqual(cont, "testing replace_contents yet again")
 470
 471     def testReplaceContentsAsync(self):
 472
 473         def callback(contents, result):
 474             try:
 475                 newetag = contents.replace_contents_finish(result)
 476                 cont, leng, etag = self.file.load_contents()
 477                 self.assertEqual(cont, "testing replace_contents_async")
 478                 self.assertEqual(leng, 30)
 479                 self.assertEqual(etag, newetag)
 480                 self.assertNotEqual(newetag, '')
 481             finally:
 482                 loop.quit()
 483
 484         canc = gio.Cancellable()
 485         self.file.replace_contents_async("testing replace_contents_async", callback, cancellable=canc)
 486
 487         loop = glib.MainLoop()
 488         loop.run()
 489
 490     def test_eq(self):
 491         self.assertEqual(gio.File('foo'),
 492                          gio.File('foo'))
 493         self.assertNotEqual(gio.File('foo'),
 494                             gio.File('bar'))
 495
 496     def test_hash(self):
 497         self.assertEquals(hash(gio.File('foo')),
 498                           hash(gio.File('foo')))
 499
 500     def testSetDisplayNameAsync(self):
 501         def callback(gfile, result):
 502             try:
 503                 new_gfile = gfile.set_display_name_finish(result)
 504                 new_name = new_gfile.get_basename()
 505                 self.assertEqual(new_name, "new.txt")
 506                 deleted = new_gfile.delete()
 507                 self.assertEqual(deleted, True)
 508             finally:
 509                 loop.quit()
 510
 511         canc = gio.Cancellable()
 512         self.file.set_display_name_async("new.txt", callback, cancellable=canc)
 513
 514         loop = glib.MainLoop()
 515         loop.run()
 516
 517 class TestGFileEnumerator(unittest.TestCase):
 518     def setUp(self):
 519         self.file = gio.File(os.path.dirname(__file__))
 520
 521     def testEnumerateChildren(self):
 522         enumerator = self.file.enumerate_children(
 523             "standard::*", gio.FILE_QUERY_INFO_NOFOLLOW_SYMLINKS)
 524         for file_info in enumerator:
 525             if file_info.get_name() == os.path.basename(__file__):
 526                 break
 527         else:
 528             raise AssertionError
 529
 530     def testEnumerateChildrenAsync(self):
 531         def callback(gfile, result):
 532             try:
 533                 for file_info in gfile.enumerate_children_finish(result):
 534                     if file_info.get_name() == __file__:
 535                         break
 536                 else:
 537                     raise AssertionError
 538             finally:
 539                 loop.quit()
 540
 541         self.file.enumerate_children_async(
 542             "standard::*", callback)
 543         loop = glib.MainLoop()
 544         loop.run()
 545
 546     def testNextFilesAsync(self):
 547         def callback(enumerator, result):
 548             try:
 549                 for file_info in enumerator.next_files_finish(result):
 550                     if file_info.get_name() == __file__:
 551                         break
 552                 else:
 553                     raise AssertionError
 554             finally:
 555                 loop.quit()
 556
 557         enumerator = self.file.enumerate_children(
 558             "standard::*", gio.FILE_QUERY_INFO_NOFOLLOW_SYMLINKS)
 559         enumerator.next_files_async(1000, callback)
 560         loop = glib.MainLoop()
 561         loop.run()
 562
 563     def testCloseFilesAsync(self):
 564         def callback(enumerator, result):
 565             try:
 566                 enumerator.close_finish(result)
 567             finally:
 568                 loop.quit()
 569
 570         enumerator = self.file.enumerate_children(
 571             "standard::*", gio.FILE_QUERY_INFO_NOFOLLOW_SYMLINKS)
 572
 573         enumerator.close_async(callback)
 574
 575         loop = glib.MainLoop()
 576         loop.run()
 577
 578
 579 class TestInputStream(unittest.TestCase):
 580     def setUp(self):
 581         self._f = open("inputstream.txt", "w+")
 582         self._f.write("testing")
 583         self._f.seek(0)
 584         self.stream = gio.unix.InputStream(self._f.fileno(), False)
 585
 586     def tearDown(self):
 587         self._f.close()
 588         os.unlink("inputstream.txt")
 589
 590     def testRead(self):
 591         self.assertEquals(self.stream.read(), "testing")
 592
 593         self.stream = gio.MemoryInputStream()
 594         self.assertEquals(self.stream.read(), '')
 595
 596         self.stream = gio.MemoryInputStream()
 597         some_data = open(__file__, "rb").read()
 598         self.stream.add_data(some_data)
 599         self.assertEquals(self.stream.read(), some_data)
 600
 601         stream = gio.MemoryInputStream()
 602         stream.add_data(some_data)
 603         self.assertEquals(self._read_in_loop(stream,
 604                                              lambda: stream.read(50),
 605                                              50),
 606                           some_data)
 607
 608     def testSkip(self):
 609         self.stream.skip(2)
 610         res = self.stream.read()
 611         self.assertEqual(res, "sting")
 612
 613     def testSkipAsync(self):
 614         def callback(stream, result):
 615             try:
 616                 size = stream.skip_finish(result)
 617                 self.assertEqual(size, 2)
 618                 res = stream.read()
 619                 self.assertEqual(res, "sting")
 620             finally:
 621                 loop.quit()
 622
 623         self.stream.skip_async(2, callback)
 624
 625         loop = glib.MainLoop()
 626         loop.run()
 627
 628     def test_read_part(self):
 629         self.assertEquals(self._read_in_loop(self.stream,
 630                                              lambda: self.stream.read_part()),
 631                           'testing')
 632
 633         stream = gio.MemoryInputStream()
 634         some_data = open(__file__, 'rb').read()
 635         stream.add_data(some_data)
 636         self.assertEquals(self._read_in_loop(stream,
 637                                              lambda: stream.read_part(50),
 638                                              50),
 639                           some_data)
 640
 641     def _read_in_loop(self, stream, reader, size_limit=0):
 642         read_data = ''
 643         while True:
 644             read_part = reader()
 645             if read_part:
 646                 read_data += read_part
 647                 if size_limit > 0:
 648                     self.assert_(len(read_part) <= size_limit,
 649                                  '%d <= %d' % (len(read_part), size_limit))
 650             else:
 651                 return read_data
 652
 653     def testReadAsync(self):
 654         def callback(stream, result):
 655             self.assertEquals(result.get_op_res_gssize(), 7)
 656             try:
 657                 data = stream.read_finish(result)
 658                 self.assertEquals(data, "testing")
 659                 stream.close()
 660             finally:
 661                 loop.quit()
 662
 663         self.stream.read_async(7, callback)
 664
 665         loop = glib.MainLoop()
 666         loop.run()
 667
 668     def testReadAsyncError(self):
 669         self.count = 0
 670         def callback(stream, result):
 671             try:
 672                 self.count += 1
 673                 if self.count == 1:
 674                     return
 675                 self.assertRaises(gio.Error, stream.read_finish, result)
 676             finally:
 677                 loop.quit()
 678
 679         self.stream.read_async(10240, callback)
 680         self.stream.read_async(10240, callback)
 681
 682         loop = glib.MainLoop()
 683         loop.run()
 684
 685         self.assertEquals(self.count, 2)
 686
 687         self.assertRaises(TypeError, self.stream.read_async)
 688         self.assertRaises(TypeError, self.stream.read_async, "foo")
 689         self.assertRaises(TypeError, self.stream.read_async, 1024, "bar")
 690         self.assertRaises(TypeError, self.stream.read_async, 1024,
 691                           priority="bar")
 692         self.assertRaises(TypeError, self.stream.read_async, 1024,
 693                           priority="bar")
 694         self.assertRaises(TypeError, self.stream.read_async, 1024,
 695                           priority=1, cancellable="bar")
 696         self.assertRaises(TypeError, self.stream.read_async, 1024, 1, "bar")
 697
 698
 699     # FIXME: this makes 'make check' freeze
 700     def _testCloseAsync(self):
 701         def callback(stream, result):
 702             try:
 703                 self.failUnless(stream.close_finish(result))
 704             finally:
 705                 loop.quit()
 706
 707         self.stream.close_async(callback)
 708
 709         loop = glib.MainLoop()
 710         loop.run()
 711
 712
 713 class TestDataInputStream(unittest.TestCase):
 714     def setUp(self):
 715         self.base_stream = gio.MemoryInputStream()
 716         self.data_stream = gio.DataInputStream(self.base_stream)
 717
 718     def test_read_line(self):
 719         self.base_stream.add_data('foo\nbar\n\nbaz')
 720         self.assertEquals('foo', self.data_stream.read_line())
 721         self.assertEquals('bar', self.data_stream.read_line())
 722         self.assertEquals('', self.data_stream.read_line())
 723         self.assertEquals('baz', self.data_stream.read_line())
 724
 725     def test_read_line_async(self):
 726         def do_read_line_async():
 727             loop = glib.MainLoop()
 728             line = []
 729
 730             def callback(stream, result):
 731                 try:
 732                     line.append(stream.read_line_finish(result))
 733                 finally:
 734                     loop.quit()
 735
 736             self.data_stream.read_line_async(callback)
 737             loop.run()
 738             return line[0]
 739
 740         self.base_stream.add_data('foo\nbar\n\nbaz')
 741         self.assertEquals('foo', do_read_line_async())
 742         self.assertEquals('bar', do_read_line_async())
 743         self.assertEquals('', do_read_line_async())
 744         self.assertEquals('baz', do_read_line_async())
 745
 746     def test_read_until(self):
 747         self.base_stream.add_data('sentence.end of line\nthe rest')
 748         self.assertEquals('sentence', self.data_stream.read_until('.!?'))
 749         self.assertEquals('end of line', self.data_stream.read_until('\n\r'))
 750         self.assertEquals('the rest', self.data_stream.read_until('#$%^&'))
 751
 752     def test_read_until_async(self):
 753         def do_read_until_async(stop_chars):
 754             loop = glib.MainLoop()
 755             data = []
 756
 757             def callback(stream, result):
 758                 try:
 759                     data.append(stream.read_until_finish(result))
 760                 finally:
 761                     loop.quit()
 762
 763             self.data_stream.read_until_async(stop_chars, callback)
 764             loop.run()
 765             return data[0]
 766
 767         # Note the weird difference between synchronous and
 768         # asynchronous version.  See bug #584284.
 769         self.base_stream.add_data('sentence.end of line\nthe rest')
 770         self.assertEquals('sentence', do_read_until_async('.!?'))
 771         self.assertEquals('.end of line', do_read_until_async('\n\r'))
 772         self.assertEquals('\nthe rest', do_read_until_async('#$%^&'))
 773
 774
 775 class TestMemoryInputStream(unittest.TestCase):
 776     def setUp(self):
 777         self.stream = gio.MemoryInputStream()
 778
 779     def test_add_data(self):
 780         self.stream.add_data('foobar')
 781         self.assertEquals('foobar', self.stream.read())
 782
 783         self.stream.add_data('ham ')
 784         self.stream.add_data(None)
 785         self.stream.add_data('spam')
 786         self.assertEquals('ham spam', self.stream.read())
 787
 788     def test_new_from_data(self):
 789         stream = gio.memory_input_stream_new_from_data('spam')
 790         self.assertEquals('spam', stream.read())
 791
 792
 793 class TestOutputStream(unittest.TestCase):
 794     def setUp(self):
 795         self._f = open("outputstream.txt", "w")
 796         self.stream = gio.unix.OutputStream(self._f.fileno(), False)
 797
 798     def tearDown(self):
 799         self._f.close()
 800         os.unlink("outputstream.txt")
 801
 802     def testWrite(self):
 803         self.stream.write("testing")
 804         self.stream.close()
 805         self.failUnless(os.path.exists("outputstream.txt"))
 806         self.assertEquals(open("outputstream.txt").read(), "testing")
 807
 808     def test_write_part(self):
 809         stream = gio.MemoryOutputStream()
 810         some_data = open(__file__, 'rb').read()
 811         buffer = some_data
 812
 813         # In fact this makes only one looping (memory stream is fast,
 814         # write_part behaves just like write), but let's still be
 815         # complete.
 816         while buffer:
 817             written = stream.write_part(buffer)
 818             if written == len(buffer):
 819                 break
 820             else:
 821                 buffer = buffer[written:]
 822
 823         self.assertEquals(stream.get_contents(), some_data)
 824
 825     def testWriteAsync(self):
 826         def callback(stream, result):
 827             self.assertEquals(result.get_op_res_gssize(), 7)
 828             try:
 829                 self.assertEquals(stream.write_finish(result), 7)
 830                 self.failUnless(os.path.exists("outputstream.txt"))
 831                 self.assertEquals(open("outputstream.txt").read(), "testing")
 832             finally:
 833                 loop.quit()
 834
 835         self.stream.write_async("testing", callback)
 836
 837         loop = glib.MainLoop()
 838         loop.run()
 839
 840     def testWriteAsyncError(self):
 841         def callback(stream, result):
 842             self.assertEquals(result.get_op_res_gssize(), 0)
 843             try:
 844                 self.assertRaises(gio.Error, stream.write_finish, result)
 845             finally:
 846                 loop.quit()
 847
 848         self.stream.close()
 849         self.stream.write_async("testing", callback)
 850
 851         loop = glib.MainLoop()
 852         loop.run()
 853
 854         self.assertRaises(TypeError, self.stream.write_async)
 855         self.assertRaises(TypeError, self.stream.write_async, 1138)
 856         self.assertRaises(TypeError, self.stream.write_async, "foo", "bar")
 857         self.assertRaises(TypeError, self.stream.write_async, "foo",
 858                           priority="bar")
 859         self.assertRaises(TypeError, self.stream.write_async, "foo",
 860                           priority="bar")
 861         self.assertRaises(TypeError, self.stream.write_async, "foo",
 862                           priority=1, cancellable="bar")
 863         self.assertRaises(TypeError, self.stream.write_async, "foo", 1, "bar")
 864
 865     # FIXME: this makes 'make check' freeze
 866     def _testCloseAsync(self):
 867         def callback(stream, result):
 868             try:
 869                 self.failUnless(stream.close_finish(result))
 870             finally:
 871                 loop.quit()
 872
 873         self.stream.close_async(callback)
 874
 875         loop = glib.MainLoop()
 876         loop.run()
 877
 878     def testFlushAsync(self):
 879         def callback(stream, result):
 880             try:
 881                 self.failUnless(stream.flush_finish(result))
 882             finally:
 883                 loop.quit()
 884
 885         self.stream.flush_async(callback)
 886
 887         loop = glib.MainLoop()
 888         loop.run()
 889
 890     def testSpliceAsync(self):
 891         _f = open("stream.txt", "w+")
 892         _f.write("testing")
 893         _f.seek(0)
 894         instream = gio.unix.InputStream(_f.fileno(), False)
 895
 896         def callback(stream, result):
 897             try:
 898                 size = stream.splice_finish(result)
 899                 self.assertEqual(size, 7)
 900
 901             finally:
 902                 os.unlink("stream.txt")
 903                 loop.quit()
 904
 905         self.stream.splice_async(instream, callback)
 906
 907         loop = glib.MainLoop()
 908         loop.run()
 909
 910 class TestMemoryOutputStream(unittest.TestCase):
 911     def setUp(self):
 912         self.stream = gio.MemoryOutputStream()
 913
 914     def test_get_contents(self):
 915         self.stream.write('foobar')
 916         self.assertEquals('foobar', self.stream.get_contents())
 917
 918         self.stream.write('baz')
 919         self.assertEquals('foobarbaz', self.stream.get_contents())
 920
 921
 922 class TestVolumeMonitor(unittest.TestCase):
 923     def setUp(self):
 924         self.monitor = gio.volume_monitor_get()
 925
 926     def testGetConnectedDrives(self):
 927         drives = self.monitor.get_connected_drives()
 928         self.failUnless(isinstance(drives, list))
 929
 930     def testGetVolumes(self):
 931         volumes = self.monitor.get_volumes()
 932         self.failUnless(isinstance(volumes, list))
 933
 934     def testGetMounts(self):
 935         mounts = self.monitor.get_mounts()
 936         self.failUnless(isinstance(mounts, list))
 937         if not mounts:
 938             return
 939
 940         self.failUnless(isinstance(mounts[0], gio.Mount))
 941         # Bug 538601
 942         icon = mounts[0].get_icon()
 943         if not icon:
 944             return
 945         self.failUnless(isinstance(icon, gio.Icon))
 946
 947
 948 class TestContentTypeGuess(unittest.TestCase):
 949     def testFromName(self):
 950         mime_type = gio.content_type_guess('diagram.svg')
 951         self.assertEquals('image/svg+xml', mime_type)
 952
 953     def testFromContents(self):
 954         mime_type = gio.content_type_guess(data='<html></html>')
 955         self.assertEquals('text/html', mime_type)
 956
 957     def testFromContentsUncertain(self):
 958         mime_type, result_uncertain = gio.content_type_guess(
 959             data='<html></html>', want_uncertain=True)
 960         self.assertEquals('text/html', mime_type)
 961         self.assertEquals(bool, type(result_uncertain))
 962
 963
 964 class TestFileInfo(unittest.TestCase):
 965     def setUp(self):
 966         self.fileinfo = gio.File(__file__).query_info("*")
 967
 968     def testListAttributes(self):
 969         attributes = self.fileinfo.list_attributes("standard")
 970         self.failUnless(attributes)
 971         self.failUnless('standard::name' in attributes)
 972
 973     def testGetModificationTime(self):
 974         mtime = self.fileinfo.get_modification_time()
 975         self.assertEqual(type(mtime), float)
 976
 977     def testSetModificationTime(self):
 978         self.fileinfo.set_modification_time(1000)
 979         mtime = self.fileinfo.get_modification_time()
 980         self.assertEqual(mtime, 1000)
 981
 982
 983 class TestAppInfo(unittest.TestCase):
 984     def setUp(self):
 985         self.appinfo = gio.AppInfo("does-not-exist")
 986
 987     def testSimple(self):
 988         self.assertEquals(self.appinfo.get_description(),
 989                           "Custom definition for does-not-exist")
 990
 991     def test_eq(self):
 992         info1 = gio.app_info_get_all()[0]
 993         info2 = info1.dup()
 994         self.assert_(info1 is not info2)
 995         self.assertEquals(info1, info2)
 996
 997         self.assertNotEqual(gio.app_info_get_all()[0], gio.app_info_get_all()[1])
 998
 999 class TestVfs(unittest.TestCase):
1000     def setUp(self):
1001         self.vfs = gio.vfs_get_default()
1002
1003     def testGetSupportedURISchemes(self):
1004         result = self.vfs.get_supported_uri_schemes()
1005         self.failUnless(type(result), [])
1006
1007 class TestVolume(unittest.TestCase):
1008     def setUp(self):
1009         self.monitor = gio.volume_monitor_get()
1010
1011     def testVolumeEnumerate(self):
1012         volumes = self.monitor.get_volumes()
1013         self.failUnless(isinstance(volumes, list))
1014         for v in volumes:
1015             if v is not None:
1016                 ids = v.enumerate_identifiers()
1017                 self.failUnless(isinstance(ids, list))
1018                 for id in ids:
1019                     if id is not None:
1020                         self.failUnless(isinstance(id, str))
1021
1022 class TestFileInputStream(unittest.TestCase):
1023     def setUp(self):
1024         self._f = open("file.txt", "w+")
1025         self._f.write("testing")
1026         self._f.seek(0)
1027         self.file = gio.File("file.txt")
1028
1029     def tearDown(self):
1030         self._f.close()
1031         if os.path.exists('file.txt'):
1032             os.unlink("file.txt")
1033
1034     def testQueryInfoAsync(self):
1035         def callback(stream, result):
1036             try:
1037                 info = stream.query_info_finish(result)
1038                 self.failUnless(isinstance(info, gio.FileInfo))
1039                 self.failUnless(info.get_attribute_uint64("standard::size"), 7)
1040             finally:
1041                 loop.quit()
1042
1043         inputstream = self.file.read()
1044         inputstream.query_info_async("standard", callback)
1045
1046         loop = glib.MainLoop()
1047         loop.run()
1048
1049 class TestFileOutputStream(unittest.TestCase):
1050     def setUp(self):
1051         self._f = open("file.txt", "w+")
1052         self._f.write("testing")
1053         self._f.seek(0)
1054         self.file = gio.File("file.txt")
1055
1056     def tearDown(self):
1057         self._f.close()
1058         if os.path.exists('file.txt'):
1059             os.unlink("file.txt")
1060
1061     def testQueryInfoAsync(self):
1062         def callback(stream, result):
1063             try:
1064                 info = stream.query_info_finish(result)
1065                 self.failUnless(isinstance(info, gio.FileInfo))
1066                 self.failUnless(info.get_attribute_uint64("standard::size"), 7)
1067             finally:
1068                 loop.quit()
1069
1070         outputstream = self.file.append_to()
1071         outputstream.query_info_async("standard", callback)
1072
1073         loop = glib.MainLoop()
1074         loop.run()
1075
1076 class TestBufferedInputStream(unittest.TestCase):
1077     def setUp(self):
1078         self._f = open("buffer.txt", "w+")
1079         self._f.write("testing")
1080         self._f.seek(0)
1081         stream = gio.unix.InputStream(self._f.fileno(), False)
1082         self.buffered = gio.BufferedInputStream(stream)
1083
1084     def tearDown(self):
1085         self._f.close()
1086         os.unlink("buffer.txt")
1087
1088     def test_fill_async(self):
1089         def callback(stream, result):
1090             try:
1091                 size = stream.fill_finish(result)
1092                 self.failUnlessEqual(size, 4)
1093             finally:
1094                 loop.quit()
1095
1096         self.buffered.fill_async(4, callback)
1097
1098         loop = glib.MainLoop()
1099         loop.run()
1100
1101 class TestIOStream(unittest.TestCase):
1102     def setUp(self):
1103         self.file = gio.File("file.txt")
1104         self.iofile = self.file.create_readwrite(gio.FILE_CREATE_NONE)
1105
1106     def tearDown(self):
1107         if os.path.exists('file.txt'):
1108             os.unlink("file.txt")
1109
1110     def testIOStreamCloseAsync(self):
1111         def callback(stream, result):
1112             try:
1113                 self.failUnless(stream.close_finish(result))
1114             finally:
1115                 loop.quit()
1116
1117         self.iofile.close_async(callback)
1118
1119         loop = glib.MainLoop()
1120         loop.run()
1121
1122
1123     def testQueryInfoAsync(self):
1124         def callback(stream, result):
1125             try:
1126                 info = stream.query_info_finish(result)
1127                 self.failUnless(isinstance(info, gio.FileInfo))
1128                 self.failUnless(info.get_attribute_uint64("standard::size"), 7)
1129             finally:
1130                 loop.quit()
1131
1132         ostream = self.iofile.get_output_stream()
1133         ostream.write("testing")
1134
1135         self.iofile.query_info_async("standard", callback)
1136
1137         loop = glib.MainLoop()
1138         loop.run()