Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _test_overwrite_file(task):
filename = "{}/{}-{}".format(BASEPATH, task.host, str(uuid.uuid4()))
r = task.run(files.write_file, dry_run=False, filename=filename, content=content_a)
assert r.diff.splitlines()[1:] == diff_new.splitlines()[1:]
assert r.changed
r = task.run(files.write_file, dry_run=False, filename=filename, content=content_b)
assert r.diff.splitlines()[1:] == diff_overwrite_file.splitlines()[1:]
assert r.changed
r = task.run(files.write_file, dry_run=False, filename=filename, content=content_b)
assert not r.diff
assert not r.changed
def create_yaml_file(task, python_object_input, filename="hosts_new.yaml"):
print(python_object_input)
yaml_file = yaml.dump(python_object_input, default_flow_style=False)
print("yaml file\n\n")
print(yaml_file)
task.run(
task=write_file,
filename=filename,
content=yaml_file,
)
def write_configurations(task):
task.run(
task=files.write_file,
filename=f"rendered_configs/{task.host}_bgp",
content=task.host["bgp_config"],
)
task.run(
task=files.write_file,
filename=f"rendered_configs/{task.host}_intf",
content=task.host["intf_config"],
)
def write_configurations(task):
task.run(
task=files.write_file,
filename=f"rendered_configs/{task.host}_bgp",
content=task.host["bgp_config"],
)
task.run(
task=files.write_file,
filename=f"rendered_configs/{task.host}_intf",
content=task.host["intf_config"],
)
rosetta_device_driver = rosetta_get_driver(task.host.platform)
new_vlans = {'openconfig-network-instance:network-instances':
{'network-instance': [{'name': 'default', 'config': {'name': 'default'},
'vlans': {'vlan': [{'vlan-id': 10, 'config': {'vlan-id': 10, 'name': 'prod', 'status': 'ACTIVE'}},
{'vlan-id': 20, 'config': {'vlan-id': 20, 'name': 'dev', 'status': 'SUSPENDED'}}]}}]}}
native = rosetta_device_driver.translate(candidate=new_vlans)
try:
running_config = json.load(task.host["rosetta_parsed_config"])
except AttributeError: # fixing for AttributeError: 'str' object has no attribute 'read'
running_config = json.loads(task.host["rosetta_parsed_config"])
merged_config = rosetta_device_driver.merge(candidate=new_vlans, running=running_config)
print("Proposed New Native config from Rosetta is ")
print(merged_config)
# syntax errors for merge config need to be fixed
task.run(
task=write_file,
filename=f"{path}/{task.host}_rosetta_proposed_merge_config.conf",
content=merged_config,
)
host_dir = task.host.name
# Assign the destination directory to a variable. i.e facts/hostname/
entry_dir = fact_dir + "/" + host_dir
# If Else to handle creating directories. Default is to create them
if dir is False:
# Create facts directory
pathlib.Path(fact_dir).mkdir(exist_ok=True)
# Create entry directory
pathlib.Path(entry_dir).mkdir(exist_ok=True)
# Try/except block to catch exceptions, such as NotImplementedError
try:
# Gather facts using napalm_get and assign to a variable
facts_result = task.run(task=napalm_get, getters=[getter])
# Write the results to a JSON, using the convention .json
task.run(
task=write_file,
content=json.dumps(facts_result[0].result[getter], indent=2),
filename=f"" + str(entry_dir) + "/" + str(getter) + ".json",
)
# Handle NAPALM Not Implemented Error exceptions
except NotImplementedError:
return "Getter Not Implemented"
except AttributeError:
return "AttributeError: Driver has no attribute"
host_dir = task.host.name
# Assign the destination directory to a variable. i.e configs/hostname/
entry_dir = config_dir + "/" + host_dir
# If Else to handle creating directories. Default is to create them
if dir is False:
# Create facts directory
pathlib.Path(config_dir).mkdir(exist_ok=True)
# Create entry directory
pathlib.Path(entry_dir).mkdir(exist_ok=True)
# Try/except block to catch exceptions, such as NotImplementedError
try:
# Gather config using napalm_get and assign to a variable
config_result = task.run(task=napalm_get, getters=["config"])
# Write the results to a JSON, using the convention .txt
task.run(
task=write_file,
content=config_result.result["config"][getter],
filename=f"" + str(entry_dir) + "/" + str(getter) + ".txt",
)
# Handle NAPALM Not Implemented Error exceptions
except NotImplementedError:
print("NAPALM get filter not implemented " + str(getter))