| 1433 | } |
| 1434 | |
| 1435 | CapabilityResult invoke_capability(const std::string& name, const std::string& args_json) { |
| 1436 | ensure_initialized(); |
| 1437 | const GilAcquire gil; |
| 1438 | CapabilityResult result; |
| 1439 | |
| 1440 | PyObject* lichtfeld = import_lichtfeld_module("Failed to import lichtfeld while invoking capability"); |
| 1441 | if (!lichtfeld) { |
| 1442 | return {false, "", "Failed to import lichtfeld"}; |
| 1443 | } |
| 1444 | Py_DECREF(lichtfeld); |
| 1445 | |
| 1446 | ensure_plugins_loaded(); |
| 1447 | |
| 1448 | PyObject* lfs_plugins = PyImport_ImportModule("lfs_plugins"); |
| 1449 | if (!lfs_plugins) { |
| 1450 | PyErr_Print(); |
| 1451 | return {false, "", "Failed to import lfs_plugins"}; |
| 1452 | } |
| 1453 | |
| 1454 | PyObject* registry_class = PyObject_GetAttrString(lfs_plugins, "CapabilityRegistry"); |
| 1455 | if (!registry_class) { |
| 1456 | Py_DECREF(lfs_plugins); |
| 1457 | return {false, "", "CapabilityRegistry not found"}; |
| 1458 | } |
| 1459 | |
| 1460 | PyObject* instance_method = PyObject_GetAttrString(registry_class, "instance"); |
| 1461 | PyObject* registry = PyObject_CallNoArgs(instance_method); |
| 1462 | Py_DECREF(instance_method); |
| 1463 | Py_DECREF(registry_class); |
| 1464 | |
| 1465 | if (!registry) { |
| 1466 | Py_DECREF(lfs_plugins); |
| 1467 | return {false, "", "Failed to get capability registry instance"}; |
| 1468 | } |
| 1469 | |
| 1470 | PyObject* json_module = PyImport_ImportModule("json"); |
| 1471 | PyObject* loads = PyObject_GetAttrString(json_module, "loads"); |
| 1472 | PyObject* dumps = PyObject_GetAttrString(json_module, "dumps"); |
| 1473 | PyObject* py_args_str = PyUnicode_FromString(args_json.c_str()); |
| 1474 | PyObject* args_dict = PyObject_CallOneArg(loads, py_args_str); |
| 1475 | Py_DECREF(py_args_str); |
| 1476 | |
| 1477 | if (!args_dict) { |
| 1478 | PyErr_Clear(); |
| 1479 | args_dict = PyDict_New(); |
| 1480 | } |
| 1481 | |
| 1482 | PyObject* invoke_method = PyObject_GetAttrString(registry, "invoke"); |
| 1483 | PyObject* py_name = PyUnicode_FromString(name.c_str()); |
| 1484 | PyObject* py_result = PyObject_CallFunctionObjArgs(invoke_method, py_name, args_dict, nullptr); |
| 1485 | Py_DECREF(py_name); |
| 1486 | Py_DECREF(args_dict); |
| 1487 | Py_DECREF(invoke_method); |
| 1488 | |
| 1489 | if (py_result && PyDict_Check(py_result)) { |
| 1490 | PyObject* success = PyDict_GetItemString(py_result, "success"); |
| 1491 | result.success = success && PyObject_IsTrue(success); |
| 1492 |
no test coverage detected