diff --git a/cffi/cdefs.h b/cffi/cdefs.h index 52f00b2..8b6d426 100644 --- a/cffi/cdefs.h +++ b/cffi/cdefs.h @@ -182,6 +182,7 @@ enum ly_stmt { #define LY_LOSTORE ... #define LY_LOSTORE_LAST ... int ly_log_options(int); +uint32_t *ly_temp_log_options(uint32_t *); LY_LOG_LEVEL ly_log_level(LY_LOG_LEVEL); extern "Python" void lypy_log_cb(LY_LOG_LEVEL, const char *, const char *, const char *, uint64_t); diff --git a/libyang/__init__.py b/libyang/__init__.py index ff15755..d8f9555 100644 --- a/libyang/__init__.py +++ b/libyang/__init__.py @@ -65,8 +65,9 @@ ) from .extension import ExtensionPlugin, LibyangExtensionError from .keyed_list import KeyedList -from .log import configure_logging +from .log import configure_logging, temp_log_options from .schema import ( + Enum, Extension, ExtensionCompiled, ExtensionParsed, @@ -144,6 +145,7 @@ "DefaultRemoved", "DescriptionAdded", "DescriptionRemoved", + "Enum", "EnumAdded", "EnumRemoved", "Extension", diff --git a/libyang/context.py b/libyang/context.py index 05ad6ec..4489680 100644 --- a/libyang/context.py +++ b/libyang/context.py @@ -203,6 +203,8 @@ def __init__( leafref_extended: bool = False, leafref_linking: bool = False, builtin_plugins_only: bool = False, + all_implemented: bool = False, + enable_imp_features: bool = False, yanglib_path: Optional[str] = None, yanglib_fmt: str = "json", cdata=None, # C type: "struct ly_ctx *" @@ -225,6 +227,10 @@ def __init__( options |= lib.LY_CTX_LEAFREF_LINKING if builtin_plugins_only: options |= lib.LY_CTX_BUILTIN_PLUGINS_ONLY + if all_implemented: + options |= lib.LY_CTX_ALL_IMPLEMENTED + if enable_imp_features: + options |= lib.LY_CTX_ENABLE_IMP_FEATURES # force priv parsed options |= lib.LY_CTX_SET_PRIV_PARSED @@ -615,24 +621,10 @@ def parse_data( if ret != lib.LY_SUCCESS: raise self.error("failed to read input data") - if parent is not None: - ret = lib.lyd_parse_data( - self.cdata, - parent.cdata, - data[0], - fmt, - parser_flgs, - validation_flgs, - ffi.NULL, - ) - lib.ly_in_free(data[0], 0) - if ret != lib.LY_SUCCESS: - raise self.error("failed to parse data tree") - return None - + parent_cdata = parent.cdata if parent is not None else ffi.NULL dnode = ffi.new("struct lyd_node **") ret = lib.lyd_parse_data( - self.cdata, ffi.NULL, data[0], fmt, parser_flgs, validation_flgs, dnode + self.cdata, parent_cdata, data[0], fmt, parser_flgs, validation_flgs, dnode ) lib.ly_in_free(data[0], 0) if ret != lib.LY_SUCCESS: diff --git a/libyang/log.py b/libyang/log.py index f92c70f..564cf60 100644 --- a/libyang/log.py +++ b/libyang/log.py @@ -2,6 +2,7 @@ # Copyright (c) 2020 6WIND S.A. # SPDX-License-Identifier: MIT +from contextlib import contextmanager import logging from _libyang import ffi, lib @@ -26,6 +27,15 @@ def get_libyang_level(py_level): return None +@contextmanager +def temp_log_options(opt: int = 0): + opts = ffi.new("uint32_t *", opt) + + lib.ly_temp_log_options(opts) + yield + lib.ly_temp_log_options(ffi.NULL) + + @ffi.def_extern(name="lypy_log_cb") def libyang_c_logging_callback(level, msg, data_path, schema_path, line): args = [c2str(msg)] @@ -38,7 +48,7 @@ def libyang_c_logging_callback(level, msg, data_path, schema_path, line): args.append(c2str(schema_path)) if line != 0: fmt += " line %u" - args.append(str(line)) + args.append(line) LOG.log(LOG_LEVELS.get(level, logging.NOTSET), fmt, *args) diff --git a/libyang/schema.py b/libyang/schema.py index c9f2a5e..23c1b1b 100644 --- a/libyang/schema.py +++ b/libyang/schema.py @@ -1440,13 +1440,23 @@ def parent(self) -> Optional["SNode"]: return None def when_conditions(self): - wh = ffi.new("struct lysc_when **") wh = lib.lysc_node_when(self.cdata) if wh == ffi.NULL: return for cond in ly_array_iter(wh): yield c2str(lib.lyxp_get_expr(cond.cond)) + def when_conditions_nodes(self) -> Iterator[Optional["SNode"]]: + wh = lib.lysc_node_when(self.cdata) + if wh == ffi.NULL: + return + for cond in ly_array_iter(wh): + yield ( + None + if cond.context == ffi.NULL + else SNode.new(self.context, cond.context) + ) + def parsed(self) -> Optional["PNode"]: if self.cdata_parsed is None or self.cdata_parsed == ffi.NULL: return None diff --git a/tests/test_log.py b/tests/test_log.py new file mode 100644 index 0000000..2834414 --- /dev/null +++ b/tests/test_log.py @@ -0,0 +1,55 @@ +# Copyright (c) 2025, LabN Consulting, L.L.C. +# SPDX-License-Identifier: MIT + +import logging +import os +import sys +import unittest + +from libyang import Context, LibyangError, configure_logging, temp_log_options + + +YANG_DIR = os.path.join(os.path.dirname(__file__), "yang") + + +class LogTest(unittest.TestCase): + def setUp(self): + self.ctx = Context(YANG_DIR) + configure_logging(False, logging.INFO) + + def tearDown(self): + if self.ctx is not None: + self.ctx.destroy() + self.ctx = None + + def _cause_log(self): + try: + assert self.ctx is not None + _ = self.ctx.parse_data_mem("bad", fmt="xml") + except LibyangError: + pass + + @unittest.skipIf(sys.version_info < (3, 10), "Test requires Python 3.10+") + def test_configure_logging(self): + """Test configure_logging API.""" + with self.assertNoLogs("libyang", level="ERROR"): + self._cause_log() + + configure_logging(True, logging.INFO) + with self.assertLogs("libyang", level="ERROR"): + self._cause_log() + + @unittest.skipIf(sys.version_info < (3, 10), "Test requires Python 3.10+") + def test_with_temp_log(self): + """Test configure_logging API.""" + configure_logging(True, logging.INFO) + + with self.assertLogs("libyang", level="ERROR"): + self._cause_log() + + with self.assertNoLogs("libyang", level="ERROR"): + with temp_log_options(0): + self._cause_log() + + with self.assertLogs("libyang", level="ERROR"): + self._cause_log() diff --git a/tests/test_schema.py b/tests/test_schema.py index e27e001..b1862ab 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -918,6 +918,14 @@ def tearDown(self): self.ctx.destroy() self.ctx = None + def test_anydata(self): + snode = next(self.ctx.find_path("/yolo-nodetypes:any1")) + self.assertIsInstance(snode, SAnydata) + assert next(snode.when_conditions()) is not None + snode2 = next(snode.when_conditions_nodes()) + assert isinstance(snode2, SAnydata) + assert snode2.cdata == snode.cdata + def test_anydata_parsed(self): snode = next(self.ctx.find_path("/yolo-nodetypes:any1")) self.assertIsInstance(snode, SAnydata)