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
4 changes: 0 additions & 4 deletions annet/rulebook/deploying.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,10 +76,6 @@ def match_deploy_rule(rules, cmd_path, context):
if syntax.match_context(ifcontext, context):
if depth == len(cmd_path) - 1:
return rule
else:
rules = rule["children"]
if len(rules) == 0:
break
# default match
return {
"attrs": {
Expand Down
65 changes: 65 additions & 0 deletions tests/annet/test_deploying.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,13 @@
from collections import OrderedDict
from unittest import mock

import annet.vendors
from annet.annlib.command import Command, Question
from annet.annlib.rbparser.deploying import Answer, MakeMessageMatcher
from annet.deploy import apply_deploy_rulebook
from annet.patching import PatchTree
from annet.rulebook.deploying import compile_deploying_text
from tests import make_hw_stub


def test_compile_deploying_text_cisco_2_dialogs(ann_connectors):
Expand Down Expand Up @@ -44,3 +49,63 @@ def test_compile_deploying_text_cisco_2_dialogs(ann_connectors):
)

assert res == expected


def test_deploying_rulebook_ignores_nesting():
"""Test that apply_deploy_rulebook correctly processes block and command structure."""
# Mock HardwareView
text = """
block
dialog: Question? ::: Y
command
dialog: Question? ::: Y
"""
rules = compile_deploying_text(text, "huawei")
hw = make_hw_stub("huawei")

# Create the PatchTree with block and command structure
p = PatchTree()
p.add("block", {})
p.itms[-1].child = PatchTree()
p.itms[-1].child.add("command", {})
p.itms[-1].child.add("quit", {})
p.add("command", {})
p.itms[-1].child = PatchTree()
p.itms[-1].child.add("subcommand", {})
p.itms[-1].child.add("quit", {})

# Expected result from apply_deploy_rulebook
expected = [
Command(cmd="system-view", questions=[], timeout=30, read_timeout=30),
Command(cmd="block", questions=[Question(question="Question?", answer="Y")], timeout=30, read_timeout=None),
Command(cmd="command", questions=[Question(question="Question?", answer="Y")], timeout=30, read_timeout=None),
Command(cmd="quit", questions=[], timeout=30, read_timeout=None),
Command(cmd="command", questions=[Question(question="Question?", answer="Y")], timeout=30, read_timeout=None),
Command(cmd="subcommand", questions=[], timeout=30, read_timeout=None),
Command(cmd="quit", questions=[], timeout=30, read_timeout=None),
Command(cmd="q", questions=[], timeout=30, read_timeout=30),
]

# Mock get_rulebook to return our compiled rules
with mock.patch("annet.deploy.get_rulebook") as mock_get_rulebook:
mock_get_rulebook.return_value = {
"deploying": rules,
"patching": {},
"ordering": None,
"texts": {
"patching": "",
"ordering": "",
"deploying": text,
},
}

# Apply deploy rulebook
formatter = annet.vendors.registry_connector.get().match(hw).make_formatter(indent="")
result = apply_deploy_rulebook(hw, formatter.cmd_paths(p), do_finalize=False, do_commit=False)

assert len(result) == len(expected)
for actual, exp in zip(result, expected):
assert actual.cmd == exp.cmd
assert actual.questions == exp.questions
assert actual.timeout == exp.timeout
assert actual.read_timeout == exp.read_timeout
Loading