|
2 | 2 | import subprocess
|
3 | 3 | import os
|
4 | 4 | import sys
|
| 5 | +import time |
5 | 6 | import tempfile
|
6 | 7 | from termcolor import colored
|
7 | 8 |
|
|
10 | 11 | TEST_RESOURCES_DIR = os.path.join('..', '..', 'test-resources')
|
11 | 12 | CONFIGURATION_DIR = os.path.join('..', '..', 'configuration')
|
12 | 13 |
|
| 14 | +DONT_CLEANUP = False |
| 15 | +if len(sys.argv) > 1: |
| 16 | + if '--dont-cleanup' in sys.argv or '-d' in sys.argv: |
| 17 | + DONT_CLEANUP = True |
| 18 | + |
13 | 19 | # Get the name of the python executable
|
14 | 20 | python_executable = 'python3'
|
15 | 21 | try:
|
|
32 | 38 | # Create a test namespace
|
33 | 39 | subprocess.check_call(['kubectl', 'create', 'namespace','test-namespace'])
|
34 | 40 |
|
35 |
| -# Apply the configuraton CRD |
36 |
| -subprocess.check_call(['kubectl', 'apply', '-f', os.path.join(CONFIGURATION_DIR, 'policy-configuration-definition.yaml')]) |
37 |
| - |
38 |
| -# Open policy yaml |
39 |
| -with open(os.path.join('policy.yaml'), 'r') as f: |
40 |
| - policy = yaml.load(f, Loader=yaml.FullLoader) |
41 |
| - |
42 |
| -# Test result variable |
43 |
| -all_tests_passed = True |
44 |
| - |
45 |
| -# Loop through the tests |
46 |
| -for test in tests: |
47 |
| - name = test['name'] |
48 |
| - template = test['template'] |
49 |
| - field_change_list = test['field_change_list'] if 'field_change_list' in test else [] |
50 |
| - expected = test['expected'] |
51 |
| - |
52 |
| - print('-'*120) |
53 |
| - print('Running test: ' + name) |
54 |
| - |
55 |
| - # Generate temporary file name for the output file with tempfile |
56 |
| - with tempfile.NamedTemporaryFile() as temp_file: |
57 |
| - test_object_yaml = temp_file.name |
58 |
| - |
59 |
| - # Run the change yaml field script |
60 |
| - if len(field_change_list) > 0: |
61 |
| - print('Changing fields: ' + str(field_change_list)) |
62 |
| - subprocess.check_call([python_executable, os.path.join(SCRIPTS_DIR, 'change-yaml-field.py'), '-i', os.path.join(TEST_RESOURCES_DIR, template), '-o', test_object_yaml] + field_change_list) |
63 |
| - else: |
64 |
| - subprocess.check_call(['cp', os.path.join(TEST_RESOURCES_DIR, template), test_object_yaml]) |
65 |
| - print('Generated test object: ' + test_object_yaml) |
66 |
| - |
67 |
| - # Generate temporary file name for the binding file with tempfile |
68 |
| - with tempfile.NamedTemporaryFile() as temp_file: |
69 |
| - policy_bind_temp_file_name = temp_file.name |
70 |
| - # Create the policy binding file |
71 |
| - policy_name = policy['metadata']['name'] |
72 |
| - policy_bind_change_list = ['spec.policyName=' + policy_name, 'metadata.name=' + policy_name + '-binding', 'spec.paramRef.name=' + policy_name + '-params'] |
73 |
| - subprocess.check_call([python_executable, os.path.join(SCRIPTS_DIR, 'change-yaml-field.py'), '-i', os.path.join(TEST_RESOURCES_DIR, 'policy-binding.yaml'), '-o', policy_bind_temp_file_name] + policy_bind_change_list) |
74 |
| - print('Generated policy binding: ' + policy_bind_temp_file_name) |
75 |
| - |
76 |
| - # Create parameter file |
77 |
| - with tempfile.NamedTemporaryFile() as temp_file: |
78 |
| - param_file_name = temp_file.name |
79 |
| - param_file_change_list = ['metadata.name=' + policy_name + '-params'] |
80 |
| - subprocess.check_call([python_executable, os.path.join(SCRIPTS_DIR, 'change-yaml-field.py'), '-i', os.path.join(TEST_RESOURCES_DIR, 'default-control-configuration.yaml'), '-o', param_file_name] + param_file_change_list) |
81 |
| - print('Generated parameter file: ' + param_file_name) |
82 |
| - |
83 |
| - |
84 |
| - # Run kubectl apply on the policy and policy binding |
85 |
| - subprocess.check_call(['kubectl', 'apply', '-f', param_file_name]) |
86 |
| - subprocess.check_call(['kubectl', 'apply', '-f', policy_bind_temp_file_name]) |
87 |
| - subprocess.check_call(['kubectl', 'apply', '-f', 'policy.yaml']) |
88 |
| - |
89 |
| - # Run kubectl apply on the test object |
90 |
| - result = None |
91 |
| - try: |
92 |
| - subprocess.check_call(['kubectl', '--dry-run=server', 'apply', '-f' ,test_object_yaml]) |
93 |
| - result = 0 |
94 |
| - except subprocess.CalledProcessError as e: |
95 |
| - result = e.returncode |
96 |
| - |
97 |
| - test_passed = False |
98 |
| - # Check if the result is as expected |
99 |
| - if expected == 'pass' and result != 0: |
100 |
| - print(colored('Test failed: expected to pass but failed','red')) |
101 |
| - elif expected == 'fail' and result == 0: |
102 |
| - print(colored('Test failed: expected to fail but passed','red')) |
103 |
| - else: |
104 |
| - test_passed = True |
105 |
| - print(colored('Test passed!','green')) |
106 |
| - |
107 |
| - print(colored('Cleaning up...', 'yellow')) |
108 |
| - # Run kubectl delete on the policy and policy binding. |
109 |
| - try: |
110 |
| - subprocess.check_call(['kubectl', 'delete', '-f', 'policy.yaml'],stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
111 |
| - subprocess.check_call(['kubectl', 'delete', '-f', policy_bind_temp_file_name],stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
112 |
| - subprocess.check_call(['kubectl', 'delete', '-f', test_object_yaml],stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
113 |
| - subprocess.check_call(['kubectl', 'delete', '-f', param_file_name],stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
114 |
| - except: |
115 |
| - pass |
| 41 | +try: |
| 42 | + # Apply the configuraton CRD |
| 43 | + subprocess.check_call(['kubectl', 'apply', '-f', os.path.join(CONFIGURATION_DIR, 'policy-configuration-definition.yaml')]) |
| 44 | + |
| 45 | + # Open policy yaml |
| 46 | + with open(os.path.join('policy.yaml'), 'r') as f: |
| 47 | + policy = yaml.load(f, Loader=yaml.FullLoader) |
| 48 | + |
| 49 | + # Test result variable |
| 50 | + all_tests_passed = True |
| 51 | + |
| 52 | + # Loop through the tests |
| 53 | + for test in tests: |
| 54 | + name = test['name'] |
| 55 | + template = test['template'] |
| 56 | + field_change_list = test['field_change_list'] if 'field_change_list' in test else [] |
| 57 | + expected = test['expected'] |
| 58 | + |
| 59 | + print('-'*120) |
| 60 | + print('Running test: ' + name) |
| 61 | + |
| 62 | + # Generate temporary file name for the output file with tempfile |
| 63 | + with tempfile.NamedTemporaryFile() as temp_file: |
| 64 | + test_object_yaml = temp_file.name |
| 65 | + |
| 66 | + # Run the change yaml field script |
| 67 | + if len(field_change_list) > 0: |
| 68 | + print('Changing fields: ' + str(field_change_list)) |
| 69 | + subprocess.check_call([python_executable, os.path.join(SCRIPTS_DIR, 'change-yaml-field.py'), '-i', os.path.join(TEST_RESOURCES_DIR, template), '-o', test_object_yaml] + field_change_list) |
| 70 | + else: |
| 71 | + subprocess.check_call(['cp', os.path.join(TEST_RESOURCES_DIR, template), test_object_yaml]) |
| 72 | + print('Generated test object: ' + test_object_yaml) |
| 73 | + |
| 74 | + # Generate temporary file name for the binding file with tempfile |
| 75 | + with tempfile.NamedTemporaryFile() as temp_file: |
| 76 | + policy_bind_temp_file_name = temp_file.name |
| 77 | + # Create the policy binding file |
| 78 | + policy_name = policy['metadata']['name'] |
| 79 | + policy_bind_change_list = ['spec.policyName=' + policy_name, 'metadata.name=' + policy_name + '-binding', 'spec.paramRef.name=' + policy_name + '-params'] |
| 80 | + subprocess.check_call([python_executable, os.path.join(SCRIPTS_DIR, 'change-yaml-field.py'), '-i', os.path.join(TEST_RESOURCES_DIR, 'policy-binding.yaml'), '-o', policy_bind_temp_file_name] + policy_bind_change_list) |
| 81 | + print('Generated policy binding: ' + policy_bind_temp_file_name) |
| 82 | + |
| 83 | + # Create parameter file |
| 84 | + with tempfile.NamedTemporaryFile() as temp_file: |
| 85 | + param_file_name = temp_file.name |
| 86 | + param_file_change_list = ['metadata.name=' + policy_name + '-params'] |
| 87 | + subprocess.check_call([python_executable, os.path.join(SCRIPTS_DIR, 'change-yaml-field.py'), '-i', os.path.join(TEST_RESOURCES_DIR, 'default-control-configuration.yaml'), '-o', param_file_name] + param_file_change_list) |
| 88 | + print('Generated parameter file: ' + param_file_name) |
| 89 | + |
| 90 | + |
| 91 | + # Run kubectl apply on the policy and policy binding |
| 92 | + subprocess.check_call(['kubectl', 'apply', '-f', param_file_name]) |
| 93 | + subprocess.check_call(['kubectl', 'apply', '-f', policy_bind_temp_file_name]) |
| 94 | + subprocess.check_call(['kubectl', 'apply', '-f', 'policy.yaml']) |
| 95 | + |
| 96 | + time.sleep(1) |
| 97 | + |
| 98 | + # Run kubectl apply on the test object |
| 99 | + result = None |
| 100 | + try: |
| 101 | + subprocess.check_call(['kubectl', '--dry-run=server', 'apply', '-f' ,test_object_yaml]) |
| 102 | + result = 0 |
| 103 | + except subprocess.CalledProcessError as e: |
| 104 | + result = e.returncode |
| 105 | + |
| 106 | + test_passed = False |
| 107 | + # Check if the result is as expected |
| 108 | + if expected == 'pass' and result != 0: |
| 109 | + print(colored('Test failed: expected to pass but failed','red')) |
| 110 | + elif expected == 'fail' and result == 0: |
| 111 | + print(colored('Test failed: expected to fail but passed','red')) |
| 112 | + else: |
| 113 | + test_passed = True |
| 114 | + print(colored('Test passed!','green')) |
116 | 115 |
|
117 |
| - # Call kubectl wait --for=delete pod -l app=test-pod --timeout=360s |
118 |
| - subprocess.check_call(['kubectl', 'wait', '--for=delete', 'pod', '-l', 'app=test-pod', '--timeout=360s']) |
119 |
| - |
120 |
| - if not test_passed: |
121 |
| - os.remove(policy_bind_temp_file_name) |
122 |
| - os.remove(test_object_yaml) |
123 |
| - os.remove(param_file_name) |
124 |
| - print(colored('Done (left generated object in place)', 'yellow')) |
125 |
| - else: |
126 |
| - print(colored('Done', 'yellow')) |
| 116 | + print(colored('Cleaning up...', 'yellow')) |
| 117 | + # Run kubectl delete on the policy and policy binding. |
| 118 | + try: |
| 119 | + subprocess.check_call(['kubectl', 'delete', '-f', 'policy.yaml'],stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
| 120 | + subprocess.check_call(['kubectl', 'delete', '-f', policy_bind_temp_file_name],stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
| 121 | + subprocess.check_call(['kubectl', 'delete', '-f', test_object_yaml],stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
| 122 | + subprocess.check_call(['kubectl', 'delete', '-f', param_file_name],stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
| 123 | + except: |
| 124 | + pass |
| 125 | + |
| 126 | + # Call kubectl wait --for=delete pod -l app=test-pod --timeout=360s |
| 127 | + subprocess.check_call(['kubectl', 'wait', '--for=delete', 'pod', '-l', 'app=test-pod', '--timeout=360s']) |
| 128 | + |
| 129 | + if not test_passed: |
| 130 | + if not DONT_CLEANUP: |
| 131 | + os.remove(policy_bind_temp_file_name) |
| 132 | + os.remove(test_object_yaml) |
| 133 | + os.remove(param_file_name) |
| 134 | + print(colored('Done (left generated object in place)', 'yellow')) |
| 135 | + else: |
| 136 | + print(colored('Done', 'yellow')) |
| 137 | + else: |
| 138 | + print(colored('Done', 'yellow')) |
| 139 | + |
| 140 | + all_tests_passed = all_tests_passed and test_passed |
| 141 | + |
| 142 | + print('-'*120) |
| 143 | + print('') |
| 144 | +except Exception as e: |
| 145 | + print(colored('Exception: ' + str(e),'red')) |
| 146 | + all_tests_passed = False |
127 | 147 |
|
128 |
| - all_tests_passed = all_tests_passed and test_passed |
129 | 148 |
|
130 |
| - print('-'*120) |
131 |
| - print('') |
132 | 149 |
|
133 | 150 |
|
134 | 151 | # Delete the test namespace
|
|
0 commit comments