pygobject r737 - in trunk: . gio tests

[johan svn gnome org]

Author: johan
Date: Sat Jan 19 19:54:46 2008
New Revision: 737
URL: http://svn.gnome.org/viewvc/pygobject?rev=737&view=rev

Log:
2008-01-19  Johan Dahlin  <johan gnome org>

	* gio/gio.override (_wrap_g_output_stream_write): Impl.
	* gio/gio.override:
	* gio/unix.defs:
	Add GUnixInputStream type and methods

	* tests/common.py:
	* tests/test_gio.py:
	Add GIO tests.



Added:
   trunk/tests/test_gio.py   (contents, props changed)
Modified:
   trunk/ChangeLog
   trunk/gio/gio.override
   trunk/gio/unix.defs
   trunk/tests/common.py

Modified: trunk/gio/gio.override
==============================================================================
--- trunk/gio/gio.override	(original)
+++ trunk/gio/gio.override	Sat Jan 19 19:54:46 2008
@@ -265,7 +265,7 @@
 			    "user_data", NULL };
   long count = -1;
   int io_priority = G_PRIORITY_DEFAULT;
-  PyGObject *pycancellable = NULL;
+  PyGObject *pycancellable;
   PyObject *buffer;
   GCancellable *cancellable;
   PyGAsyncRequestNotify *notify;
@@ -316,3 +316,42 @@
   Py_INCREF(Py_None);
   return Py_None;
 }
+%%
+override g_output_stream_write kwargs
+static PyObject *
+_wrap_g_output_stream_write(PyGObject *self, PyObject *args, PyObject *kwargs)
+{
+  static char *kwlist[] = { "buffer", "cancellable", NULL };
+  PyGObject *pycancellable = NULL;
+  gchar *buffer;
+  long count = 0; 
+  GCancellable *cancellable;
+  GError *error = NULL;
+  gssize written;
+  
+  if (!PyArg_ParseTupleAndKeywords(args, kwargs,
+				   "s#|O!:OutputStream.write",
+				   kwlist, &buffer, &count,
+				   &PyGCancellable_Type, &pycancellable))
+    return NULL;
+  
+  if (!pycancellable)
+    cancellable = NULL;
+  else if (pygobject_check(pycancellable, &PyGCancellable_Type))
+      cancellable = G_CANCELLABLE(pycancellable->obj);
+  else
+    {
+      PyErr_SetString(PyExc_TypeError, "cancellable should be a gio.Cancellable");
+      return NULL;
+    }
+
+  pyg_begin_allow_threads;
+  written = g_output_stream_write(G_OUTPUT_STREAM(self->obj),
+				  buffer, count, cancellable, &error);
+  pyg_end_allow_threads;
+
+  if (pyg_error_check(&error))
+    return NULL;
+      
+  return PyInt_FromLong(written);
+}

Modified: trunk/gio/unix.defs
==============================================================================
--- trunk/gio/unix.defs	(original)
+++ trunk/gio/unix.defs	Sat Jan 19 19:54:46 2008
@@ -266,6 +266,24 @@
 
 
 
+;; From gunixinputstream.h
+
+(define-function unix_input_stream_get_type
+  (c-name "g_unix_input_stream_get_type")
+  (return-type "GType")
+)
+
+(define-function unix_input_stream_new
+  (c-name "g_unix_input_stream_new")
+  (is-constructor-of "GUnixInputStream")
+  (return-type "GInputStream*")
+  (parameters
+    '("int" "fd")
+    '("gboolean" "close_fd_at_close")
+  )
+)
+
+
 ;; From gunixoutputstream.h
 
 (define-function unix_output_stream_get_type

Modified: trunk/tests/common.py
==============================================================================
--- trunk/tests/common.py	(original)
+++ trunk/tests/common.py	Sat Jan 19 19:54:46 2008
@@ -9,6 +9,7 @@
     sys.path.insert(0, srcDir)
     sys.path.insert(0, buildDir)
     sys.path.insert(0, os.path.join(buildDir, 'gobject'))
+    sys.path.insert(0, os.path.join(buildDir, 'gio'))
     import ltihooks
 
     # testhelper
@@ -17,6 +18,7 @@
 
     testhelper = importModule('testhelper', '.')
     gobject = importModule('gobject', buildDir, 'gobject')
+    gio = importModule('gio', buildDir, 'gio')
 
     ltihooks.uninstall()
     del ltihooks

Added: trunk/tests/test_gio.py
==============================================================================
--- (empty file)
+++ trunk/tests/test_gio.py	Sat Jan 19 19:54:46 2008
@@ -0,0 +1,49 @@
+# -*- Mode: Python -*-
+
+import os
+import unittest
+
+from common import gio, gobject
+
+
+class TestInputStream(unittest.TestCase):
+    def setUp(self):
+        f = open("inputstream.txt", "w")
+        f.write("testing")
+
+        self._f = open("inputstream.txt", "r")
+        self.stream = gio.unix.InputStream(self._f.fileno(), False)
+
+    def tearDown(self):
+        self._f.close()
+        os.unlink("inputstream.txt")
+
+    def testWrite(self):
+        self.assertEquals(self.stream.read(), "testing")
+
+
+class TestOutputStream(unittest.TestCase):
+    def setUp(self):
+        self._f = open("outputstream.txt", "w")
+        self.stream = gio.unix.OutputStream(self._f.fileno(), False)
+
+    def tearDown(self):
+        self._f.close()
+        os.unlink("outputstream.txt")
+
+    def testWrite(self):
+        self.stream.write("testing")
+        self.stream.close()
+        self.failUnless(os.path.exists("outputstream.txt"))
+        self.assertEquals(open("outputstream.txt").read(), "testing")
+
+    def testWriteAsync(self):
+        def callback(stream, result):
+            loop.quit()
+
+        f = gio.file_new_for_path("outputstream.txt")
+        stream = f.read()
+        stream.read_async(10240, 0, None, callback)
+
+        loop = gobject.MainLoop()
+        loop.run()