test/infamy: add patch_config() for single operations on running

Used by the new NACM basic test to verify ACL rules on explicit XPaths
or modules.  For this we cannot rely on put_config_dict() since it now
uses the candidate datastore for all changes.

Signed-off-by: Joachim Wiberg <troglobit@gmail.com>
This commit is contained in:
Joachim Wiberg
2026-01-23 06:12:25 +01:00
parent a63f2474a2
commit 7ac77b2eef
3 changed files with 79 additions and 0 deletions
+13
View File
@@ -343,6 +343,19 @@ class Device(Transport):
# print(f"Send new XML config: {config}")
return self.put_config(config, retries=retries)
def patch_config(self, modname, edit, retries=3):
"""Merge configuration for a single model to running-config
For NETCONF, this is identical to put_config_dict() since
edit-config already has proper NACM support.
Args:
modname: YANG module name
edit: Configuration dictionary
retries: Number of retry attempts on failure (default 3)
"""
return self.put_config_dict(modname, edit, retries=retries)
def call(self, call):
"""Call RPC, XML version"""
return self.ncc.dispatch(call)
+62
View File
@@ -398,6 +398,68 @@ class Device(Transport):
# Copy candidate to running (acts as "commit", triggers sysrepo callbacks)
self.copy("candidate", "running")
def patch_config(self, modname, edit, retries=3):
"""PATCH configuration directly to running datastore
This bypasses the candidate datastore, avoiding full datastore
copy operations. Useful for NACM-restricted users who only have
access to specific paths. Note: may not trigger sysrepo callbacks
for all configuration types.
Args:
modname: YANG module name
edit: Configuration dictionary
retries: Number of retry attempts on failure (default 3)
"""
try:
mod = self.lyctx.get_module(modname)
except libyang.util.LibyangError:
raise Exception(f"YANG model '{modname}' not found on device. "
f"Model may not be installed or enabled. "
f"Available models can be checked with get_schema_list()") from None
# Parse and convert to get proper structure with module prefix
lyd = mod.parse_data_dict(edit, no_state=True, validate=False)
patch_data = json.loads(lyd.print_mem("json", with_siblings=True, pretty=False))
# PATCH directly to running datastore
url = f"{self.restconf_url}/ds/ietf-datastores:running"
last_error = None
for attempt in range(0, retries):
try:
response = requests_workaround_patch(
url,
json=patch_data,
headers=self.headers,
auth=self.auth,
verify=False
)
response.raise_for_status()
last_error = None
break
except Exception as e:
last_error = e
# Try to extract detailed error message from response
error_detail = str(e)
if hasattr(e, 'response') and e.response is not None:
try:
err_json = e.response.json()
if 'ietf-restconf:errors' in err_json:
errors = err_json['ietf-restconf:errors'].get('error', [])
if errors:
error_detail = errors[0].get('error-message', str(e))
except:
pass
if attempt < retries - 1:
print(f"Failed PATCH: {error_detail} Retrying ...")
time.sleep(1)
else:
print(f"Failed PATCH: {error_detail}")
continue
if last_error is not None:
raise last_error
def call_dict(self, model, call):
pass # Need implementation
+4
View File
@@ -26,6 +26,10 @@ class Transport(ABC):
def put_config_dict(self, modname, edit):
pass
@abstractmethod
def patch_config(self, modname, edit):
pass
@abstractmethod
def get_dict(self, xpath=None):
pass