Skip to content

Commit

Permalink
Integration tests predicate on source version check
Browse files Browse the repository at this point in the history
  • Loading branch information
@dichn
dichn committed Apr 18, 2022
1 parent 921713e commit f16246b
Show file tree
Hide file tree
Showing 2 changed files with 75 additions and 27 deletions.
45 changes: 45 additions & 0 deletions tests/conftest.py
View file
Expand Up @@ -4,6 +4,7 @@

import boto3
import pytest
import requests


def get_distribution_url(stack_name):
Expand All @@ -30,6 +31,11 @@ def pytest_addoption(parser):
action="store",
help="enable integration tests against this lambda stack",
)
parser.addoption(
"--src-version",
action="store",
help="enable lambda source version validation",
)


@pytest.fixture
Expand All @@ -47,6 +53,17 @@ def cdn_test_url(request):
return url


@pytest.fixture
def src_version(request):
if request.config.getoption("--src-version"):
version = request.config.getoption("--src-version")
elif os.environ.get("CODEBUILD_RESOLVED_SOURCE_VERSION"):
version = os.environ.get("CODEBUILD_RESOLVED_SOURCE_VERSION")
else:
pytest.skip("Test skipped because failed to get lambda version")
return version


def mock_conf_file():
temp_file = tempfile.NamedTemporaryFile(prefix="lambda_unittests_")

