Skip to content

Commit

Permalink
Add SdBus.method_call_timeout_usec property
Browse files Browse the repository at this point in the history
Controls the D-Bus method call timeout. Is in microseconds as
that is what sd-bus uses.
  • Loading branch information
igo95862 committed Feb 10, 2024
1 parent 540696b commit e7f3bb1
Show file tree
Hide file tree
Showing 3 changed files with 43 additions and 2 deletions.
1 change: 1 addition & 0 deletions src/sdbus/sd_bus_internals.py
Original file line number Diff line number Diff line change
Expand Up @@ -222,6 +222,7 @@ def start(self) -> None:
raise NotImplementedError(__STUB_ERROR)

address: Optional[str] = None
method_call_timeout_usec: int = 0


def sd_bus_open() -> SdBus:
Expand Down
23 changes: 23 additions & 0 deletions src/sdbus/sd_bus_internals_bus.c
Original file line number Diff line number Diff line change
Expand Up @@ -669,8 +669,31 @@ static PyObject* SdBus_address_getter(SdBusObject* self, void* Py_UNUSED(closure
return PyUnicode_FromString(bus_address);
}

static PyObject* SdBus_method_call_timeout_usec_getter(SdBusObject* self, void* Py_UNUSED(closure)) {
uint64_t timeout_usec = 0;
CALL_SD_BUS_AND_CHECK(sd_bus_get_method_call_timeout(self->sd_bus_ref, &timeout_usec));

return PyLong_FromUnsignedLongLong((unsigned long long)timeout_usec);
}

static int SdBus_method_call_timeout_usec_setter(SdBusObject* self, PyObject* new_value, void* Py_UNUSED(closure)) {
if (NULL == new_value) {
PyErr_SetString(PyExc_ValueError, "Cannot delete method call timeout value");
return -1;
}

unsigned long long new_timeout_usec = PyLong_AsUnsignedLongLong(new_value);
if ((((unsigned long long)-1) == new_timeout_usec) && (PyErr_Occurred() != NULL)) {
return -1;
}
CALL_SD_BUS_CHECK_RETURN_NEG1(sd_bus_set_method_call_timeout(self->sd_bus_ref, (uint64_t)new_timeout_usec));
return 0;
}

static PyGetSetDef SdBus_properies[] = {
{"address", (getter)SdBus_address_getter, NULL, PyDoc_STR("Bus address."), NULL},
{"method_call_timeout_usec", (getter)SdBus_method_call_timeout_usec_getter, (setter)SdBus_method_call_timeout_usec_setter,
PyDoc_STR("D-Bus call timeout in microseconds."), NULL},
{0},
};

Expand Down
21 changes: 19 additions & 2 deletions test/test_low_level_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
from __future__ import annotations

from unittest import SkipTest, main
from unittest import SkipTest, TestCase, main

from sdbus.sd_bus_internals import (
SdBus,
Expand All @@ -31,13 +31,15 @@
from sdbus.unittest import IsolatedDbusTestCase


class TestDbusTypes(IsolatedDbusTestCase):
class TestInitDbus(IsolatedDbusTestCase):
def test_init_bus(self) -> None:
not_connected_bus = SdBus()
self.assertIsNone(not_connected_bus.address)

self.assertIsNotNone(self.bus.address)


class TestLowLeveApi(TestCase):
def test_validation_funcs(self) -> None:
try:
self.assertTrue(
Expand Down Expand Up @@ -79,6 +81,21 @@ def test_validation_funcs(self) -> None:
)
)

def test_bus_method_call_timeout(self) -> None:
bus = SdBus()

self.assertIsNotNone(bus.method_call_timeout_usec)

test_timeout_usec = 10 * 10**6 # 10 seconds
bus.method_call_timeout_usec = test_timeout_usec
self.assertEqual(test_timeout_usec, bus.method_call_timeout_usec)

with self.assertRaises(TypeError):
bus.method_call_timeout_usec = "test" # type: ignore

with self.assertRaises(ValueError):
del bus.method_call_timeout_usec


if __name__ == "__main__":
main()

0 comments on commit e7f3bb1

Please sign in to comment.