diff options
-rw-r--r-- | lib/smbconf/pysmbconf.c | 52 | ||||
-rw-r--r-- | python/samba/tests/smbconf.py | 65 |
2 files changed, 117 insertions, 0 deletions
diff --git a/lib/smbconf/pysmbconf.c b/lib/smbconf/pysmbconf.c index 6edca974ccd..b0aca4508af 100644 --- a/lib/smbconf/pysmbconf.c +++ b/lib/smbconf/pysmbconf.c @@ -518,6 +518,39 @@ static PyObject *obj_delete_global_parameter(py_SMBConf_Object * self, Py_RETURN_NONE; } +static PyObject *obj_transaction_start(py_SMBConf_Object * self, + PyObject * Py_UNUSED(ignored)) +{ + sbcErr err = smbconf_transaction_start(self->conf_ctx); + if (err != SBC_ERR_OK) { + py_raise_SMBConfError(err); + return NULL; + } + Py_RETURN_NONE; +} + +static PyObject *obj_transaction_commit(py_SMBConf_Object * self, + PyObject * Py_UNUSED(ignored)) +{ + sbcErr err = smbconf_transaction_commit(self->conf_ctx); + if (err != SBC_ERR_OK) { + py_raise_SMBConfError(err); + return NULL; + } + Py_RETURN_NONE; +} + +static PyObject *obj_transaction_cancel(py_SMBConf_Object * self, + PyObject * Py_UNUSED(ignored)) +{ + sbcErr err = smbconf_transaction_cancel(self->conf_ctx); + if (err != SBC_ERR_OK) { + py_raise_SMBConfError(err); + return NULL; + } + Py_RETURN_NONE; +} + PyDoc_STRVAR(obj_requires_messaging_doc, "requires_messaging() -> bool\n" "\n" @@ -583,6 +616,19 @@ PyDoc_STRVAR(obj_delete_global_parameter_doc, "delete_parameter(str, str) -> None\n" "Delete a single global configuration parameter.\n"); +PyDoc_STRVAR(obj_transaction_start_doc, +"transaction_start() -> None\n" +"Start a transaction.\n" +"Transactions allow making compound sets of changes atomically.\n"); + +PyDoc_STRVAR(obj_transaction_commit_doc, +"transaction_commit() -> None\n" +"Commit the transaction.\n"); + +PyDoc_STRVAR(obj_transaction_cancel_doc, +"transaction_cancel() -> None\n" +"Cancel the transaction.\n"); + static PyMethodDef py_smbconf_obj_methods[] = { { "requires_messaging", (PyCFunction) obj_requires_messaging, METH_NOARGS, obj_requires_messaging_doc }, @@ -610,6 +656,12 @@ static PyMethodDef py_smbconf_obj_methods[] = { obj_delete_parameter_doc }, { "delete_global_parameter", (PyCFunction) obj_delete_global_parameter, METH_VARARGS, obj_delete_global_parameter_doc }, + { "transaction_start", (PyCFunction) obj_transaction_start, METH_NOARGS, + obj_transaction_start_doc }, + { "transaction_commit", (PyCFunction) obj_transaction_commit, + METH_NOARGS, obj_transaction_commit_doc }, + { "transaction_cancel", (PyCFunction) obj_transaction_cancel, + METH_NOARGS, obj_transaction_cancel_doc }, { 0 }, }; diff --git a/python/samba/tests/smbconf.py b/python/samba/tests/smbconf.py index cebad7fba11..e023e2ad59b 100644 --- a/python/samba/tests/smbconf.py +++ b/python/samba/tests/smbconf.py @@ -259,6 +259,71 @@ class SMBConfTests(samba.tests.TestCase): s1 = sconf.get_share("global") self.assertEqual(s1, ("global", [("workgroup", "EXAMPLE")])) + def test_transaction_direct(self): + sconf = self.s3smbconf.init_reg(None) + sconf.drop() + sconf.set_global_parameter("workgroup", "EXAMPLE") + + sconf.transaction_start() + sconf.set_global_parameter("client min protocol", "NT1") + sconf.set_global_parameter("server min protocol", "SMB2") + sconf.transaction_cancel() + + s1 = sconf.get_share("global") + self.assertEqual(s1, ("global", [("workgroup", "EXAMPLE")])) + + sconf.transaction_start() + sconf.set_global_parameter("client min protocol", "NT1") + sconf.set_global_parameter("server min protocol", "SMB2") + sconf.transaction_commit() + + s1 = sconf.get_share("global") + self.assertEqual( + s1, + ( + "global", + [ + ("workgroup", "EXAMPLE"), + ("client min protocol", "NT1"), + ("server min protocol", "SMB2"), + ], + ), + ) + + def test_transaction_tryexc(self): + sconf = self.s3smbconf.init_reg(None) + sconf.drop() + + def _mkshares(shares): + sconf.transaction_start() + try: + for name, params in shares: + sconf.create_set_share(name, params) + sconf.transaction_commit() + except Exception: + sconf.transaction_cancel() + raise + + _mkshares( + [ + ("hello", [("path", "/srv/world")]), + ("goodnight", [("path", "/srv/moon")]), + ] + ) + # this call to _mkshares will fail the whole transaction because + # share name "goodnight" already exists + self.assertRaises( + self.smbconf.SMBConfError, + _mkshares, + [ + ("mars", [("path", "/srv/mars")]), + ("goodnight", [("path", "/srv/phobos")]), + ], + ) + + names = sconf.share_names() + self.assertEqual(names, ["hello", "goodnight"]) + if __name__ == "__main__": import unittest |