Skip to content
Merged
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
9 changes: 9 additions & 0 deletions docs/utilities.md
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,15 @@ v2_options = {
{"lineage": [{"startswith": "router bgp"}], "exit_text": "exit"}
],
"idempotent_commands": [{"lineage": [{"startswith": "interface"}]}],
"negation_negate_with": [
{
"lineage": [
{"startswith": "interface Ethernet"},
{"startswith": "spanning-tree port type"},
],
"use": "no spanning-tree port type",
}
],
}
platform = Platform.CISCO_IOS
driver = load_hconfig_v2_options(v2_options, platform)
Expand Down
175 changes: 87 additions & 88 deletions hier_config/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
IndentAdjustRule,
MatchRule,
NegationDefaultWhenRule,
NegationDefaultWithRule,
OrderingRule,
ParentAllowsDuplicateChildRule,
PerLineSubRule,
Expand Down Expand Up @@ -125,129 +126,127 @@ def hconfig_v3_platform_v2_os_mapper(platform: Platform) -> str:
return "generic"


def load_hconfig_v2_options(
v2_options: dict[str, Any] | str, platform: Platform
) -> HConfigDriverBase:
"""Load Hier Config v2 options to v3 driver format from either a dictionary or a file.
def _process_simple_rules(
v2_options: dict[str, Any],
key: str,
rule_class: type[Any],
append_to: Callable[[Any], None],
) -> None:
"""Process v2 rules that only need match_rules."""
for rule in v2_options.get(key, ()):
match_rules = _collect_match_rules(rule.get("lineage", []))
append_to(rule_class(match_rules=match_rules))

Args:
v2_options (Union[dict, str]): Either a dictionary containing v2 options or
a file path to a YAML file containing the v2 options.
platform (Platform): The Hier Config v3 Platform enum for the target platform.

Returns:
HConfigDriverBase: A v3 driver instance with the migrated rules.

"""
# Load options from a file if a string is provided
if isinstance(v2_options, str):
v2_options = yaml.safe_load(read_text_from_file(file_path=v2_options))

# Ensure v2_options is a dictionary
if not isinstance(v2_options, dict):
msg = "v2_options must be a dictionary or a valid file path."
raise TypeError(msg)

driver = get_hconfig_driver(platform)

def process_rules(
key: str,
rule_class: type[Any],
append_to: Callable[[Any], None],
lineage_key: str = "lineage",
) -> None:
"""Helper to process rules."""
for rule in v2_options.get(key, ()):
match_rules = _collect_match_rules(rule.get(lineage_key, []))
append_to(rule_class(match_rules=match_rules))

# sectional_overwrite
process_rules(
"sectional_overwrite",
SectionalOverwriteRule,
driver.rules.sectional_overwrite.append,
)

# sectional_overwrite_no_negate
process_rules(
"sectional_overwrite_no_negate",
SectionalOverwriteNoNegateRule,
driver.rules.sectional_overwrite_no_negate.append,
)

def _process_custom_rules(
v2_options: dict[str, Any], driver: HConfigDriverBase
) -> None:
"""Process v2 rules that require custom handling."""
for rule in v2_options.get("ordering", ()):
lineage_rules = rule.get("lineage", [])
match_rules = _collect_match_rules(lineage_rules)
match_rules = _collect_match_rules(rule.get("lineage", []))
weight = rule.get("order", 500) - 500

driver.rules.ordering.append(
OrderingRule(match_rules=match_rules, weight=weight),
)

# indent_adjust
for rule in v2_options.get("indent_adjust", ()):
start_expression = rule.get("start_expression")
end_expression = rule.get("end_expression")

driver.rules.indent_adjust.append(
IndentAdjustRule(
start_expression=start_expression, end_expression=end_expression
start_expression=rule.get("start_expression"),
end_expression=rule.get("end_expression"),
)
)

# parent_allows_duplicate_child
process_rules(
"parent_allows_duplicate_child",
ParentAllowsDuplicateChildRule,
driver.rules.parent_allows_duplicate_child.append,
)

# sectional_exiting
for rule in v2_options.get("sectional_exiting", ()):
lineage_rules = rule.get("lineage", [])
match_rules = _collect_match_rules(lineage_rules)
exit_text = rule.get("exit_text", "")

