[libcamera-devel] [PATCH v5 2/3] py: add unittests.py
Kieran Bingham
kieran.bingham at ideasonboard.com
Mon Mar 14 17:21:30 CET 2022
Hi Tomi,
Quoting Tomi Valkeinen (2022-03-14 15:46:32)
> Add a simple unittests.py as a base for python unittests.
>
> Signed-off-by: Tomi Valkeinen <tomi.valkeinen at ideasonboard.com>
> ---
> src/py/test/unittests.py | 364 +++++++++++++++++++++++++++++++++++++++
Is it feasible to get this to run/integrate with the meson.build test hooks?
It might want to be under test/python/ perhaps then too.
> 1 file changed, 364 insertions(+)
> create mode 100755 src/py/test/unittests.py
>
> diff --git a/src/py/test/unittests.py b/src/py/test/unittests.py
> new file mode 100755
> index 00000000..ca3c37bf
> --- /dev/null
> +++ b/src/py/test/unittests.py
> @@ -0,0 +1,364 @@
> +#!/usr/bin/env python3
> +
> +# SPDX-License-Identifier: GPL-2.0-or-later
> +# Copyright (C) 2021, Tomi Valkeinen <tomi.valkeinen at ideasonboard.com>
> +
> +from collections import defaultdict
> +import errno
> +import gc
> +import libcamera as libcam
> +import os
> +import selectors
> +import time
> +import unittest
> +import weakref
> +
> +class MyTestCase(unittest.TestCase):
> + def assertZero(self, a, msg=None):
> + self.assertEqual(a, 0, msg)
> +
> +class SimpleTestMethods(MyTestCase):
> + def test_find_ref(self):
> + cm = libcam.CameraManager.singleton()
> + wr_cm = weakref.ref(cm)
> +
> + cam = cm.find("platform/vimc")
> + self.assertIsNotNone(cam)
> + wr_cam = weakref.ref(cam)
> +
> + cm = None
> + gc.collect()
> + self.assertIsNotNone(wr_cm())
> +
> + cam = None
> + gc.collect()
> + self.assertIsNone(wr_cm())
> + self.assertIsNone(wr_cam())
> +
> + def test_get_ref(self):
> + cm = libcam.CameraManager.singleton()
> + wr_cm = weakref.ref(cm)
> +
> + cam = cm.get("platform/vimc.0 Sensor B")
> + self.assertTrue(cam != None)
> + wr_cam = weakref.ref(cam)
> +
> + cm = None
> + gc.collect()
> + self.assertIsNotNone(wr_cm())
> +
> + cam = None
> + gc.collect()
> + self.assertIsNone(wr_cm())
> + self.assertIsNone(wr_cam())
> +
> + def test_acquire_release(self):
> + cm = libcam.CameraManager.singleton()
> + cam = cm.get("platform/vimc.0 Sensor B")
> + self.assertTrue(cam != None)
> +
> + ret = cam.acquire()
> + self.assertZero(ret)
> +
> + ret = cam.release()
> + self.assertZero(ret)
> +
> + def test_double_acquire(self):
> + cm = libcam.CameraManager.singleton()
> + cam = cm.get("platform/vimc.0 Sensor B")
> + self.assertTrue(cam != None)
> +
> + ret = cam.acquire()
> + self.assertZero(ret)
> +
> + libcam.logSetLevel("Camera", "FATAL")
> + ret = cam.acquire()
> + self.assertEqual(ret, -errno.EBUSY)
> + libcam.logSetLevel("Camera", "ERROR")
> +
> + ret = cam.release()
> + self.assertZero(ret)
> +
> + ret = cam.release()
> + # I expected EBUSY, but looks like double release works fine
> + self.assertZero(ret)
> +
> +class CameraTesterBase(MyTestCase):
> + def setUp(self):
> + self.cm = libcam.CameraManager.singleton()
> + self.cam = self.cm.find("platform/vimc")
> + if self.cam == None:
> + self.cm = None
> + raise Exception("No vimc found")
> +
> + ret = self.cam.acquire()
> + if ret != 0:
> + self.cam = None
> + self.cm = None
> + raise Exception("Failed to acquire camera")
> +
> + def tearDown(self):
> + # If a test fails, the camera may be in running state. So always stop.
> + self.cam.stop()
> +
> + ret = self.cam.release()
> + if ret != 0:
> + raise Exception("Failed to release camera")
> +
> + self.cam = None
> + self.cm = None
> +
> +
> +class AllocatorTestMethods(CameraTesterBase):
> + def test_allocator(self):
> + cam = self.cam
> +
> + camconfig = cam.generateConfiguration([libcam.StreamRole.StillCapture])
> + self.assertTrue(camconfig.size == 1)
> + wr_camconfig = weakref.ref(camconfig)
> +
> + streamconfig = camconfig.at(0)
> + wr_streamconfig = weakref.ref(streamconfig)
> +
> + ret = cam.configure(camconfig);
> + self.assertZero(ret)
> +
> + stream = streamconfig.stream
> + wr_stream = weakref.ref(stream)
> +
> + # stream should keep streamconfig and camconfig alive
> + streamconfig = None
> + camconfig = None
> + gc.collect()
> + self.assertIsNotNone(wr_camconfig())
> + self.assertIsNotNone(wr_streamconfig())
> +
> + allocator = libcam.FrameBufferAllocator(cam);
> + ret = allocator.allocate(stream)
> + self.assertTrue(ret > 0)
> + wr_allocator = weakref.ref(allocator)
> +
> + buffers = allocator.buffers(stream)
> + buffers = None
> +
> +
> + buffer = allocator.buffers(stream)[0]
> + self.assertIsNotNone(buffer)
> + wr_buffer = weakref.ref(buffer)
> +
> + allocator = None
> + gc.collect()
> + self.assertIsNotNone(wr_buffer())
> + self.assertIsNotNone(wr_allocator())
> + self.assertIsNotNone(wr_stream())
> +
> + buffer = None
> + gc.collect()
> + self.assertIsNone(wr_buffer())
> + self.assertIsNone(wr_allocator())
> + self.assertIsNotNone(wr_stream())
> +
> + stream = None
> + gc.collect()
> + self.assertIsNone(wr_stream())
> + self.assertIsNone(wr_camconfig())
> + self.assertIsNone(wr_streamconfig())
> +
> +
> +class SimpleCaptureMethods(CameraTesterBase):
> + def test_sleep(self):
> + cm = self.cm
> + cam = self.cam
> +
> + camconfig = cam.generateConfiguration([libcam.StreamRole.StillCapture])
> + self.assertTrue(camconfig.size == 1)
> +
> + streamconfig = camconfig.at(0)
> + fmts = streamconfig.formats
> +
> + ret = cam.configure(camconfig);
> + self.assertZero(ret)
> +
> + stream = streamconfig.stream
> +
> + allocator = libcam.FrameBufferAllocator(cam);
> + ret = allocator.allocate(stream)
> + self.assertTrue(ret > 0)
> +
> + num_bufs = len(allocator.buffers(stream))
> +
> + reqs = []
> + for i in range(num_bufs):
> + req = cam.createRequest(i)
> + self.assertIsNotNone(req)
> +
> + buffer = allocator.buffers(stream)[i]
> + ret = req.addBuffer(stream, buffer)
> + self.assertZero(ret)
> +
> + reqs.append(req)
> +
> + buffer = None
> +
> + ret = cam.start()
> + self.assertZero(ret)
> +
> + for req in reqs:
> + ret = cam.queueRequest(req)
> + self.assertZero(ret)
> +
> + reqs = None
> + gc.collect()
> +
> + time.sleep(0.5)
> +
> + reqs = cm.getReadyRequests()
> +
> + self.assertTrue(len(reqs) == num_bufs)
> +
> + for i, req in enumerate(reqs):
> + self.assertTrue(i == req.cookie)
> +
> + reqs = None
> + gc.collect()
> +
> + ret = cam.stop()
> + self.assertZero(ret)
> +
> +
> + def test_select(self):
> + cm = self.cm
> + cam = self.cam
> +
> + camconfig = cam.generateConfiguration([libcam.StreamRole.StillCapture])
> + self.assertTrue(camconfig.size == 1)
> +
> + streamconfig = camconfig.at(0)
> + fmts = streamconfig.formats
> +
> + ret = cam.configure(camconfig);
> + self.assertZero(ret)
> +
> + stream = streamconfig.stream
> +
> + allocator = libcam.FrameBufferAllocator(cam);
> + ret = allocator.allocate(stream)
> + self.assertTrue(ret > 0)
> +
> + num_bufs = len(allocator.buffers(stream))
> +
> + reqs = []
> + for i in range(num_bufs):
> + req = cam.createRequest(i)
> + self.assertIsNotNone(req)
> +
> + buffer = allocator.buffers(stream)[i]
> + ret = req.addBuffer(stream, buffer)
> + self.assertZero(ret)
> +
> + reqs.append(req)
> +
> + buffer = None
> +
> + ret = cam.start()
> + self.assertZero(ret)
> +
> + for req in reqs:
> + ret = cam.queueRequest(req)
> + self.assertZero(ret)
> +
> + reqs = None
> + gc.collect()
> +
> + sel = selectors.DefaultSelector()
> + sel.register(cm.efd, selectors.EVENT_READ, 123)
> +
> + reqs = []
> +
> + running = True
> + while running:
> + events = sel.select()
> + for key, mask in events:
> + os.read(key.fileobj, 8)
> +
> + l = cm.getReadyRequests()
> +
> + self.assertTrue(len(l) > 0)
> +
> + reqs += l
> +
> + if len(reqs) == num_bufs:
> + running = False
> +
> + self.assertTrue(len(reqs) == num_bufs)
> +
> + for i, req in enumerate(reqs):
> + self.assertTrue(i == req.cookie)
> +
> + reqs = None
> + gc.collect()
> +
> + ret = cam.stop()
> + self.assertZero(ret)
> +
> +
> +
> +# Recursively expand slist's objects into olist, using seen to track already
> +# processed objects.
> +def _getr(slist, olist, seen):
> + for e in slist:
> + if id(e) in seen:
> + continue
> + seen.add(id(e))
> + olist.append(e)
> + tl = gc.get_referents(e)
> + if tl:
> + _getr(tl, olist, seen)
> +
> +def get_all_objects(ignored = []):
> + gcl = gc.get_objects()
> + olist = []
> + seen = set()
> +
> + seen.add(id(gcl))
> + seen.add(id(olist))
> + seen.add(id(seen))
> + seen.update(set([id(o) for o in ignored]))
> +
> + _getr(gcl, olist, seen)
> +
> + return olist
> +
> +def create_type_count_map(olist):
> + map = defaultdict(int)
> + for o in olist:
> + map[type(o)] += 1
> + return map
> +
> +def diff_type_count_maps(before, after):
> + return [(k, after[k] - before[k]) for k in after if after[k] != before[k]]
> +
> +if __name__ == '__main__':
> + # doesn't work very well, as things always leak a bit
> + test_leaks = False
> +
> + if test_leaks:
> + gc.unfreeze()
> + gc.collect()
> +
> + obs_before = get_all_objects()
> +
> + unittest.main(exit=False)
> +
> + if test_leaks:
> + gc.unfreeze()
> + gc.collect()
> +
> + obs_after = get_all_objects([obs_before])
> +
> + before = create_type_count_map(obs_before)
> + after = create_type_count_map(obs_after)
> +
> + leaks = diff_type_count_maps(before, after)
> + if len(leaks) > 0:
> + print(leaks)
> --
> 2.25.1
>
More information about the libcamera-devel
mailing list