Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Refactor test_response to use Requests instead of curl #96

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Refactor test_response to use Requests instead of curl #96

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
aa64ea3
Refactor test_response to use Requests instead of curl
SlouchyButton Dec 4, 2024
67179f0
Make while clause simpler, use logger instead of print.
SlouchyButton Dec 4, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
96 changes: 55 additions & 41 deletions container_ci_suite/engines/s2i_container.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import time
import subprocess
import shutil
import requests

from typing import List
from os import getenv
Expand Down Expand Up @@ -464,49 +465,62 @@ def doc_content_old(self, strings: List) -> bool:
# }

def test_response(
self, url: str = "",
expected_code: int = 200, port: int = 8080,
expected_output: str = "", max_tests: int = 20
self, url: str = "", port: int = 8080,
expected_code: int = 200, expected_output: str = "",
max_tests: int = 20
) -> bool:
url = f"{url}:{port}"
print(f"URL address to get response from container: {url}")
cmd_to_run = "curl --connect-timeout 10 -k -s -w '%{http_code}' " + f"{url}"
# Check if application returns proper HTTP_CODE
print("Check if HTTP_CODE is valid.")
for count in range(max_tests):
"""
Test HTTP response of specified url

If expected output is empty, check will be skipped
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It does not make sense to call a response in case we do not want to check.
We always want to check an output to know the connection works properly.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Imagine that we use some external library (which we do in case of ruby, for example) as a test sample app. We might want to check that the app works - code 200 is returned, and content might be irrelevant for us as it can change without our knowledge.

I feel like having an option to pass the test only based on correct HTTP response code ignoring content is worth having (and even preferred for test stability).

to know the connection works properly

If it is not, you won't get 200 HTTP code.

This feature of API is optional, so it is up to test developer to correctly assess whether checking response body is needed.


Args:
url (str): URL where to send request
port (int): Port of the web service
expected_code (int): Check that response has this code
expected_output (str): Check that response has this text
max_tests (int): Stop after this many unsuccessful tries
"""

url = f"http://{url}:{port}"
logger.debug("URL address to get response from container: %s", url)

response = None
checks_passed = False
while not checks_passed and max_tests > 0:
checks_passed = True
try:
output_code = run_command(cmd=f"{cmd_to_run}", return_output=True)
return_code = output_code[-3:]
print(f"Output is: {output_code} and Return Code is: {return_code}")
try:
int_ret_code = int(return_code)
if int_ret_code == expected_code:
print(f"HTTP_CODE is VALID {int_ret_code}")
break
except ValueError:
logger.info(return_code)
time.sleep(1)
continue
time.sleep(3)
continue
except subprocess.CalledProcessError as cpe:
print(f"Error from {cmd_to_run} is {cpe.stderr}, {cpe.stdout}")
time.sleep(3)

cmd_to_run = "curl --connect-timeout 10 -k -s " + f"{url}"
# Check if application returns proper output
for count in range(max_tests):
output_code = run_command(cmd=f"{cmd_to_run}", return_output=True)
print(f"Check if expected output {expected_output} is in {cmd_to_run}.")
if expected_output in output_code:
print(f"Expected output '{expected_output}' is present.")
return True
print(
f"check_response_inside_cluster:"
f"expected_output {expected_output} not found in output of {cmd_to_run} command. See {output_code}"
)
time.sleep(5)
return False
response = requests.get(url, timeout=10)
phracek marked this conversation as resolved.
Show resolved Hide resolved
except requests.exceptions.ConnectionError:
response = None

if response is None:
checks_passed = False

# This will very probably timeout first few times and it is
# expected to do so. Prevent spam and inform only when timeout
# is fatal.
if max_tests == 1:
logger.error("Timeout on last retry")
else:
if response.status_code != expected_code:
checks_passed = False
if (response.text != expected_output and
expected_output != ""):
checks_passed = False

# Log both code and text together as text can show error
# message
if not checks_passed:
logger.error("Unexpected code %d or output %s, \
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please use also "f" output like
logger.error(f"Unexpected code {response.status_code} or output {response.text}....)

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

https://pylint.readthedocs.io/en/latest/user_guide/messages/warning/logging-fstring-interpolation.html
https://docs.python.org/3/howto/logging.html#optimization

TL;DR It is preferred to use '%s formatting' in logging functions as it allows logger to parse everything lazily - so it is preferred for optimization purposes.

expecting %d %s",
response.status_code, response.text,
expected_code, expected_output)

max_tests -= 1
time.sleep(3)

return checks_passed

# Replacement for ct_check_exec_env_vars
def test_check_exec_env_vars(self, env_filter: str = "^X_SCLS=|/opt/rh|/opt/app-root"):
Expand Down
Loading