python: Add overrides for Buffer/Query/Event/Context to handle writability

And make them look more like proper MiniObject

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/9027>

subprojects/gst-python/gi/overrides/Gst.py

@@ -110,6 +110,13 @@ Bin = override(Bin)
 __all__.append('Bin')
 
 
+class NotWritableMiniObject(Exception):
+    pass
+
+
+__all__.append('NotWritableMiniObject')
+
+
 class MiniObject:
     def make_writable(self):
         return _gi_gst.mini_object_make_writable(self)
@@ -117,6 +124,71 @@ class MiniObject:
     def is_writable(self):
         return _gi_gst.mini_object_is_writable(self)
 
+    def flags(self):
+        return _gi_gst.mini_object_flags(self)
+
+    def set_flags(self, flags):
+        _gi_gst.mini_object_set_flags(self, flags)
+
+
+class NotWritableQuery(Exception):
+    pass
+
+
+__all__.append('NotWritableQuery')
+
+
+class Query(Gst.Query, MiniObject):
+    def get_structure(self):
+        return StructureWrapper(_gi_gst.query_get_structure(self), self,
+                                False)
+
+    def writable_structure(self):
+        return StructureWrapper(_gi_gst.query_writable_structure(self), self, True)
+
+
+Query = override(Query)
+__all__.append('Query')
+
+
+class NotWritableEvent(Exception):
+    pass
+
+
+__all__.append('NotWritableEvent')
+
+
+class Event(Gst.Event, MiniObject):
+    def get_structure(self):
+        return StructureWrapper(_gi_gst.event_get_structure(self), self,
+                                False)
+
+    def writable_structure(self):
+        return StructureWrapper(_gi_gst.event_writable_structure(self), self, True)
+
+
+Event = override(Event)
+__all__.append('Event')
+
+class NotWritableContext(Exception):
+    pass
+
+
+__all__.append('NotWritableContext')
+
+
+class Context(Gst.Context, MiniObject):
+    def get_structure(self):
+        return StructureWrapper(_gi_gst.context_get_structure(self), self,
+                                False)
+
+    def writable_structure(self):
+        return StructureWrapper(_gi_gst.context_writable_structure(self), self, True)
+
+
+Context = override(Context)
+__all__.append('Context')
+
 
 class NotWritableCaps(Exception):
     pass
@@ -826,7 +898,42 @@ class MapInfo:
 __all__.append("MapInfo")
 
 
-class Buffer(Gst.Buffer):
+class Buffer(Gst.Buffer, MiniObject):
+    def flags(self):
+        return MiniObject.flags(self)
+
+    def set_flags(self, flags):
+        return MiniObject.set_flags(self, flags)
+
+    def dts(self):
+        return _gi_gst.buffer_get_dts(self)
+
+    def set_dts(self, dts):
+        _gi_gst.buffer_set_dts(self, dts)
+
+    def pts(self):
+        return _gi_gst.buffer_get_pts(self)
+
+    def set_pts(self, pts):
+        _gi_gst.buffer_set_pts(self, pts)
+
+    def set_duration(self, duration):
+        _gi_gst.buffer_set_duration(self, duration)
+
+    def duration(self):
+        return _gi_gst.buffer_get_duration(self)
+
+    def offset(self):
+        return _gi_gst.buffer_get_offset(self)
+
+    def set_offset(self, offset):
+        _gi_gst.buffer_set_offset(self, offset)
+
+    def offset_end(self):
+        return _gi_gst.buffer_get_offset_end(self)
+
+    def set_offset_end(self, offset_end):
+        _gi_gst.buffer_set_offset_end(self, offset_end)
 
     def map_range(self, idx, length, flags):
         mapinfo = MapInfo()

subprojects/gst-python/gi/overrides/gstmodule.c

@@ -965,6 +965,67 @@ _gst_mini_object_make_writable (PyObject * self, PyObject * args)
   return res;
 }
 
