Tests: Eligibility API

tests/pytest/eligibility/keys/README.md

@@ -0,0 +1,3 @@
+# keys
+
+*These keys are just samples*. They cannot be used for production systems.

tests/pytest/eligibility/keys/client.pub

@@ -0,0 +1,9 @@
+-----BEGIN PUBLIC KEY-----
+MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA1pt0ZoOuPEVPJJS+5r88
+4zcjZLkZZ2GcPwr79XOLDbOi46onCa79kjRnhS0VUK96SwUPS0z9J5mDA5LSNL2R
+oxFb5QGaevnJY828NupzTNdUd0sYJK3kRjKUggHWuB55hwJcH/Dx7I3DNH4NL68U
+AlK+VjwJkfYPrhq/bl5z8ZiurvBa5C1mDxhFpcTZlCfxQoas7D1d+uPACF6mEMbQ
+Nd3RaIaSREO50NvNywXIIt/OmCiRqI7JtOcn4eyh1I4j9WtlbMhRJLfwPMAgY5ep
+TsWcURmhVofF2wVoFbib3JGCfA7tz/gmP5YoEKnf/cumKmF3e9LrZb8zwm7bTHUV
+iwIDAQAB
+-----END PUBLIC KEY-----

tests/pytest/eligibility/keys/server.key

@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEpQIBAAKCAQEAyYo6Pe9OSfPGX0oQXyLAblOwrMgc/j1JlF07b1ahq1lc3FH0
+XEk3Dzqbt9NuQs8hz6493vBNtNWTpVmvbGe4VX3UjhpEARhN3m4jf/Z2OEuDt2A9
+q19NLSjgeyhieLkYLwN1ezYXrkn7cfOngcJMnGDXp45CaA+g3DzasrjETnKUdqec
+CzJ3FJ/RRwfibrju7eS/8s6H03nvydzeAJzTkEv7Fic2JJEUhh2rJhyLxt+qKkIY
+eBG+5fBri4miaS8FPnD/yjZzEAFsQc7n0dGqDAhSJS8tYNmXFmGlaCfRUBNV3mvO
+x0vFPuH21WQ5KKZxZP0e64/uQdotbPIImiyRJwIDAQABAoIBAQCt0ezXe+yOtZQS
+nSMvmh5TSRTogBMZZyxtrFdVeGcpDIKddoWFjpPRK6Af1FeVgWXM459zBthOLaIQ
+iyBUI8SE32iSQq8CLr8CJwWxGJTvipmIb5XglupOF6I8NiFvs1vbOGV7pbSY2i/m
+INoIfNZsTM3SMkytyUTYjhek6txMNtc2yi/3HIVhpEaP8sZrufVGXFbLBOUKgjZC
+h7la/jeSOfb48xoZ8wRq/MHQ1dedH5M19voxEBAcrZlIYiqd4cTr724NQHOJZXNf
+frVq89jKqRvqkPblCPaXqwk8wBfVQyH9LFLbnul2QTxbFRxLXNeoO9qc1ZDwqSXF
+7uRam8y5AoGBAPWs0l2Iilbo+sCSuZK2numdXxnFiJTVuipHpmJXAbKZY/CvayAQ
+pz/mwX34kpTqN9dotnSJYv8y+HQdfdMrPGKQ96RVsl0HJbWHtbiAPtHRXo6gJYho
+th1BhBa1NjJfTXeO7ulPT7OMmKRWC9CEk/OX+rlcHOmpuuebOPKFiSLlAoGBANIC
+kCPL1Ol4sP1RkcDEu06+bqUdi4QvKSgHBmzLb5w+0Ufl8ay3Zp64p4rGMd29L7IV
+wTXPl/B4TdpDKYw84bcsXE2NWfdT6kDaIMWCuiB/iJXTpntHVejRyrd3dz7jwHfy
+PaD5k+KbN2XROIkag0xg7IRmjhJLN5ZxIJIvgScbAoGBAPMmA+J8w+Z2mc7EqRQy
+2J8AmWIpZh9gVOuJlHxZ/p0kQYyyIUVQFighm7mwrmriUThKM+KtIyTO7qYFlkXM
+0ev/7IliI7D85O6AjXM4wnPpUzu39s3GTRAxiqjq2uQJ/OLqvTx+ubRL37suSm0q
++j+qWITiTN9alFisATXOwkadAoGBAL6mEwJcHZohtdMSBNZSApS2ri15B9nlEmDD
+F+MWP+lA4a56og4gpKl8iqShzk01XSI3O6JFJfLo1AxLomEsN+CZBeZlZwHvjR54
+pv2G8r9j57PUYzNRDD2CjpxFeNx/149MOwRy7fzu2bi12bQlfIKPDsgXbexPmlQZ
+uO7c70t3AoGAFrdmmr0Ygt8/b1/j7NwvdaDcQj2uadz0sbqmzBFQSEgbRpD9JRC2
+d21vhv00lZ5VwJ+Bgr35zZ2LeNna1+phlj+rySSHNtz/iDplMMZQvyIHpoUMaccp
+Trt9yCdC1nTavTHbChT4AYkXR87g0EHFhs5w20ILFpPHT1NAARonkJo=
+-----END RSA PRIVATE KEY-----

