1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
|
# -*- Mode: Python -*-
# vi:si:et:sw=4:sts=4:ts=4
#
# gst-python - Python bindings for GStreamer
# Copyright (C) 2002 David I. Lehn
# Copyright (C) 2004 Johan Dahlin
# Copyright (C) 2005 Edward Hervey
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
try:
from dl import RTLD_LAZY, RTLD_GLOBAL
except ImportError:
# dl doesn't seem to be available on 64bit systems
try:
from DLFCN import RTLD_LAZY, RTLD_GLOBAL
except ImportError:
pass
import os
import sys
import gc
import unittest
import pygtk
pygtk.require('2.0')
import gobject
try:
gobject.threads_init()
except:
print "WARNING: gobject doesn't have threads_init, no threadsafety"
# Don't insert before .
# sys.path.insert(1, os.path.join('..'))
# Load GST and make sure we load it from the current build
sys.setdlopenflags(RTLD_LAZY | RTLD_GLOBAL)
topbuilddir = os.path.abspath(os.path.join('..'))
topsrcdir = os.path.abspath(os.path.join('..'))
if topsrcdir.endswith('_build'):
topsrcdir = os.path.dirname(topsrcdir)
# gst's __init__.py is in topsrcdir/gst
path = os.path.abspath(os.path.join(topsrcdir, 'gst'))
import gst
file = gst.__file__
assert file.startswith(path), 'bad gst path: %s' % file
# gst's interfaces and play are in topbuilddir/gst
path = os.path.abspath(os.path.join(topbuilddir, 'gst'))
try:
import gst.interfaces
except ImportError:
# hack: we import it from our builddir/gst/.libs instead; ugly
import interfaces
gst.interfaces = interfaces
file = gst.interfaces.__file__
assert file.startswith(path), 'bad gst.interfaces path: %s' % file
# testhelper needs gstltihooks
import gstltihooks
import testhelper
gstltihooks.uninstall()
_stderr = None
def disable_stderr():
global _stderr
_stderr = file('/tmp/stderr', 'w+')
sys.stderr = os.fdopen(os.dup(2), 'w')
os.close(2)
os.dup(_stderr.fileno())
def enable_stderr():
global _stderr
os.close(2)
os.dup(sys.stderr.fileno())
_stderr.seek(0, 0)
data = _stderr.read()
_stderr.close()
os.remove('/tmp/stderr')
return data
def run_silent(function, *args, **kwargs):
disable_stderr()
try:
function(*args, **kwargs)
except Exception, exc:
enable_stderr()
raise exc
output = enable_stderr()
return output
class TestCase(unittest.TestCase):
_types = [gst.Object, gst.MiniObject]
def gccollect(self):
# run the garbage collector
ret = 0
gst.debug('garbage collecting')
while True:
c = gc.collect()
ret += c
if c == 0: break
gst.debug('done garbage collecting, %d objects' % ret)
return ret
def gctrack(self):
# store all gst objects in the gc in a tracking dict
# call before doing any allocation in your test, from setUp
gst.debug('tracking gc GstObjects for types %r' % self._types)
self.gccollect()
self._tracked = {}
for c in self._types:
self._tracked[c] = [o for o in gc.get_objects() if isinstance(o, c)]
def gcverify(self):
# verify no new gst objects got added to the gc
# call after doing all cleanup in your test, from tearDown
gst.debug('verifying gc GstObjects for types %r' % self._types)
new = []
for c in self._types:
objs = [o for o in gc.get_objects() if isinstance(o, c)]
new.extend([o for o in objs if o not in self._tracked[c]])
self.failIf(new, new)
#self.failIf(new, ["%r:%d" % (type(o), id(o)) for o in new])
del self._tracked
def setUp(self):
"""
Override me by chaining up to me at the start of your setUp.
"""
# Using private variables is BAD ! this variable changed name in
# python 2.5
try:
methodName = self.__testMethodName
except:
methodName = self._testMethodName
gst.debug('%s.%s' % (self.__class__.__name__, methodName))
self.gctrack()
def tearDown(self):
"""
Override me by chaining up to me at the end of your tearDown.
"""
# Using private variables is BAD ! this variable changed name in
# python 2.5
try:
methodName = self.__testMethodName
except:
methodName = self._testMethodName
gst.debug('%s.%s' % (self.__class__.__name__, methodName))
self.gccollect()
self.gcverify()
|
