-
-
Notifications
You must be signed in to change notification settings - Fork 153
/
test_example_09.py
59 lines (44 loc) · 2.32 KB
/
test_example_09.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
import os.path
import subprocess
import time
import kopf.testing
import pytest
obj_yaml = os.path.relpath(os.path.join(os.path.dirname(__file__), '..', 'obj.yaml'))
example_py = os.path.relpath(os.path.join(os.path.dirname(__file__), 'example.py'))
@pytest.fixture(scope='session')
def crd_yaml():
crd_api = os.environ.get('CRDAPI') or 'v1'
crd_file = 'crd.yaml' if crd_api == 'v1' else f'crd-{crd_api}.yaml'
return os.path.relpath(os.path.join(os.path.dirname(__file__), '..', crd_file))
@pytest.fixture(autouse=True)
def crd_exists(crd_yaml):
subprocess.run(f"kubectl apply -f {crd_yaml}",
check=True, timeout=10, capture_output=True, shell=True)
@pytest.fixture(autouse=True)
def obj_absent():
# Operator is not running in fixtures, so we need a force-delete (or this patch).
subprocess.run(['kubectl', 'patch', '-f', obj_yaml,
'-p', '{"metadata":{"finalizers":[]}}',
'--type', 'merge'],
check=False, timeout=10, capture_output=True)
subprocess.run(f"kubectl delete -f {obj_yaml}",
check=False, timeout=10, capture_output=True, shell=True)
def test_resource_lifecycle():
# To prevent lengthy threads in the loop executor when the process exits.
settings = kopf.OperatorSettings()
settings.watching.server_timeout = 10
# Run an operator and simulate some activity with the operated resource.
with kopf.testing.KopfRunner(['run', '--verbose', '--standalone', example_py],
timeout=60, settings=settings) as runner:
subprocess.run(f"kubectl create -f {obj_yaml}",
shell=True, check=True, timeout=10, capture_output=True)
time.sleep(5) # give it some time to react
subprocess.run(f"kubectl delete -f {obj_yaml}",
shell=True, check=True, timeout=10, capture_output=True)
time.sleep(1) # give it some time to react
# Ensure that the operator did not die on start, or during the operation.
assert runner.exception is None
assert runner.exit_code == 0
# There are usually more than these messages, but we only check for the certain ones.
assert '[default/kopf-example-1] Creation event:' in runner.stdout
assert '[default/kopf-example-1] Something was logged here.' in runner.stdout