Commit d97662ad authored by Thibault Saunier's avatar Thibault Saunier 🌵

test: Bring back the testsuite and test if the initialization override works

Summary:
Simplify the Makefile taking example on pitivi and copy several pitivi
testing files, simplifying them a bit for our use case

Reviewers: Mathieu_Du

Differential Revision: http://phabricator.freedesktop.org/D97
parent 6b32ccbb
SUBDIRS = common gi plugin
SUBDIRS = common gi plugin testsuite
# Examples and testsuite need to be ported to 1.0
#examples testsuite
......
......@@ -152,5 +152,6 @@ AC_OUTPUT([
common/m4/Makefile
gi/Makefile
plugin/Makefile
testsuite/Makefile
gi/overrides/Makefile
])
# This is a hack to make sure a shared library is built
tests = \
test_adapter.py \
test_audio.py \
test_bin.py \
test_buffer.py \
test_caps.py \
test_element.py \
test_event.py \
test_ghostpad.py \
test_interface.py \
test_message.py \
test_pad.py \
test_pipeline.py \
test_registry.py \
test_struct.py \
test_segment.py \
test_taglist.py \
test_xml.py \
test_pbutils.py \
# Don't try to use wildcards to replace the list of tests below.
# http://www.gnu.org/software/automake/manual/automake.html#Wildcards
# Keep this list sorted!
tests = \
test_gst.py
EXTRA_DIST = $(tests) common.py cleanup.py runtests.py test-object.h python.supp gstpython.supp
EXTRA_DIST = \
__init__.py \
common.py \
runtests.py \
overrides_hack.py \
$(tests)
if HAVE_VALGRIND
check-valgrind:
$(PYTHON) cleanup.py
make valgrind
else
check-valgrind:
@true
endif
clean-local:
rm -rf *.pyc *.pyo
GSTSUPP = $(top_srcdir)/common/gst.supp
PYTHONSUPP = $(top_srcdir)/testsuite/python.supp
GSTPYTHONSUPP = $(top_srcdir)/testsuite/gstpython.supp
TESTS_ENVIRONMENT = LC_ALL=C
# gdb any given test_x.py by running make test_x.py.gdb
%.gdb: %
$(TESTS_ENVIRONMENT) \
gdb --args \
$(PYTHON) \
$*
# valgrind any given test_x.py by running make test_x.py.valgrind
%.valgrind: %
$(TESTS_ENVIRONMENT) \
G_DEBUG=gc-friendly G_SLICE=always-malloc \
$(VALGRIND_PATH) -q \
--suppressions=$(GSTSUPP) \
--suppressions=$(PYTHONSUPP) \
--suppressions=$(GSTPYTHONSUPP) \
--tool=memcheck --leak-check=full --trace-children=yes \
--leak-resolution=high --num-callers=50 \
$(PYTHON) \
$* 2>&1 | tee valgrind.log
@if grep "tely lost" valgrind.log; then \
rm valgrind.log; \
exit 1; \
fi
@rm valgrind.log
# valgrind any given test_x.py by running make test_x.py.valgrind
%.gen-suppressions: %
$(TESTS_ENVIRONMENT) \
G_DEBUG=gc-friendly G_SLICE=always-malloc \
$(VALGRIND_PATH) -q \
--suppressions=$(GSTSUPP) \
--suppressions=$(PYTHONSUPP) \
--suppressions=$(GSTPYTHONSUPP) \
--tool=memcheck --leak-check=full --trace-children=yes \
--leak-resolution=high --num-callers=50 \
--gen-suppressions=all \
$(PYTHON) \
$* 2>&1 | tee valgrind.log
@if grep "tely lost" valgrind.log; then \
rm valgrind.log; \
exit 1; \
fi
@rm valgrind.log
check-local:
$(PYTHON) $(srcdir)/runtests.py $(tests)
%.check: %
$(PYTHON) $(srcdir)/cleanup.py
$(PYTHON) $(srcdir)/runtests.py $*
@rm -fr *.pyc
%.forever: %
$(srcdir)/cleanup.py
@while true; do \
......
# -*- Mode: Python -*-
# vi:si:et:sw=4:sts=4:ts=4
# -*- Mode: Python; py-indent-offset: 4 -*-
# vim: tabstop=4 shiftwidth=4 expandtab
#
# gst-python - Python bindings for GStreamer
# Copyright (C) 2002 David I. Lehn
# Copyright (C) 2004 Johan Dahlin
# Copyright (C) 2005 Edward Hervey
# Copyright (C) 2015 Thibault Saunier <thibault.saunier@collabora.com>
#
# This library is free software; you can redistribute it and/or
# This program 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,
# This program 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 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
# License along with this program; if not, write to the
# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
# Boston, MA 02110-1301, USA.
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3, or (at your option)
# any later version.
"""
A collection of objects to use for testing
Copyied from pitivi
"""
import os
import sys
import gc
import unittest
import gi.overrides
gi.overrides
from gi.repository import Gst
detect_leaks = os.environ.get("TEST_DETECT_LEAKS", "1") not in ("0", "")
import pygtk
pygtk.require('2.0')
import gobject
try:
gobject.threads_init()
except:
print "WARNING: gobject doesn't have threads_init, no threadsafety"
# Detect the version of pygobject
# In pygobject >= 2.13.0 the refcounting of objects has changed.
pgmaj,pgmin,pgmac = gobject.pygobject_version
if pgmaj >= 2 and pgmin >= 13:
pygobject_2_13 = True
else:
pygobject_2_13 = False
# 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 topbuilddir/gst
path = os.path.abspath(os.path.join(topbuilddir, 'gst'))
import gst
file = gst.__file__
assert file.startswith(path), 'bad gst path: %s' % file
# gst's interfaces is 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
# gst's tags is in topbuilddir/gst
path = os.path.abspath(os.path.join(topbuilddir, 'gst'))
try:
import gst.tag
except ImportError:
# hack: we import it from our builddir/gst/.libs instead; ugly
import tag
gst.tag = tag
file = gst.tag.__file__
assert file.startswith(path), 'bad gst.tag path: %s' % file
# gst's pbutils is in topbuilddir/gst
path = os.path.abspath(os.path.join(topbuilddir, 'gst'))
try:
import gst.pbutils
except ImportError:
# hack: we import it from our builddir/gst/.libs instead; ugly
import pbutils
gst.pbutils = pbutils
file = gst.pbutils.__file__
assert file.startswith(path), 'bad gst.pbutils path: %s' % file
# testhelper needs gstlibtoolimporter
import gstlibtoolimporter
gstlibtoolimporter.install()
import testhelper
gstlibtoolimporter.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):
_tracked_types = (Gst.MiniObject, Gst.Element, Gst.Pad, Gst.Caps)
def gctrack(self):
self.gccollect()
self._tracked = []
for obj in gc.get_objects():
if not isinstance(obj, self._tracked_types):
continue
_types = [gst.Object, gst.MiniObject]
self._tracked.append(obj)
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)
if c == 0:
break
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)
def gcverify(self):
leaked = []
for obj in gc.get_objects():
if not isinstance(obj, self._tracked_types) or \
obj in self._tracked:
continue
leaked.append(obj)
# we collect again here to get rid of temporary objects created in the
# above loop
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])
for elt in leaked:
print(elt)
for i in gc.get_referrers(elt):
print(" ", i)
self.assertFalse(leaked, leaked)
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()
self._num_failures = len(getattr(self._result, 'failures', []))
self._num_errors = len(getattr(self._result, 'errors', []))
if detect_leaks:
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()
# don't barf gc info all over the console if we have already failed a
# test case
if (self._num_failures < len(getattr(self._result, 'failures', []))
or self._num_errors < len(getattr(self._result, 'failures', []))):
return
if detect_leaks:
self.gccollect()
self.gcverify()
# override run() to save a reference to the test result object
def run(self, result=None):
if not result:
result = self.defaultTestResult()
self._result = result
unittest.TestCase.run(self, result)
class SignalMonitor(object):
def __init__(self, obj, *signals):
self.signals = signals
self.connectToObj(obj)
def connectToObj(self, obj):
self.obj = obj
for signal in self.signals:
obj.connect(signal, self._signalCb, signal)
setattr(self, self._getSignalCounterName(signal), 0)
setattr(self, self._getSignalCollectName(signal), [])
def disconnectFromObj(self, obj):
obj.disconnect_by_func(self._signalCb)
del self.obj
def _getSignalCounterName(self, signal):
field = '%s_count' % signal.replace('-', '_')
return field
def _getSignalCollectName(self, signal):
field = '%s_collect' % signal.replace('-', '_')
return field
def _signalCb(self, obj, *args):
name = args[-1]
field = self._getSignalCounterName(name)
setattr(self, field, getattr(self, field, 0) + 1)
field = self._getSignalCollectName(name)
setattr(self, field, getattr(self, field, []) + [args[:-1]])
import gi.overrides
if not gi.overrides.__path__[0].endswith("gst-python/gi/overrides"):
local_overrides = None
# our overrides don't take precedence, let's fix it
for i, path in enumerate(gi.overrides.__path__):
if path.endswith("gst-python/gi/overrides"):
local_overrides = path
gi.overrides.__path__.remove(local_overrides)
gi.overrides.__path__.insert(0, local_overrides)
......@@ -21,40 +21,50 @@
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
import glob
import os
import sys
import unittest
SKIP_FILES = ['common', 'runtests']
os.environ['LC_MESSAGES'] = 'C'
def gettestnames(which):
if not which:
dir = os.path.split(os.path.abspath(__file__))[0]
which = [os.path.basename(p) for p in glob.glob('%s/test_*.py' % dir)]
names = map(lambda x: x[:-3], which)
for f in SKIP_FILES:
if f in names:
names.remove(f)
return names
suite = unittest.TestSuite()
loader = unittest.TestLoader()
for name in gettestnames(sys.argv[1:]):
suite.addTest(loader.loadTestsFromName(name))
descriptions = 1
verbosity = 1
if os.environ.has_key('VERBOSE'):
descriptions = 2
verbosity = 2
testRunner = unittest.TextTestRunner(descriptions=descriptions,
verbosity=verbosity)
result = testRunner.run(suite)
if result.failures or result.errors:
sys.exit(1)
def _testcases(filenames):
"""Yield testcases out of filenames."""
for filename in filenames:
if filename.endswith(".py"):
yield filename[:-3]
def _tests_suite():
"""Pick which tests to run."""
testcase = os.getenv("TESTCASE")
if testcase:
testcases = [testcase]
else:
testcases = _testcases(sys.argv[1:])
loader = unittest.TestLoader()
return loader.loadTestsFromNames(testcases)
def setup():
return
if __name__ == "__main__":
setup()
# Set verbosity.
descriptions = 1
verbosity = 1
if 'VERBOSE' in os.environ:
descriptions = 2
verbosity = 2
suite = _tests_suite()
if not list(suite):
raise Exception("No tests found")
# Run the tests.
testRunner = unittest.TextTestRunner(descriptions=descriptions,
verbosity=verbosity)
result = testRunner.run(suite)
if result.failures or result.errors:
sys.exit(1)
......@@ -17,20 +17,53 @@
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
from common import gst, TestCase, unittest
import overrides_hack
overrides_hack
from common import TestCase, unittest
from gi.repository import Gst
class TimeArgsTest(TestCase):
def testNoneTime(self):
self.assertRaises(TypeError, gst.TIME_ARGS, None)
self.assertRaises(TypeError, Gst.TIME_ARGS, None)
def testStringTime(self):
self.assertRaises(TypeError, gst.TIME_ARGS, "String")
self.assertRaises(TypeError, Gst.TIME_ARGS, "String")
def testClockTimeNone(self):
self.assertEquals(gst.TIME_ARGS(gst.CLOCK_TIME_NONE), 'CLOCK_TIME_NONE')
self.assertEquals(Gst.TIME_ARGS(Gst.CLOCK_TIME_NONE), 'CLOCK_TIME_NONE')
def testOneSecond(self):
self.assertEquals(gst.TIME_ARGS(gst.SECOND), '0:00:01.000000000')
self.assertEquals(Gst.TIME_ARGS(Gst.SECOND), '0:00:01.000000000')
class TestNotInitialized(TestCase):
def testNotInitialized(self):
with self.assertRaises(Gst.NotInitalized):
Gst.Caps.from_string("audio/x-raw")
with self.assertRaises(Gst.NotInitalized):
Gst.Structure.from_string("audio/x-raw")
with self.assertRaises(Gst.NotInitalized):
Gst.ElementFactory.make("identity", None)
def testNotDeinitialized(self):
Gst.init(None)
assert(Gst.Caps.from_string("audio/x-raw"))
assert(Gst.Structure.from_string("audio/x-raw"))
assert(Gst.ElementFactory.make("identity", None))
Gst.deinit()
with self.assertRaises(Gst.NotInitalized):
Gst.Caps.from_string("audio/x-raw")
with self.assertRaises(Gst.NotInitalized):
Gst.Structure.from_string("audio/x-raw")
with self.assertRaises(Gst.NotInitalized):
Gst.ElementFactory.make("identity", None)
if __name__ == "__main__":
unittest.main()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment