tests: use DBusMock template, port to ApkPolkit2 interface

Using the template allows to run more code on method calls and
do more complicated logic.

Helps #32

tests/apkpolkit2.py

@@ -0,0 +1,107 @@
+# apkpolkit1.py -- apkPolkit1 mock template
+#
+# Copyright (C) 2022 Pablo Correa Gomez <[email protected]>
+#
+# SPDX-License-Identifier: GPL-2.0+
+
+import dbus
+import dbusmock
+
+import time
+
+# These should ideally be directly fetched from C
+APK_POLKIT_STATE_AVAILABLE = 0
+APK_POLKIT_STATE_INSTALLED = 1
+APK_POLKIT_STATE_UPGRADABLE = 4
+APK_POLKIT_STATE_DOWNGRADABLE = 5
+
+BUS_NAME   = 'dev.Cogitri.apkPolkit2'
+MAIN_OBJ   = '/dev/Cogitri/apkPolkit2'
+MAIN_IFACE = 'dev.Cogitri.apkPolkit2'
+SYSTEM_BUS = True
+
+def load(mock, parameters):
+    repos = [
+        (True, "a", "https://alpine.org/alpine/edge/main"),
+        (False, "b", "https://pmos.org/pmos/master"),
+        (True, "c", "/home/data/foo/bar/baz"),
+    ]
+    mock.repos = repos
+
+    pkgs = [
+        {"name": "apk-test-app",
+         "description": "desktop app",
+         "installed_size": dbus.UInt64(50),
+         "version": "0.1.0-r0",
+         "license": "GPL",
+         "package_state": dbus.UInt32(APK_POLKIT_STATE_AVAILABLE),
+         "size": dbus.UInt64(40),
+         "url": "url"},
+        {"name": "system-pkg",
+         "description": "system package",
+         "installed_size": dbus.UInt64(50),
+         "version": "2.0-r0",
+         "license": "GPL",
+         "package_state": dbus.UInt32(APK_POLKIT_STATE_AVAILABLE),
+         "size": dbus.UInt64(40),
+         "url": "url"},
+    ]
+    mock.pkgs = pkgs
+
+    mock.AddMethods(MAIN_IFACE, [
+        ('AddRepository', 's', '', ''),
+        ('RemoveRepository', 's', '', ''),
+        ('UpdateRepositories', '', '', ''),
+    ])
+
+
+@dbus.service.method(MAIN_IFACE, in_signature='s', out_signature='')
+def AddPackage(self, pkg_name):
+    if (pkg_name == "slow"):
+        time.sleep(10)
+
+@dbus.service.method(MAIN_IFACE, in_signature='s', out_signature='')
+def DeletePackage(self, pkg_name):
+    if (pkg_name == "slow"):
+        time.sleep(10)
+
+@dbus.service.method(MAIN_IFACE, in_signature='asu', out_signature='aa{sv}')
+def GetPackagesDetails(self, packages, requestedProperties):
+    apps = []
+    # We only expect to refine the desktop app for now.
+    # Ideally, the state should be updated on the different DBus calls
+    for pkg in packages:
+        if pkg == "apk-test-app":
+            apps.append(self.pkgs[0])
+        else:
+            apps.append({"name": pkg, "error": "pkg not found!"})
+
+    return apps
+
+@dbus.service.method(MAIN_IFACE, in_signature='', out_signature='a(bss)')
+def ListRepositories(self):
+    return self.repos
+
+@dbus.service.method(MAIN_IFACE, in_signature='u', out_signature='aa{sv}')
+def ListUpgradablePackages(self, requestedProperties):
+    for pkg in self.pkgs:
+        if pkg["name"] == "apk-test-app":
+            pkg["package_state"] = dbus.UInt32(APK_POLKIT_STATE_UPGRADABLE)
+            pkg["staging_version"] = "0.2.0-r0"
+        if pkg["name"] == "system-pkg":
+            pkg["package_state"] = dbus.UInt32(APK_POLKIT_STATE_DOWNGRADABLE)
+            pkg["staging_version"] = "0.1.0-r0"
+    return self.pkgs
+
+
+@dbus.service.method(MAIN_IFACE, in_signature='as', out_signature='')
+def UpgradePackages(self, packages):
+    for pkg in packages:
+        for p in self.pkgs:
+            if p["name"] == pkg:
+                p["version"] = p["staging_version"]
+                p["package_state"] = dbus.UInt32(APK_POLKIT_STATE_INSTALLED)
+
+# @dbus.service.method(MAIN_IFACE, in_signature='asu', out_signature='aa{sv}')
+# def SearchFilesOwners(self, paths, requestedProperties):
+#     return

tests/test.py

@@ -15,51 +15,18 @@ class GsPluginApkTest (DBusTestCase):
     @classmethod
     def setUpClass(klass):
         klass.start_system_bus()
-        klass.dbus_con = klass.get_dbus(system_bus=True)
 
     def setUp(self):
-        self.log = open(os.getenv('DBUS_TEST_LOG'), "w")
-        self.p_mock = self.spawn_server('dev.Cogitri.apkPolkit1',
-                                        '/dev/Cogitri/apkPolkit1',
-                                        'dev.Cogitri.apkPolkit1',
-                                        system_bus=True,
-                                        stdout=self.log)
-
-        self.apk_polkit_mock = dbus.Interface(self.dbus_con.
-                                              get_object('dev.Cogitri.apkPolkit1',
-                                                         '/dev/Cogitri/apkPolkit1'),
-                                              MOCK_IFACE)
-
-        self.apk_polkit_mock.AddMethods('', [
-            ('AddPackage', 's', '', ''),
-            ('DeletePackage', 's', '', ''),
-            ('ListRepositories', '', 'a(bss)', 'ret = [' +
-             '(True, "a", "https://alpine.org/alpine/edge/main"),' +
-             '(False, "b", "https://pmos.org/pmos/master"),' +
-             '(True, "c", "/home/data/foo/bar/baz"),' + ']'),
-            ('UpdateRepositories', '', '', ''),
-            ('AddRepository', 's', '', ''),
-            ('RemoveRepository', 's', '', ''),
-            ('ListUpgradablePackages', '', 'a(ssssssttu)', 'ret = [' +
-             '("apk-test-app", "0.2.0", "desktop app", "GPL", "0.1.0", "url", 50, 40, 4),' + # 4 = UPGRADABLE
-             '("b", "0.2.0", "system package", "GPL", "0.3.0", "url", 50, 40, 5),' + # 5 = DOWNGRADABLE
-             ']'),
-            ('UpgradePackage', 's', '', ''),
-            # We only expect to refine the desktop app for now.
-            # Ideally, the state should be updated on the different DBus calls
-            ('GetPackageDetails', 's', '(ssssssttu)', 'ret = ' +
-             '("apk-test-app", "0.2.0", "desktop app", "GPL", "0.1.0", "url", 50, 40, 2)' # 2 = AVAILABLE
-            ),
-            ('GetPackagesDetails', 'as', 'a(ssssssttu)', 'ret = ' +
-             '[("apk-test-app", "0.2.0", "desktop app", "GPL", "0.1.0", "url", 50, 40, 2),]' # 2 = AVAILABLE
-            ),
-        ])
-
+        self.log = None
+        if os.getenv('DBUS_TEST_LOG') is not None:
+            self.log = open(os.getenv('DBUS_TEST_LOG'), "w")
+        template = os.path.join(os.getenv('G_TEST_SRCDIR'), 'apkpolkit2.py')
+        (self.p_mock, _) = self.spawn_server_template(template,
+                                                      stdout=self.log)
 
     def tearDown(self):
-        self.log.close()
-        self.p_mock.terminate()
-        self.p_mock.wait()
+        if self.log is not None:
+            self.log.close()
 
     def test_apk(self):
         builddir = os.getenv('G_TEST_BUILDDIR')