Add e2e test for user bearer token

tests_e2e/conftest.py

@@ -239,6 +239,32 @@ def auth_cookies(earthdata_auth_session, url_earthdata, api_host, urs_username,
     return cookiejar
 
 
+@pytest.fixture(scope="module")
+def user_bearer_token(url_earthdata, urs_username, urs_password):
+    parse_result = urllib.parse.urlparse(url_earthdata)
+    edl_url = f"{parse_result.scheme}://{parse_result.netloc}"
+
+    # Create a new token
+    response = requests.post(
+        f"{edl_url}/api/users/token",
+        auth=requests.auth.HTTPBasicAuth(urs_username, str(urs_password)),
+    )
+    response.raise_for_status()
+
+    token = response.json()["access_token"]
+
+    yield token
+
+    # Revoke the token to clean up after ourselves. EDL only allows 2 active
+    # tokens at a time.
+    response = requests.post(
+        f"{edl_url}/api/users/revoke_token",
+        params={"token": token},
+        auth=requests.auth.HTTPBasicAuth(urs_username, str(urs_password)),
+    )
+    response.raise_for_status()
+
+
 # Functions that generate the JSON report file
 def pytest_sessionstart(session):
     session.results = {}

tests_e2e/test_protected.py

@@ -106,19 +106,33 @@ def find_bearer_token(auth_cookies):
     return None
 
 
-def validate_bearer_token_works(auth_cookies, url):
+def test_validate_app_bearer_token(urls, auth_cookies):
+    url = urls.join(urls.METADATA_FILE)
     token = find_bearer_token(auth_cookies)
     assert token is not None
 
     r = requests.get(url, headers={"Authorization": f"Bearer {token}"})
     assert r.status_code == 200
 
 
-def test_validate_bearer_token_works(urls, auth_cookies):
+def test_validate_app_bearer_token_private_file(urls, auth_cookies):
+    url = urls.join("PRIVATE", "ACCESS", "testfile")
+    token = find_bearer_token(auth_cookies)
+    assert token is not None
+
+    r = requests.get(url, headers={"Authorization": f"Bearer {token}"})
+    assert r.status_code == 200
+
+
+def test_validate_user_bearer_token(urls, user_bearer_token):
     url = urls.join(urls.METADATA_FILE)
-    validate_bearer_token_works(auth_cookies, url)
+
+    r = requests.get(url, headers={"Authorization": f"Bearer {user_bearer_token}"})
+    assert r.status_code == 200
 
 
-def test_validate_private_file_bearer_token_works(urls, auth_cookies):
+def test_validate_user_bearer_token_private_file(urls, user_bearer_token):
     url = urls.join("PRIVATE", "ACCESS", "testfile")
-    validate_bearer_token_works(auth_cookies, url)
+
+    r = requests.get(url, headers={"Authorization": f"Bearer {user_bearer_token}"})
+    assert r.status_code == 200