tests/pytest/eligibility/test_views.py

@@ -1,8 +1,21 @@
 from django.urls import reverse
 import pytest
+import httpretty
+import requests
 
-from benefits.core.models import TransitAgency
-from tests.pytest.conftest import with_agency
+import datetime
+import os
+import uuid
+
+from pathlib import Path
+from jwcrypto import jwk, jwt
+from typing import Tuple
+
+from benefits.core import session
+from benefits.core.models import TransitAgency, EligibilityVerifier
+from benefits.eligibility.api import ApiError, TokenError
+from benefits.eligibility.views import confirm
+from tests.pytest.conftest import with_agency, initialize_request
 
 
 def set_agency(mocker):
@@ -12,13 +25,15 @@ def set_agency(mocker):
     return agency
 
 
-def set_verifier(mocker):
+def set_verifier(mocker) -> Tuple[TransitAgency, EligibilityVerifier]:
     agency = set_agency(mocker)
 
     mock = mocker.patch("benefits.core.session.verifier", autospec=True)
     verifier = agency.eligibility_verifiers.first()
+    mocker.patch.object(verifier, "api_url", "http://localhost/verify")
     assert verifier
     mock.return_value = verifier
+    return (agency, verifier)
 
 
 @pytest.mark.django_db
@@ -52,3 +67,201 @@ def test_start_without_verifier(mocker, client):
     path = reverse("eligibility:start")
     with pytest.raises(AttributeError, match=r"verifier"):
         client.get(path)