Expand Down Expand Up @@ -105,3 +122,31 @@ def dummy_private_key():
K8PZxUBy9cZ0KOEpAkA1b7cZpW40ZowMvAH6sF+7Ok1NFd+08AMXLiSJ6z7Sk29s
UrfAc2T6ZnfNC4qLIaDyo87CzVG/wk1Upr21z0YD
-----END RSA PRIVATE KEY-----"""


class VersionCheckingAdapter(requests.adapters.HTTPAdapter):
def __init__(self, expected_version):
self.expected_version = expected_version
super().__init__()

def send(self, request, *args, **kwargs):
request.headers["X-Exodus-Query"] = "1"
response = super().send(request, *args, **kwargs)

version = response.headers["X-Exodus-Version"]
if self.expected_version not in version:
raise AssertionError(
"Expected to run against version %s but server sent X-Exodus-Version: %s"
% (self.expected_version, version)
)
return response


@pytest.fixture
def requests_session(src_version):
session = requests.Session()
if src_version:
# we have an expected version, mount an adapter which
# will check for that version
session.mount("https://", VersionCheckingAdapter(src_version))
return session
57 changes: 30 additions & 27 deletions tests/integration/test_exodus.py
View file
Expand Up @@ -2,31 +2,30 @@
import re

import pytest
import requests

from ..test_utils.utils import generate_test_cookies

TEST_COOKIES = generate_test_cookies()


def test_exodus_basic(cdn_test_url):
def test_exodus_basic(cdn_test_url, requests_session):
url = (
cdn_test_url
+ "/content/aus/rhel/server/6/6.5/x86_64/os/Packages/c/cpio-2.10-12.el6_5.x86_64.rpm"
)

r = requests.get(url, cookies=TEST_COOKIES)
r = requests_session.get(url, cookies=TEST_COOKIES)
print(json.dumps(dict(r.headers), indent=2))
assert r.status_code == 200
assert "cache-control" not in r.headers


def test_header_not_exist_file(cdn_test_url):
def test_header_not_exist_file(cdn_test_url, requests_session):
url = (
cdn_test_url
+ "/content/aus/rhel/server/6/6.5/x86_64/os/Packages/c/cpio-2.10-12.el6_5.x86_64.rpm_not_exist" # noqa: E501
)
r = requests.get(url, cookies=TEST_COOKIES)
r = requests_session.get(url, cookies=TEST_COOKIES)
print(json.dumps(dict(r.headers), indent=2))
assert r.status_code == 404
assert "cache-control" not in r.headers
Expand All @@ -42,21 +41,21 @@ def test_header_not_exist_file(cdn_test_url):


@pytest.mark.parametrize("testdata_path", testdata_cache_control_path)
def test_header_cache_control(cdn_test_url, testdata_path):
def test_header_cache_control(cdn_test_url, requests_session, testdata_path):
url = cdn_test_url + testdata_path
r = requests.get(url, cookies=TEST_COOKIES)
r = requests_session.get(url, cookies=TEST_COOKIES)
print(json.dumps(dict(r.headers), indent=2))
assert r.status_code == 200
assert re.match("^max-age=[0-9]+$", r.headers["cache-control"])


def test_header_want_digest_GET(cdn_test_url):
def test_header_want_digest_GET(cdn_test_url, requests_session):
headers = {"want-digest": "id-sha-256"}
url = (
cdn_test_url
+ "/content/dist/rhel/server/7/7.2/x86_64/rhev-mgmt-agent/3/os/repodata/repomd.xml"
)
r = requests.get(url, headers=headers, cookies=TEST_COOKIES)
r = requests_session.get(url, headers=headers, cookies=TEST_COOKIES)
print(json.dumps(dict(r.headers), indent=2))
assert r.status_code == 200
assert (
Expand All @@ -65,13 +64,13 @@ def test_header_want_digest_GET(cdn_test_url):
)


def test_header_want_digest_HEAD(cdn_test_url):
def test_header_want_digest_HEAD(cdn_test_url, requests_session):
headers = {"want-digest": "id-sha-256"}
url = (
cdn_test_url
+ "/content/dist/rhel/server/7/7.2/x86_64/rhev-mgmt-agent/3/os/repodata/repomd.xml"
)
r = requests.head(url, headers=headers, cookies=TEST_COOKIES)
r = requests_session.head(url, headers=headers, cookies=TEST_COOKIES)
print(json.dumps(dict(r.headers), indent=2))
assert r.status_code == 200
assert (
Expand Down Expand Up @@ -101,18 +100,22 @@ def assert_content_type(url, content_type):


@pytest.mark.parametrize("testdata_path", testdata_content_type_path)
def test_content_type_header_GET(cdn_test_url, testdata_path):
def test_content_type_header_GET(
cdn_test_url, requests_session, testdata_path
):
url = cdn_test_url + testdata_path
r = requests.get(url, cookies=TEST_COOKIES)
r = requests_session.get(url, cookies=TEST_COOKIES)
print(json.dumps(dict(r.headers), indent=2))
assert r.status_code == 200
assert_content_type(url, r.headers["Content-Type"])


@pytest.mark.parametrize("testdata_path", testdata_content_type_path)
def test_content_type_header_HEAD(cdn_test_url, testdata_path):
def test_content_type_header_HEAD(
cdn_test_url, requests_session, testdata_path
):
url = cdn_test_url + testdata_path
r = requests.head(url, cookies=TEST_COOKIES)
r = requests_session.head(url, cookies=TEST_COOKIES)
print(json.dumps(dict(r.headers), indent=2))
assert r.status_code == 200
assert_content_type(url, r.headers["Content-Type"])
Expand All @@ -128,10 +131,10 @@ def test_content_type_header_HEAD(cdn_test_url, testdata_path):

# use Want-Digest/Digest to check if alias take effect
@pytest.mark.parametrize("testdata_path", testdata_origin_alias_path)
def test_origin_path_alias(cdn_test_url, testdata_path):
def test_origin_path_alias(cdn_test_url, requests_session, testdata_path):
headers = {"want-digest": "id-sha-256"}
url = cdn_test_url + testdata_path
r = requests.head(url, headers=headers, cookies=TEST_COOKIES)
r = requests_session.head(url, headers=headers, cookies=TEST_COOKIES)
print(json.dumps(dict(r.headers), indent=2))
assert r.status_code == 200
assert (
Expand All @@ -147,10 +150,10 @@ def test_origin_path_alias(cdn_test_url, testdata_path):


@pytest.mark.parametrize("testdata_path", testdata_rhui_alias_path_aus)
def test_rhui_path_alias_aus(cdn_test_url, testdata_path):
def test_rhui_path_alias_aus(cdn_test_url, requests_session, testdata_path):
headers = {"want-digest": "id-sha-256"}
url = cdn_test_url + testdata_path
r = requests.head(url, headers=headers, cookies=TEST_COOKIES)
r = requests_session.head(url, headers=headers, cookies=TEST_COOKIES)
print(json.dumps(dict(r.headers), indent=2))
assert r.status_code == 200
assert (
Expand All @@ -166,10 +169,10 @@ def test_rhui_path_alias_aus(cdn_test_url, testdata_path):


@pytest.mark.parametrize("testdata_path", testdata_rhui_alias_path_rhel8)
def test_rhui_path_alias_rhel8(cdn_test_url, testdata_path):
def test_rhui_path_alias_rhel8(cdn_test_url, requests_session, testdata_path):
headers = {"want-digest": "id-sha-256"}
url = cdn_test_url + testdata_path
r = requests.head(url, headers=headers, cookies=TEST_COOKIES)
r = requests_session.head(url, headers=headers, cookies=TEST_COOKIES)
print(json.dumps(dict(r.headers), indent=2))
assert r.status_code == 200
assert (
Expand All @@ -185,10 +188,10 @@ def test_rhui_path_alias_rhel8(cdn_test_url, testdata_path):


@pytest.mark.parametrize("testdata_path", testdata_releasever_alias_rhel6)
def test_releasever_alias_rhel6(cdn_test_url, testdata_path):
def test_releasever_alias_rhel6(cdn_test_url, requests_session, testdata_path):
headers = {"want-digest": "id-sha-256"}
url = cdn_test_url + testdata_path
r = requests.head(url, headers=headers, cookies=TEST_COOKIES)
r = requests_session.head(url, headers=headers, cookies=TEST_COOKIES)
print(json.dumps(dict(r.headers), indent=2))
assert r.status_code == 200
assert (
Expand All @@ -203,9 +206,9 @@ def test_releasever_alias_rhel6(cdn_test_url, testdata_path):


@pytest.mark.parametrize("testdata_path", testdata_no_content_type)
def test_no_content_type(cdn_test_url, testdata_path):
def test_no_content_type(cdn_test_url, requests_session, testdata_path):
url = cdn_test_url + testdata_path
r = requests.get(url, cookies=TEST_COOKIES)
r = requests_session.get(url, cookies=TEST_COOKIES)
print(json.dumps(dict(r.headers), indent=2))
assert r.status_code == 200
assert r.headers["Content-Type"] == "application/octet-stream"
Expand All @@ -217,8 +220,8 @@ def test_no_content_type(cdn_test_url, testdata_path):


@pytest.mark.parametrize("testdata_path", testdata_absent_item)
def test_absent_item(cdn_test_url, testdata_path):
def test_absent_item(cdn_test_url, requests_session, testdata_path):
url = cdn_test_url + testdata_path
r = requests.get(url, cookies=TEST_COOKIES)
r = requests_session.get(url, cookies=TEST_COOKIES)
print(json.dumps(dict(r.headers), indent=2))
assert r.status_code == 404

0 comments on commit f16246b

Please sign in to comment.