test_transformation.py

#!/usr/bin/env python3
#
# Create expanded topology file, Ansible inventory, host vars, or Vagrantfile from
# topology file
#

importsys
importos
importglob
importpathlib
importpytest
importdifflib
importpathlib
fromboximportBox

importutils

fromnetsim.utilsimportlog,strings,readas_read
fromnetsimimportaugment
fromnetsim.outputsimport_TopologyOutput,ansible
fromnetsim.dataimporttypesas_types

defrun_test(fname):
log.init_log_system(header=False)
topology=_read.load(fname,relative_topo_name=True,user_defaults=[])
log.exit_on_error()
augment.main.transform(topology)
log.exit_on_error()
returntopology

@pytest.mark.filterwarnings("ignore::PendingDeprecationWarning")
deftest_transformation_cases(tmpdir):
print("Starting transformation test cases")
fortest_caseinlist(glob.glob('topology/input/*yml')):
print("Test case: %s"%test_case)
topology=run_test(test_case)

iftopology.defaults.get("inventory"):
print("Writing inventory... %s"%topology.defaults.inventory)
ansible.ansible_inventory(topology,tmpdir+"/extra/hosts.yml",topology.defaults.get("inventory").replace("dump",""))
ansible.ansible_config(tmpdir+"/ansible.cfg",tmpdir+"/hosts.yml")
iftopology.defaults.inventory=="dump":
ansible.dump(topology)

iftopology.defaults.get("Output"):
foroutput_formatintopology.defaults.get("Output"):
output_module=_TopologyOutput.load(output_format,topology.defaults.outputs[output_format])
ifoutput_module:
output_module.write(Box(topology))
else:
log.error('Unknown output format %s'%output_format,log.IncorrectValue,'create')

result=utils.transformation_results_yaml(topology)
exp_test_case="topology/expected/"+os.path.basename(test_case)
expected=pathlib.Path(exp_test_case).read_text()
ifresult!=expected:
print("Test case: %s FAILED"%test_case)
sys.stdout.writelines(
difflib.unified_diff(
expected.splitlines(keepends=True),
result.splitlines(keepends=True),
fromfile='expected',tofile='result'))

assertresult==expected
print("... succeeded, string length = %d"%len(result))

# Verbose test cases are executed only when we're doing a coverage report
#
deftest_verbose_cases(tmpdir):
ifnotsys.gettrace():
return
log.set_verbose()
test_transformation_cases(tmpdir)

@pytest.mark.filterwarnings("ignore::PendingDeprecationWarning")
deftest_error_cases():
print("Starting error test cases")
log.set_flag(raise_error=True)
fortest_caseinlist(glob.glob('errors/*yml')):
print("Test case: %s"%test_case)
log.err_count=0
withpytest.raises(log.ErrorAbort):
topo=run_test(test_case)

error_log=log.get_error_log()
log_file=pathlib.Path(test_case.replace('.yml','.log'))
iflog_file.exists():
withlog_file.open() asf:
log_lines= [line.rstrip('\n') forlineinf]

iferror_log!=log_lines:
error_log_text="\n".join(error_log)
expected_text="\n".join(log_lines)
print(f'Accumulated error log\n{"="*70}\n{error_log_text}\n\nExpected log\n{"="*70}\n{expected_text}')
asserterror_log==log_lines

if__name__=="__main__":
test_transformation_cases("/tmp")
# test_error_cases()
# test_minimal_cases()