From patchwork Fri Feb 4 13:38:13 2022 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit X-Patchwork-Submitter: Tomi Valkeinen X-Patchwork-Id: 15331 Return-Path: X-Original-To: parsemail@patchwork.libcamera.org Delivered-To: parsemail@patchwork.libcamera.org Received: from lancelot.ideasonboard.com (lancelot.ideasonboard.com [92.243.16.209]) by patchwork.libcamera.org (Postfix) with ESMTPS id A7287BDCBF for ; Fri, 4 Feb 2022 13:38:48 +0000 (UTC) Received: from lancelot.ideasonboard.com (localhost [IPv6:::1]) by lancelot.ideasonboard.com (Postfix) with ESMTP id 55E5A609FF; Fri, 4 Feb 2022 14:38:48 +0100 (CET) Authentication-Results: lancelot.ideasonboard.com; dkim=fail reason="signature verification failed" (1024-bit key; unprotected) header.d=ideasonboard.com header.i=@ideasonboard.com header.b="jtOMakOb"; dkim-atps=neutral Received: from perceval.ideasonboard.com (perceval.ideasonboard.com [IPv6:2001:4b98:dc2:55:216:3eff:fef7:d647]) by lancelot.ideasonboard.com (Postfix) with ESMTPS id 67E56609F7 for ; Fri, 4 Feb 2022 14:38:44 +0100 (CET) Received: from deskari.lan (91-156-85-209.elisa-laajakaista.fi [91.156.85.209]) by perceval.ideasonboard.com (Postfix) with ESMTPSA id CE96D475; Fri, 4 Feb 2022 14:38:43 +0100 (CET) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/simple; d=ideasonboard.com; s=mail; t=1643981924; bh=MBBTki4/Xe1DZDjwSd1vXjfdBBQka67MCyaPDz0lnVQ=; h=From:To:Cc:Subject:Date:In-Reply-To:References:From; b=jtOMakObsgQFv9rsYTPWaRE1lCA9kNaLIrYh9/15ecacXmI0SgoatFTwmDpihmaF1 qv4M8i+BPBubBHQ0Oky28tCccNicFd1x42m3Hj5Z7HR7psFcvIp2qjkKLEG2+GWvec zVMW4DGunJSbFLQO8XH/UI6NjKRP+M4v/CQXBhIs= From: Tomi Valkeinen To: libcamera-devel@lists.libcamera.org, David Plowman , Kieran Bingham , Laurent Pinchart Date: Fri, 4 Feb 2022 15:38:13 +0200 Message-Id: <20220204133814.303217-5-tomi.valkeinen@ideasonboard.com> X-Mailer: git-send-email 2.25.1 In-Reply-To: <20220204133814.303217-1-tomi.valkeinen@ideasonboard.com> References: <20220204133814.303217-1-tomi.valkeinen@ideasonboard.com> MIME-Version: 1.0 Subject: [libcamera-devel] [PATCH v4 4/5] py: add unittests.py X-BeenThere: libcamera-devel@lists.libcamera.org X-Mailman-Version: 2.1.29 Precedence: list List-Id: List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , Errors-To: libcamera-devel-bounces@lists.libcamera.org Sender: "libcamera-devel" Add a simple unittests.py as a base for python unittests. Signed-off-by: Tomi Valkeinen --- src/py/test/unittests.py | 366 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 366 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..5fe29fd9 --- /dev/null +++ b/src/py/test/unittests.py @@ -0,0 +1,366 @@ +#!/usr/bin/env python3 + +# SPDX-License-Identifier: GPL-2.0-or-later +# Copyright (C) 2021, Tomi Valkeinen + +from collections import defaultdict +import errno +import gc +import libcamera as libcam +import os +import os +import selectors +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)