Integration tests predicate on source version check

Author: dichn

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,19 @@ def cdn_test_url(request):
     return url
 
 
+@pytest.fixture
+def src_version(request):
+    # fetch exodus-lambda git version for version check
+    # if src_version is unknown, version check will be skipped
+    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:
+        return None
+    return version
+
+
 def mock_conf_file():
     temp_file = tempfile.NamedTemporaryFile(prefix="lambda_unittests_")
 
@@ -105,3 +124,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

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