+static PyObject *
+_gst_mini_object_flags (PyObject * self, PyObject * args)
+{
+  PyObject *py_miniobj;
+  GstMiniObject *mini_object;
+
+  py_miniobj = PyTuple_GetItem (args, 0);
+  if (py_miniobj == NULL) {
+    PyErr_SetString (PyExc_TypeError, "Expected a PyGObject");
+    return NULL;
+  }
+
+  mini_object = GST_MINI_OBJECT (pygobject_get (py_miniobj));
+  return pyg_flags_from_gtype (gst_mini_object_flags_get_type (),
+      mini_object->flags);
+}
+
+
+static PyObject *
+_gst_mini_object_set_flags (PyObject * self, PyObject * args)
+{
+  PyObject *py_miniobj, *py_flags;
+  GstMiniObject *mini_object;
+
+  py_miniobj = PyTuple_GetItem (args, 0);
+  if (py_miniobj == NULL) {
+    PyErr_SetString (PyExc_TypeError, "Expected a PyGObject");
+    return NULL;
+  }
+
+  mini_object = GST_MINI_OBJECT (pygobject_get (py_miniobj));
+  if (!gst_mini_object_is_writable (mini_object)) {
+
+    PyObject *gstmodule = PyImport_ImportModule ("gi.repository.Gst");
+    PyObject *exc_class =
+        PyObject_GetAttrString (gstmodule, "NotWritableMiniObject");
+    PyObject *args_tuple =
+        PyUnicode_FromFormat ("Trying to set flags on immutable `%s`",
+        g_type_name (GST_MINI_OBJECT_TYPE (mini_object)));
+    PyObject *exc_instance = PyObject_Call (exc_class, args_tuple, NULL);
+
+    PyErr_SetObject (exc_class, exc_instance);
+    Py_DECREF (exc_instance);
+
+    return NULL;
+  }
+
+  py_flags = PyTuple_GetItem (args, 1);
+  if (py_flags == NULL) {
+    PyErr_SetString (PyExc_TypeError, "Expected a flags");
+    return NULL;
+  }
+
+  GstMiniObjectFlags flags;
+  pyg_flags_get_value (gst_mini_object_flags_get_type (), py_flags, &flags);
+
+  mini_object->flags = flags;
+
+  Py_RETURN_NONE;
+}
+
 static PyObject *
 _gst_mini_object_is_writable (PyObject * self, PyObject * args)
 {
@@ -977,7 +1038,6 @@ _gst_mini_object_is_writable (PyObject * self, PyObject * args)
     return NULL;
   }
 