match_rules = _collect_match_rules(rule.get("lineage", []))
driver.rules.sectional_exiting.append(
SectionalExitingRule(match_rules=match_rules, exit_text=exit_text),
SectionalExitingRule(
match_rules=match_rules, exit_text=rule.get("exit_text", "")
),
)

# full_text_sub
for rule in v2_options.get("full_text_sub", ()):
driver.rules.full_text_sub.append(
FullTextSubRule(
search=rule.get("search", ""), replace=rule.get("replace", "")
)
)

# per_line_sub
for rule in v2_options.get("per_line_sub", ()):
driver.rules.per_line_sub.append(
PerLineSubRule(
search=rule.get("search", ""), replace=rule.get("replace", "")
)
)

# idempotent_commands_blacklist -> idempotent_commands_avoid
process_rules(
"idempotent_commands_blacklist",
IdempotentCommandsAvoidRule,
driver.rules.idempotent_commands_avoid.append,
)
for rule in v2_options.get("negation_negate_with", ()):
match_rules = _collect_match_rules(rule.get("lineage", []))
driver.rules.negate_with.append(
NegationDefaultWithRule(match_rules=match_rules, use=rule.get("use", "")),
)

# idempotent_commands
process_rules(
"idempotent_commands",
IdempotentCommandsRule,
driver.rules.idempotent_commands.append,
)

# negation_default_when
process_rules(
"negation_default_when",
NegationDefaultWhenRule,
driver.rules.negation_default_when.append,
def load_hconfig_v2_options(
v2_options: dict[str, Any] | str, platform: Platform
) -> HConfigDriverBase:
"""Load Hier Config v2 options to v3 driver format from either a dictionary or a file.

Args:
v2_options (Union[dict, str]): Either a dictionary containing v2 options or
a file path to a YAML file containing the v2 options.
platform (Platform): The Hier Config v3 Platform enum for the target platform.

Returns:
HConfigDriverBase: A v3 driver instance with the migrated rules.

"""
if isinstance(v2_options, str):
v2_options = yaml.safe_load(read_text_from_file(file_path=v2_options))

if not isinstance(v2_options, dict):
msg = "v2_options must be a dictionary or a valid file path."
raise TypeError(msg)

driver = get_hconfig_driver(platform)

# Process simple rules that only need match_rules
simple_rules: tuple[tuple[str, type[Any], Callable[[Any], None]], ...] = (
(
"sectional_overwrite",
SectionalOverwriteRule,
driver.rules.sectional_overwrite.append,
),
(
"sectional_overwrite_no_negate",
SectionalOverwriteNoNegateRule,
driver.rules.sectional_overwrite_no_negate.append,
),
(
"parent_allows_duplicate_child",
ParentAllowsDuplicateChildRule,
driver.rules.parent_allows_duplicate_child.append,
),
(
"idempotent_commands_blacklist",
IdempotentCommandsAvoidRule,
driver.rules.idempotent_commands_avoid.append,
),
(
"idempotent_commands",
IdempotentCommandsRule,
driver.rules.idempotent_commands.append,
),
(
"negation_default_when",
NegationDefaultWhenRule,
driver.rules.negation_default_when.append,
),
)
for key, rule_class, append_to in simple_rules:
_process_simple_rules(v2_options, key, rule_class, append_to)

# Process rules that require custom handling
_process_custom_rules(v2_options, driver)

return driver

Expand Down
9 changes: 9 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,15 @@ def v2_options() -> dict[str, Any]:
}
],
"idempotent_commands": [{"lineage": [{"startswith": "interface"}]}],
"negation_negate_with": [
{
"lineage": [
{"startswith": "interface Ethernet"},
{"startswith": "spanning-tree port type"},
],
"use": "no spanning-tree port type",
}
],
}


Expand Down
9 changes: 9 additions & 0 deletions tests/test_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -173,6 +173,15 @@ def test_load_hconfig_v2_options(
assert len(driver.rules.idempotent_commands) == 1
assert driver.rules.idempotent_commands[0].match_rules[0].startswith == "interface"

# Assert negation_negate_with -> negate_with
assert len(driver.rules.negate_with) == 1
assert driver.rules.negate_with[0].match_rules[0].startswith == "interface Ethernet"
assert (
driver.rules.negate_with[0].match_rules[1].startswith
== "spanning-tree port type"
)
assert driver.rules.negate_with[0].use == "no spanning-tree port type"


def test_load_hconfig_v2_tags_valid_input() -> None:
v2_tags = [
Expand Down