+
+
+@httpretty.activate(verbose=True, allow_net_connect=False)
+@pytest.mark.django_db
+def test_confirm_success(mocker, rf):
+    agency, verifier = set_verifier(mocker)
+
+    # Mock the eligibility-server response using HTTPretty
+    # https://stackoverflow.com/questions/21877387/mocking-a-http-server-in-python
+    httpretty.register_uri(
+        httpretty.GET,
+        "http://localhost/verify",
+        status=200,
+        body=_make_token(
+            {
+                "jti": str(uuid.uuid4()),
+                "iss": "test-server",
+                "iat": int(datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc).timestamp()),
+                "eligibility": ["type1"],
+            },
+            verifier.jws_signing_alg,
+            _get_jwk("server.key"),
+            verifier.jwe_encryption_alg,
+            verifier.jwe_cek_enc,
+            _get_jwk("client.pub"),
+        ),
+    )
+
+    path = reverse("eligibility:confirm")
+    body = {"sub": "A0101011", "name": "Lastname"}
+    request = rf.post(path, body)
+
+    initialize_request(request)
+    session.update(request, agency=agency, verifier=verifier, oauth_token="token")
+
+    response = confirm(request)
+
+    assert response.status_code == 302
+    assert response.url == reverse("enrollment:index")
+
+
+def _get_jwk(filename):
+    current_path = Path(os.path.dirname(os.path.realpath(__file__)))
+    file_path = current_path / "keys" / filename
+
+    with file_path.open(mode="rb") as pemfile:
+        key = jwk.JWK.from_pem(pemfile.read())
+
+    return key
+
+
+# copied and pasted from eligibility-server code for now - replace with eligibility-api function when available
+def _make_token(payload, jws_signing_alg, server_private_key, jwe_encryption_alg, jwe_cek_enc, client_public_key):
+    """Wrap payload in a signed and encrypted JWT for response."""
+    # sign the payload with server's private key
+    header = {"typ": "JWS", "alg": jws_signing_alg}
+    signed_token = jwt.JWT(header=header, claims=payload)
+    signed_token.make_signed_token(server_private_key)
+    signed_payload = signed_token.serialize()
+    # encrypt the signed payload with client's public key
+    header = {"typ": "JWE", "alg": jwe_encryption_alg, "enc": jwe_cek_enc}
+    encrypted_token = jwt.JWT(header=header, claims=signed_payload)
+    encrypted_token.make_encrypted_token(client_public_key)
+    return encrypted_token.serialize()
+
+
+@httpretty.activate(verbose=True, allow_net_connect=False)
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    "exception", [requests.ConnectionError, requests.Timeout, requests.TooManyRedirects, requests.HTTPError]
+)
+def test_confirm_failure_error_on_request(mocker, rf, exception):
+    agency, verifier = set_verifier(mocker)
+
+    def raise_exception(*args, **kwargs):
+        raise exception()
+
+    mocker.patch("requests.get", new=raise_exception)
+
+    path = reverse("eligibility:confirm")
+    body = {"sub": "A7654321", "name": "Garcia"}
+    request = rf.post(path, body)
+
+    initialize_request(request)
+    session.update(request, agency=agency, verifier=verifier, oauth_token="token")
+
+    with pytest.raises(ApiError):
+        confirm(request)
+
+
+@httpretty.activate(verbose=True, allow_net_connect=False)
+@pytest.mark.django_db
+def test_confirm_failure_unexpected_status_code(mocker, rf):
+    agency, verifier = set_verifier(mocker)
+
+    httpretty.register_uri(httpretty.GET, "http://localhost/verify", status=404)
+
+    path = reverse("eligibility:confirm")
+    body = {"sub": "A1234567", "name": "Garcia"}
+    request = rf.post(path, body)
+
+    initialize_request(request)
+    session.update(request, agency=agency, verifier=verifier, oauth_token="token")
+
+    with pytest.raises(ApiError, match=r"Unexpected eligibility"):
+        confirm(request)
+
+
+@httpretty.activate(verbose=True, allow_net_connect=False)
+@pytest.mark.django_db
+def test_confirm_failure_error_tokenizing_request(mocker, rf):
+    agency, verifier = set_verifier(mocker)
+    agency.jws_signing_alg = "not real"
+
+    path = reverse("eligibility:confirm")
+    body = {"sub": "A0101011", "name": "Lastname"}
+    request = rf.post(path, body)
+
+    initialize_request(request)
+    session.update(request, agency=agency, verifier=verifier, oauth_token="token")
+
+    with pytest.raises(TokenError):
+        confirm(request)
+
+
+def _tokenize_response_error_scenarios():
+    return [
+        pytest.param(lambda verifier: "", id='TokenError("Invalid response format")'),
+        pytest.param(lambda verifier: "invalid token", id='TokenError("Invalid JWE token")'),
+        # Can't figure out the right way to cause these cases to be thrown.
+        # pytest.param(lambda verifier: _make_token(
+        #     {
+        #         "jti": str(uuid.uuid4()),
+        #         "iss": "test-server",
+        #         "iat": int(datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc).timestamp()),
+        #         "eligibility": ["type1"],
+        #     },
+        #     verifier.jws_signing_alg,
+        #     verifier.jws_signing_alg,
+        #     _get_jwk("server.key"),
+        #     verifier.jwe_encryption_alg,
+        #     verifier.jwe_cek_enc,
+        #     _get_jwk("client.pub"),
+        # ), id='TokenError("JWE token decryption failed")'),
+        # pytest.param(lambda verifier: _make_token(
+        #     {
+        #         "jti": str(uuid.uuid4()),
+        #         "iss": "test-server",
+        #         "iat": int(datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc).timestamp()),
+        #         "eligibility": ["type1"],
+        #     },
+        #     verifier.jws_signing_alg,
+        #     _get_jwk("server.key"),
+        #     verifier.jwe_encryption_alg,
+        #     verifier.jwe_cek_enc,
+        #     _get_jwk("client.pub"),
+        # ), id='TokenError("Invalid JWS token")'),
+        pytest.param(
+            lambda verifier: _make_token(
+                {
+                    "jti": str(uuid.uuid4()),
+                    "iss": "test-server",
+                    "iat": int(datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc).timestamp()),
+                    "eligibility": ["type1"],
+                },
+                "RS512",  # signing algorithm that doesn't match verifier.jws_signing_alg
+                _get_jwk("server.key"),
+                verifier.jwe_encryption_alg,
+                verifier.jwe_cek_enc,
+                _get_jwk("client.pub"),
+            ),
+            id='TokenError("JWS token signature verification failed")',
+        ),
+    ]
+
+
+@httpretty.activate(verbose=True, allow_net_connect=False)
+@pytest.mark.django_db
+@pytest.mark.parametrize("body_lambda", _tokenize_response_error_scenarios())
+def test_confirm_failure_error_tokenizing_response(mocker, rf, body_lambda):
+    agency, verifier = set_verifier(mocker)
+
+    httpretty.register_uri(
+        httpretty.GET,
+        "http://localhost/verify",
+        status=200,
+        body=body_lambda(verifier),
+    )
+
+    path = reverse("eligibility:confirm")
+    body = {"sub": "A1234567", "name": "Garcia"}
+    request = rf.post(path, body)
+
+    initialize_request(request)
+    session.update(request, agency=agency, verifier=None, oauth_token="token")
+
+    with pytest.raises(TokenError):
+        confirm(request)

tests/pytest/requirements.txt

@@ -1,3 +1,4 @@
+httpretty
 pytest
 pytest-cov
 pytest-django