-  /* Extract GstCaps from Gst.Caps parameter */
   mini_object = GST_MINI_OBJECT (pygobject_get (py_miniobj));
   if (gst_mini_object_is_writable (mini_object)) {
     Py_INCREF (Py_True);
@@ -1003,10 +1063,8 @@ _gst_caps_get_structure (PyObject * self, PyObject * args)
   if (!PyArg_ParseTuple (args, "O!i", gst_caps_type, &py_caps, &idx))
     return NULL;
 
-  /* Extract GstCaps from Gst.Caps parameter */
   caps = GST_CAPS (pygobject_get (py_caps));
 
-  /* Get the structure at the given index */
   py_structure = pyg_boxed_new (_gst_structure_type,
       gst_caps_get_structure (caps, idx), FALSE, FALSE);
 
@@ -1031,8 +1089,9 @@ _gst_caps_writable_structure (PyObject * self, PyObject * args)
   if (!gst_caps_is_writable (caps)) {
     PyObject *gstmodule = PyImport_ImportModule ("gi.repository.Gst");
     PyObject *exc_class = PyObject_GetAttrString (gstmodule, "NotWritableCaps");
-    PyObject *args_tuple = Py_BuildValue ("(s)",
-        "Trying to get writable structure from immutable caps");
+    PyObject *args_tuple =
+        PyUnicode_FromFormat ("Trying to write structure on immutable `%s`",
+        g_type_name (GST_MINI_OBJECT_TYPE (caps)));
     PyObject *exc_instance = PyObject_Call (exc_class, args_tuple, NULL);
 
     PyErr_SetObject (exc_class, exc_instance);
@@ -1048,6 +1107,176 @@ _gst_caps_writable_structure (PyObject * self, PyObject * args)
   return py_structure;
 }
 
+static PyObject *
+_gst_query_get_structure (PyObject * self, PyObject * args)
+{
+  PyTypeObject *gst_query_type;
+  PyObject *py_query, *py_structure;
+  GstQuery *query;
+
+  /* Look up Gst.Query and Gst.Structure parameters */
+  gst_query_type = pygobject_lookup_class (_gst_query_type);
+  if (!PyArg_ParseTuple (args, "O!", gst_query_type, &py_query))
+    return NULL;
+
+  /* Extract GstQuery from Gst.Query parameter */
+  query = GST_QUERY (pygobject_get (py_query));
+
+  /* Get the structure at the given index */
+  py_structure = pyg_boxed_new (_gst_structure_type,
+      (GstStructure *) gst_query_get_structure (query), FALSE, FALSE);
+
+  return py_structure;
+}
+
+static PyObject *
+_gst_query_writable_structure (PyObject * self, PyObject * args)
+{
+  PyTypeObject *gst_query_type;
+  PyObject *py_query, *py_structure;
+  GstQuery *query;
+
+  /* Look up Gst.Query and Gst.Structure parameters */
+  gst_query_type = pygobject_lookup_class (_gst_query_type);
+  if (!PyArg_ParseTuple (args, "O!", gst_query_type, &py_query))
+    return NULL;
+
+  /* Extract GstQuery from Gst.Query parameter */
+  query = GST_QUERY (pygobject_get (py_query));
+  if (!gst_query_is_writable (query)) {
+    PyObject *gstmodule = PyImport_ImportModule ("gi.repository.Gst");
+    PyObject *exc_class =
+        PyObject_GetAttrString (gstmodule, "NotWritableQuery");
+    PyObject *args_tuple = Py_BuildValue ("(s)",
+        "Trying to get writable structure from immutable query");
+    PyObject *exc_instance = PyObject_Call (exc_class, args_tuple, NULL);
+
+    PyErr_SetObject (exc_class, exc_instance);
+    Py_DECREF (exc_instance);
+
+    return NULL;
+  }
+
+  /* Get the structure at the given index */
+  py_structure = pyg_boxed_new (_gst_structure_type,
+      gst_query_writable_structure (query), FALSE, FALSE);
+
+  return py_structure;
+}
+
+static PyObject *
+_gst_event_get_structure (PyObject * self, PyObject * args)
+{
+  PyTypeObject *gst_event_type;
+  PyObject *py_event, *py_structure;
+  GstEvent *event;
+
+  /* Look up Gst.Event and Gst.Structure parameters */
+  gst_event_type = pygobject_lookup_class (_gst_event_type);
+  if (!PyArg_ParseTuple (args, "O!", gst_event_type, &py_event))
+    return NULL;
+
+  /* Extract GstEvent from Gst.Event parameter */
+  event = GST_EVENT (pygobject_get (py_event));
+
+  /* Get the structure at the given index */
+  py_structure = pyg_boxed_new (_gst_structure_type,
+      (GstStructure *) gst_event_get_structure (event), FALSE, FALSE);
+
+  return py_structure;
+}
+
+static PyObject *
+_gst_event_writable_structure (PyObject * self, PyObject * args)
+{
+  PyTypeObject *gst_event_type;
+  PyObject *py_event, *py_structure;
+  GstEvent *event;
+
+  /* Look up Gst.Event and Gst.Structure parameters */
+  gst_event_type = pygobject_lookup_class (_gst_event_type);
+  if (!PyArg_ParseTuple (args, "O!", gst_event_type, &py_event))
+    return NULL;
+
+  /* Extract GstEvent from Gst.Event parameter */
+  event = GST_EVENT (pygobject_get (py_event));
+  if (!gst_event_is_writable (event)) {
+    PyObject *gstmodule = PyImport_ImportModule ("gi.repository.Gst");
+    PyObject *exc_class =
+        PyObject_GetAttrString (gstmodule, "NotWritableEvent");
+    PyObject *args_tuple = Py_BuildValue ("(s)",
+        "Trying to get writable structure from immutable event");
+    PyObject *exc_instance = PyObject_Call (exc_class, args_tuple, NULL);
+
+    PyErr_SetObject (exc_class, exc_instance);
+    Py_DECREF (exc_instance);
+
+    return NULL;
+  }
+
+  /* Get the structure at the given index */
+  py_structure = pyg_boxed_new (_gst_structure_type,
+      gst_event_writable_structure (event), FALSE, FALSE);
+
+  return py_structure;
+}
+
+static PyObject *
+_gst_context_get_structure (PyObject * self, PyObject * args)
+{
+  PyTypeObject *gst_context_type;
+  PyObject *py_context, *py_structure;
+  GstContext *context;
+
+  /* Look up Gst.Context and Gst.Structure parameters */
+  gst_context_type = pygobject_lookup_class (_gst_context_type);
+  if (!PyArg_ParseTuple (args, "O!", gst_context_type, &py_context))
+    return NULL;
+
+  /* Extract GstContext from Gst.Context parameter */
+  context = GST_CONTEXT (pygobject_get (py_context));
+
+  /* Get the structure at the given index */
+  py_structure = pyg_boxed_new (_gst_structure_type,
+      (GstStructure *) gst_context_get_structure (context), FALSE, FALSE);
+
+  return py_structure;
+}
+
+static PyObject *
+_gst_context_writable_structure (PyObject * self, PyObject * args)
+{
+  PyTypeObject *gst_context_type;
+  PyObject *py_context, *py_structure;
+  GstContext *context;
+
+  /* Look up Gst.Context and Gst.Structure parameters */
+  gst_context_type = pygobject_lookup_class (_gst_context_type);
+  if (!PyArg_ParseTuple (args, "O!", gst_context_type, &py_context))
+    return NULL;
+
+  context = GST_CONTEXT (pygobject_get (py_context));
+  if (!gst_context_is_writable (context)) {
+    PyObject *gstmodule = PyImport_ImportModule ("gi.repository.Gst");
+    PyObject *exc_class =
+        PyObject_GetAttrString (gstmodule, "NotWritableContext");
+    PyObject *args_tuple = Py_BuildValue ("(s)",
+        "Trying to get writable structure from immutable context");
+    PyObject *exc_instance = PyObject_Call (exc_class, args_tuple, NULL);
+
+    PyErr_SetObject (exc_class, exc_instance);
+    Py_DECREF (exc_instance);
+
+    return NULL;
+  }
+
+  /* Get the structure at the given index */
+  py_structure = pyg_boxed_new (_gst_structure_type,
+      gst_context_writable_structure (context), FALSE, FALSE);
+
+  return py_structure;
+}
+
 static PyObject *
 _gst_memory_override_unmap (PyObject * self, PyObject * args)
 {
@@ -1136,6 +1365,203 @@ err:
   return Py_False;
 }
 
+static GstBuffer *
+get_buffer_from_args (PyObject * args)
+{
+  PyTypeObject *gst_buffer_type;
+  PyObject *py_buffer;
+
+  gst_buffer_type = pygobject_lookup_class (_gst_buffer_type);
+  if (!PyArg_ParseTuple (args, "O!", gst_buffer_type, &py_buffer)) {
+    PyErr_BadArgument ();
+    return NULL;
+  }
+
+  return GST_BUFFER (pygobject_get (py_buffer));
+}
+
+#define DECLARE_BUFFER(args) \
+  GstBuffer *buffer = get_buffer_from_args(args); \
+  if (!buffer) return NULL;
+
+static PyObject *
+_gst_buffer_get_dts (PyObject * self, PyObject * args)
+{
+  DECLARE_BUFFER (args);
+
+  return PyLong_FromUnsignedLong (buffer->dts);
+}
+
+static gboolean
+mini_object_check_writability (GstMiniObject * mini_object, const gchar * field)
+{
+  if (!gst_mini_object_is_writable (mini_object)) {
+
+    PyObject *gstmodule = PyImport_ImportModule ("gi.repository.Gst");
+    PyObject *exc_class =
+        PyObject_GetAttrString (gstmodule, "NotWritableMiniObject");
+    PyObject *args_tuple =
+        PyUnicode_FromFormat ("Trying to set %s on immutable `%s`",
+        field, g_type_name (GST_MINI_OBJECT_TYPE (mini_object)));
+    PyObject *exc_instance = PyObject_Call (exc_class, args_tuple, NULL);
+
+    PyErr_SetObject (exc_class, exc_instance);
+    Py_DECREF (exc_instance);
+
+    return FALSE;
+  }
+
+  return TRUE;
+}
+
+
+static PyObject *
+_gst_buffer_set_dts (PyObject * self, PyObject * args)
+{
+  PyTypeObject *gst_buffer_type;
+  PyObject *py_buffer;
+  GstClockTime dts;
+  GstBuffer *buffer;
+
+  gst_buffer_type = pygobject_lookup_class (_gst_buffer_type);
+  if (!PyArg_ParseTuple (args, "O!K", gst_buffer_type, &py_buffer, &dts)) {
+    PyErr_BadArgument ();
+    return NULL;
+  }
+
+  buffer = GST_BUFFER (pygobject_get (py_buffer));
+  if (!mini_object_check_writability (GST_MINI_OBJECT (buffer), "dts"))
+    return NULL;
+
+  buffer->dts = dts;
+
+  Py_RETURN_NONE;
+}
+
+static PyObject *
+_gst_buffer_get_pts (PyObject * self, PyObject * args)
+{
+  DECLARE_BUFFER (args);
+
+  return PyLong_FromUnsignedLong (buffer->pts);
+}
+
+static PyObject *
+_gst_buffer_set_pts (PyObject * self, PyObject * args)
+{
+  PyTypeObject *gst_buffer_type;
+  PyObject *py_buffer;
+  GstClockTime pts;
+  GstBuffer *buffer;
+
+  gst_buffer_type = pygobject_lookup_class (_gst_buffer_type);
+  if (!PyArg_ParseTuple (args, "O!K", gst_buffer_type, &py_buffer, &pts)) {
+    PyErr_BadArgument ();
+    return NULL;
+  }
+
+  buffer = GST_BUFFER (pygobject_get (py_buffer));
+  if (!mini_object_check_writability (GST_MINI_OBJECT (buffer), "pts"))
+    return NULL;
+
+  buffer->pts = pts;
+
+  Py_RETURN_NONE;
+}
+
+static PyObject *
+_gst_buffer_get_duration (PyObject * self, PyObject * args)
+{
+  DECLARE_BUFFER (args);
+
+  return PyLong_FromUnsignedLong (buffer->duration);
+}
+
+static PyObject *
+_gst_buffer_set_duration (PyObject * self, PyObject * args)
+{
+  PyTypeObject *gst_buffer_type;
+  PyObject *py_buffer;
+  GstClockTime duration;
+  GstBuffer *buffer;
+
+  gst_buffer_type = pygobject_lookup_class (_gst_buffer_type);
+  if (!PyArg_ParseTuple (args, "O!K", gst_buffer_type, &py_buffer, &duration)) {
+    PyErr_BadArgument ();
+    return NULL;
+  }
+
+  buffer = GST_BUFFER (pygobject_get (py_buffer));
+  if (!mini_object_check_writability (GST_MINI_OBJECT (buffer), "duration"))
+    return NULL;
+
+  buffer->duration = duration;
+
+  Py_RETURN_NONE;
+}
+
+static PyObject *
+_gst_buffer_get_offset (PyObject * self, PyObject * args)
+{
+  DECLARE_BUFFER (args);
+
+  return PyLong_FromUnsignedLong (buffer->offset);
+}
+
+static PyObject *
+_gst_buffer_set_offset (PyObject * self, PyObject * args)
+{
+  PyTypeObject *gst_buffer_type;
+  PyObject *py_buffer;
+  GstClockTime offset;
+  GstBuffer *buffer;
+
+  gst_buffer_type = pygobject_lookup_class (_gst_buffer_type);
+  if (!PyArg_ParseTuple (args, "O!K", gst_buffer_type, &py_buffer, &offset)) {
+    PyErr_BadArgument ();
+    return NULL;
+  }
+
+  buffer = GST_BUFFER (pygobject_get (py_buffer));
+  if (!mini_object_check_writability (GST_MINI_OBJECT (buffer), "offset"))
+    return NULL;
+
+  buffer->offset = offset;
+
+  Py_RETURN_NONE;
+}
+
+static PyObject *
+_gst_buffer_get_offset_end (PyObject * self, PyObject * args)
+{
+  DECLARE_BUFFER (args);
+
+  return PyLong_FromUnsignedLong (buffer->offset_end);
+}
+
+static PyObject *
+_gst_buffer_set_offset_end (PyObject * self, PyObject * args)
+{
+  PyTypeObject *gst_buffer_type;
+  PyObject *py_buffer;
+  GstClockTime offset_end;
+  GstBuffer *buffer;
+
+  gst_buffer_type = pygobject_lookup_class (_gst_buffer_type);
+  if (!PyArg_ParseTuple (args, "O!K", gst_buffer_type, &py_buffer, &offset_end)) {
+    PyErr_BadArgument ();
+    return NULL;
+  }
+
+  buffer = GST_BUFFER (pygobject_get (py_buffer));
+  if (!mini_object_check_writability (GST_MINI_OBJECT (buffer), "offset_end"))
+    return NULL;
+
+  buffer->offset_end = offset_end;
+
+  Py_RETURN_NONE;
+}
+
 static PyObject *
 _gst_buffer_override_map (PyObject * self, PyObject * args)
 {
@@ -1229,6 +1655,28 @@ static PyMethodDef _gi_gst_functions[] = {
 
   {"mini_object_make_writable", (PyCFunction) _gst_mini_object_make_writable, METH_VARARGS, NULL},
   {"mini_object_is_writable", (PyCFunction) _gst_mini_object_is_writable, METH_VARARGS, NULL},
+  {"mini_object_flags", (PyCFunction) _gst_mini_object_flags, METH_VARARGS, NULL},
+  {"mini_object_set_flags", (PyCFunction) _gst_mini_object_set_flags, METH_VARARGS, NULL},
+
+  {"event_get_structure", (PyCFunction) _gst_event_get_structure, METH_VARARGS, NULL},
+  {"event_writable_structure", (PyCFunction) _gst_event_writable_structure, METH_VARARGS, NULL},
+
+  {"query_get_structure", (PyCFunction) _gst_query_get_structure, METH_VARARGS, NULL},
+  {"query_writable_structure", (PyCFunction) _gst_query_writable_structure, METH_VARARGS, NULL},
+
+  {"context_get_structure", (PyCFunction) _gst_context_get_structure, METH_VARARGS, NULL},
+  {"context_writable_structure", (PyCFunction) _gst_context_writable_structure, METH_VARARGS, NULL},
+
+  {"buffer_get_pts", (PyCFunction) _gst_buffer_get_pts, METH_VARARGS, NULL},
+  {"buffer_set_pts", (PyCFunction) _gst_buffer_set_pts, METH_VARARGS, NULL},
+  {"buffer_get_duration", (PyCFunction) _gst_buffer_get_duration, METH_VARARGS, NULL},
+  {"buffer_set_duration", (PyCFunction) _gst_buffer_set_duration, METH_VARARGS, NULL},
+  {"buffer_get_dts", (PyCFunction) _gst_buffer_get_dts, METH_VARARGS, NULL},
+  {"buffer_set_dts", (PyCFunction) _gst_buffer_set_dts, METH_VARARGS, NULL},
+  {"buffer_get_offset", (PyCFunction) _gst_buffer_get_offset, METH_VARARGS, NULL},
+  {"buffer_set_offset", (PyCFunction) _gst_buffer_set_offset, METH_VARARGS, NULL},
+  {"buffer_get_offset_end", (PyCFunction) _gst_buffer_get_offset_end, METH_VARARGS, NULL},
+  {"buffer_set_offset_end", (PyCFunction) _gst_buffer_set_offset_end, METH_VARARGS, NULL},
 
   {"_get_object_ptr", (PyCFunction) _gst_get_object_ptr, METH_VARARGS, NULL},
   {NULL, NULL, 0, NULL}

subprojects/gst-python/testsuite/test_gst.py

@@ -191,6 +191,44 @@ class TestStructure(TestCase):
         self.assertEqual(test['test'], 1)
 
 
+class TestEvent(TestCase):
+    def test_writable(self):
+        Gst.init(None)
+        event = Gst.Event.new_caps(Gst.Caps("audio/x-raw"))
+
+        with event.writable_structure() as s:
+            s.set_value("rate", 44100)
+            s.set_value("channels", 2)
+
+        with event.get_structure() as s:
+            self.assertEqual(s["rate"], 44100)
+            self.assertEqual(s["channels"], 2)
+
+
+class TestQuery(TestCase):
+    def test_writable(self):
+        Gst.init(None)
+        query = Gst.Query.new_caps(Gst.Caps("audio/x-raw"))
+
+        with query.writable_structure() as s:
+            del query
+            s.set_value("rate", 44100)
+            s.set_value("channels", 2)
+            self.assertEqual(s["rate"], 44100)
+            self.assertEqual(s["channels"], 2)
+
+class TestContext(TestCase):
+    def test_writable(self):
+        Gst.init(None)
+        context = Gst.Context.new("test", True)
+
+        with context.writable_structure() as s:
+            s.set_value("one", 1)
+            s.set_value("two", 2)
+            self.assertEqual(s["one"], 1)
+            self.assertEqual(s["two"], 2)
+
+
 class TestBin(TestCase):
 
     def test_add_pad(self):
@@ -198,7 +236,33 @@ class TestBin(TestCase):
         self.assertEqual(Gst.ElementFactory.make("bin", None).sinkpads, [])
 
 
-class TestBufferMap(TestCase):
+class TestBuffer(TestCase):
+
+    def test_set_metas(self):
+        Gst.init(None)
+        buf = Gst.Buffer.new_wrapped([42])
+        buf.set_pts(42)
+        self.assertEqual(buf.pts(), 42)
+
+        make_not_writable = buf.mini_object
+        with self.assertRaises(Gst.NotWritableMiniObject):
+            buf.set_pts(52)
+
+        buf.make_writable()
+        buf.set_dts(62)
+        self.assertEqual(buf.dts(), 62)
+        self.assertTrue(buf.flags() & Gst.BufferFlags.DISCONT == 0)
+        buf.set_flags(Gst.BufferFlags.DISCONT)
+        buf.set_duration(72)
+        self.assertEqual(buf.duration(), 72)
+        buf.set_offset(82)
+        self.assertEqual(buf.offset(), 82)
+        buf.set_offset_end(92)
+        self.assertEqual(buf.offset_end(), 92)
+        del make_not_writable
+
+        meta = buf.add_reference_timestamp_meta(Gst.Caps("yes"), 10, 10)
+        self.assertEqual(buf.get_reference_timestamp_meta(), meta)
 
     def test_map_unmap_manual(self):
         Gst.init(None)