From f16246bbd2120892116dd8feebb5b0787c93b52a Mon Sep 17 00:00:00 2001 From: Di Chen Date: Mon, 18 Apr 2022 13:18:11 +0800 Subject: [PATCH] Integration tests predicate on source version check --- tests/conftest.py | 45 +++++++++++++++++++++++++ tests/integration/test_exodus.py | 57 +++++++++++++++++--------------- 2 files changed, 75 insertions(+), 27 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index c9e19dc1..5fd749be 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,6 +4,7 @@ import boto3 import pytest +import requests def get_distribution_url(stack_name): @@ -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 @@ -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_") @@ -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 diff --git a/tests/integration/test_exodus.py b/tests/integration/test_exodus.py index 309699c8..f724064f 100644 --- a/tests/integration/test_exodus.py +++ b/tests/integration/test_exodus.py @@ -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 @@ -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 ( @@ -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 ( @@ -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"]) @@ -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 ( @@ -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 ( @@ -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 ( @@ -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 ( @@ -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" @@ -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