1import os
2import sys
3import time
4from typing import Optional
5
6import pexpect
7import pytest
8import requests
9
10DOCKER_IMAGE = os.environ.get("AMBASSADOR_DOCKER_IMAGE", "")
11
12child: Optional[pexpect.spawnbase.SpawnBase] = None # see docker_start()
13ambassador_host: Optional[str] = None # see docker_start()
14
15
16def docker_start(logfile) -> bool:
17 # Use a global here so that the child process doesn't get killed
18 global child
19
20 global ambassador_host
21
22 cmd = (
23 f"docker run --rm --name test_docker_ambassador -p 9987:8080 -u8888:0 {DOCKER_IMAGE} --demo"
24 )
25 ambassador_host = "localhost:9987"
26
27 child = pexpect.spawn(cmd, encoding="utf-8")
28 child.logfile = logfile
29
30 i = child.expect([pexpect.EOF, pexpect.TIMEOUT, "AMBASSADOR DEMO RUNNING"])
31
32 if i == 0:
33 logfile.write("ambassador died?\n")
34 return False
35 elif i == 1:
36 logfile.write("ambassador timed out?\n")
37 return False
38 else:
39 logfile.write("ambassador running\n")
40 return True
41
42
43def docker_kill(logfile):
44 cmd = f"docker kill test_docker_ambassador"
45
46 child = pexpect.spawn(cmd, encoding="utf-8")
47 child.logfile = logfile
48
49 child.expect([pexpect.EOF, pexpect.TIMEOUT])
50
51
52def check_http(logfile) -> bool:
53 try:
54 logfile.write("QotM: making request\n")
55 response = requests.get(
56 f"http://{ambassador_host}/qotm/?json=true", headers={"Host": "localhost"}
57 )
58 text = response.text
59
60 logfile.write(f"QotM: got status {response.status_code}, text {text}\n")
61
62 if response.status_code != 200:
63 logfile.write(f"QotM: wanted 200 but got {response.status_code} {text}\n")
64 return False
65
66 return True
67 except Exception as e:
68 logfile.write(f"Could not do HTTP: {e}\n")
69
70 return False
71
72
73def check_cli() -> bool:
74 child = pexpect.spawn(f"docker run --rm --entrypoint ambassador {DOCKER_IMAGE} --help")
75
76 # v_encoded = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
77 i = child.expect([pexpect.EOF, pexpect.TIMEOUT, "Usage: ambassador"])
78
79 if i == 0:
80 print("ambassador died without usage statement?")
81 return False
82 elif i == 1:
83 print("ambassador timed out without usage statement?")
84 return False
85
86 i = child.expect([pexpect.EOF, pexpect.TIMEOUT])
87
88 if i == 0:
89 return True
90 else:
91 print("ambassador timed out after usage statement?")
92 return False
93
94
95def check_grab_snapshots() -> bool:
96 child = pexpect.spawn(f"docker run --rm --entrypoint grab-snapshots {DOCKER_IMAGE} --help")
97
98 i = child.expect([pexpect.EOF, pexpect.TIMEOUT, "Usage: grab-snapshots"])
99
100 if i == 0:
101 print("grab-snapshots died without usage statement?")
102 return False
103 elif i == 1:
104 print("grab-snapshots timed out without usage statement?")
105 return False
106
107 i = child.expect([pexpect.EOF, pexpect.TIMEOUT])
108
109 if i == 0:
110 return True
111 else:
112 print("grab-snapshots timed out after usage statement?")
113 return False
114
115
116def test_cli():
117 assert check_cli(), "CLI check failed"
118
119
120def test_grab_snapshots():
121 assert check_grab_snapshots(), "grab-snapshots check failed"
122
123
124def test_demo():
125 test_status = False
126
127 # And this tests that the Ambasasdor can run with the `--demo` argument
128 # and run normally with a sample /qotm/ Mapping.
129 with open("/tmp/test_docker_output", "w") as logfile:
130 if not DOCKER_IMAGE:
131 logfile.write("No $AMBASSADOR_DOCKER_IMAGE??\n")
132 else:
133 if docker_start(logfile):
134 logfile.write("Demo started, first check...")
135
136 if check_http(logfile):
137 logfile.write("Sleeping for second check...")
138 time.sleep(10)
139
140 if check_http(logfile):
141 test_status = True
142
143 docker_kill(logfile)
144
145 if not test_status:
146 with open("/tmp/test_docker_output", "r") as logfile:
147 for line in logfile:
148 print(line.rstrip())
149
150 assert test_status, "test failed"
151
152
153if __name__ == "__main__":
154 pytest.main(sys.argv)
View as plain text