Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 29 additions & 5 deletions panos/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -556,6 +556,23 @@ def element_str(self, pretty_print=False):
return parsed.toprettyxml(indent="\t", encoding="utf-8")
return ET.tostring(self.element(), encoding="utf-8")

def element_str_inner(self):
"""The XML of this object's children only, without the root entry/member wrapper.

Used by create() to issue a set call against the entry's own xpath rather than
the parent container xpath. This matches how the GUI writes config: target the
entry directly with its inner content so that PAN-OS records a CREATE on the
entry instead of an EDIT on the parent container (which would change admin lock
ownership to the calling admin).

Returns:
str: XML of the child elements as a concatenated byte string, or an empty
byte string if the root has no children.

"""
root = self.element()
return b"".join(ET.tostring(child, encoding="utf-8") for child in root)

def _root_element(self):
if self.SUFFIX == ENTRY:
return ET.Element("entry", {"name": self.uid})
Expand Down Expand Up @@ -669,13 +686,20 @@ def create(self):
device.id + ': create called on %s object "%s"' % (type(self), self.uid)
)
device.set_config_changed()
element = self.element_str()
# For entry/member objects, target the entry's own xpath with inner content only.
# This matches how the GUI writes config: PAN-OS records a CREATE on the entry
# rather than an EDIT on the parent container, which would change admin lock
# ownership to the calling admin.
if self.SUFFIX in (ENTRY, MEMBER):
xpath = self.xpath()
element = self.element_str_inner()
else:
xpath = self.xpath_short()
element = self.element_str()
if self.HA_SYNC:
device.active().xapi.set(
self.xpath_short(), element, retry_on_peer=self.HA_SYNC
)
device.active().xapi.set(xpath, element, retry_on_peer=self.HA_SYNC)
else:
device.xapi.set(self.xpath_short(), element, retry_on_peer=self.HA_SYNC)
device.xapi.set(xpath, element, retry_on_peer=self.HA_SYNC)
for child in self.children:
child._check_child_methods("create")

Expand Down
2 changes: 1 addition & 1 deletion panos/firewall.py
Original file line number Diff line number Diff line change
Expand Up @@ -622,7 +622,7 @@ def element(self):
ET.SubElement(partial, "shared-object").text = "excluded"
if self.exclude_policy_and_objects:
ET.SubElement(partial, "policy-and-objects").text = "excluded"
fe.append(partial)
root.append(partial)

if self.force:
fe = ET.SubElement(root, "force")
Expand Down
58 changes: 58 additions & 0 deletions tests/test_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -583,6 +583,64 @@ def test_create_without_ha_sync(self, m_uid):
for c in self.obj.children:
c._check_child_methods.assert_called_once_with("create")

@mock.patch("panos.base.PanObject.uid", new_callable=mock.PropertyMock)
def test_create_entry_suffix_uses_entry_xpath_and_inner_element(self, m_uid):
# ENTRY-suffix objects must use self.xpath() (not xpath_short()) and
# element_str_inner() (not element_str()) so that PAN-OS records a CREATE
# on the entry itself rather than an EDIT on the parent container.
PanDeviceId = "42"
PanDeviceXpath = "path/to/entry"
PanDeviceInnerElement = b"<from>any</from>"

self.obj.SUFFIX = Base.ENTRY

spec = {"id": PanDeviceId}
m_panos = mock.Mock(**spec)
self.obj.nearest_pandevice = mock.Mock(return_value=m_panos)
self.obj.xpath = mock.Mock(return_value=PanDeviceXpath)
self.obj.element_str_inner = mock.Mock(return_value=PanDeviceInnerElement)
m_uid.return_value = "uid"

ret_val = self.obj.create()

self.assertIsNone(ret_val)
m_panos.set_config_changed.assert_called_once_with()
m_panos.active().xapi.set.assert_called_once_with(
PanDeviceXpath,
PanDeviceInnerElement,
retry_on_peer=self.obj.HA_SYNC,
)
self.obj.xpath.assert_called_once_with()
self.obj.element_str_inner.assert_called_once_with()

@mock.patch("panos.base.PanObject.uid", new_callable=mock.PropertyMock)
def test_create_entry_suffix_without_ha_sync(self, m_uid):
PanDeviceId = "42"
PanDeviceXpath = "path/to/entry"
PanDeviceInnerElement = b"<from>any</from>"

self.obj.SUFFIX = Base.ENTRY
self.obj.HA_SYNC = False

spec = {"id": PanDeviceId}
m_panos = mock.Mock(**spec)
self.obj.nearest_pandevice = mock.Mock(return_value=m_panos)
self.obj.xpath = mock.Mock(return_value=PanDeviceXpath)
self.obj.element_str_inner = mock.Mock(return_value=PanDeviceInnerElement)
m_uid.return_value = "uid"

ret_val = self.obj.create()

self.assertIsNone(ret_val)
m_panos.set_config_changed.assert_called_once_with()
m_panos.xapi.set.assert_called_once_with(
PanDeviceXpath,
PanDeviceInnerElement,
retry_on_peer=self.obj.HA_SYNC,
)
self.obj.xpath.assert_called_once_with()
self.obj.element_str_inner.assert_called_once_with()

@mock.patch("panos.base.PanObject.uid", new_callable=mock.PropertyMock)
def test_delete_with_ha_sync_no_parent(self, m_uid):
PanDeviceId = "42"
Expand Down