Fix direct pytest fixture invocations

Signed-off-by: Ivan Kanakarakis <[email protected]>

Author: c00kiemon5ter

tests/satosa/backends/test_openid_connect.py

@@ -80,9 +80,7 @@ def userinfo(self):
     def signing_key(self):
         return RSAKey(key=RSA.generate(2048), alg="RS256")
 
-    def assert_expected_attributes(self, actual_attributes):
-        user_claims = self.userinfo()
-        attr_map = self.internal_attributes()
+    def assert_expected_attributes(self, attr_map, user_claims, actual_attributes):
         expected_attributes = {}
         for out_attr, in_mapping in attr_map["attributes"].items():
             expected_attributes[out_attr] = [user_claims[in_mapping["openid"][0]]]
@@ -97,10 +95,10 @@ def setup_jwks_uri(self, jwks_uri, key):
             status=200,
             content_type="application/json")
 
-    def setup_token_endpoint(self, token_endpoint_url, signing_key):
+    def setup_token_endpoint(self, token_endpoint_url, userinfo, signing_key):
         id_token_claims = {
             "iss": ISSUER,
-            "sub": self.userinfo()["sub"],
+            "sub": userinfo["sub"],
             "aud": CLIENT_ID,
             "nonce": NONCE,
             "exp": time.time() + 3600,
@@ -120,10 +118,10 @@ def setup_token_endpoint(self, token_endpoint_url, signing_key):
                       status=200,
                       content_type="application/json")
 
-    def setup_userinfo_endpoint(self, userinfo_endpoint_url):
+    def setup_userinfo_endpoint(self, userinfo_endpoint_url, userinfo):
         responses.add(responses.POST,
                       userinfo_endpoint_url,
-                      body=json.dumps(self.userinfo()),
+                      body=json.dumps(userinfo),
                       status=200,
                       content_type="application/json")
 
@@ -153,24 +151,24 @@ def test_register_endpoints(self, backend_config):
         assert re.search(regex, redirect_uri_path)
         assert callback == self.oidc_backend.response_endpoint
 
-    def test_translate_response_to_internal_response(self, userinfo):
+    def test_translate_response_to_internal_response(self, internal_attributes, userinfo):
         internal_response = self.oidc_backend._translate_response(userinfo, ISSUER)
         assert internal_response.subject_id == userinfo["sub"]
-        self.assert_expected_attributes(internal_response.attributes)
+        self.assert_expected_attributes(internal_attributes, userinfo, internal_response.attributes)
 
     @responses.activate
-    def test_response_endpoint(self, backend_config, signing_key, incoming_authn_response):
+    def test_response_endpoint(self, backend_config, internal_attributes, userinfo, signing_key, incoming_authn_response):
         self.setup_jwks_uri(backend_config["provider_metadata"]["jwks_uri"], signing_key)
-        self.setup_token_endpoint(backend_config["provider_metadata"]["token_endpoint"], signing_key)
-        self.setup_userinfo_endpoint(backend_config["provider_metadata"]["userinfo_endpoint"])
+        self.setup_token_endpoint(backend_config["provider_metadata"]["token_endpoint"], userinfo, signing_key)
+        self.setup_userinfo_endpoint(backend_config["provider_metadata"]["userinfo_endpoint"], userinfo)
 
         self.oidc_backend.response_endpoint(incoming_authn_response)
         assert self.oidc_backend.name not in incoming_authn_response.state
 
         args = self.oidc_backend.auth_callback_func.call_args[0]
         assert isinstance(args[0], Context)
         assert isinstance(args[1], InternalData)
-        self.assert_expected_attributes(args[1].attributes)
+        self.assert_expected_attributes(internal_attributes, userinfo, args[1].attributes)
 
     def test_start_auth_redirects_to_provider_authorization_endpoint(self, context, backend_config):
         auth_response = self.oidc_backend.start_auth(context, None)
@@ -188,8 +186,8 @@ def test_start_auth_redirects_to_provider_authorization_endpoint(self, context,
         assert "nonce" in auth_params
 
     @responses.activate
-    def test_entire_flow(self, context, backend_config):
-        self.setup_userinfo_endpoint(backend_config["provider_metadata"]["userinfo_endpoint"])
+    def test_entire_flow(self, context, backend_config, internal_attributes, userinfo):
+        self.setup_userinfo_endpoint(backend_config["provider_metadata"]["userinfo_endpoint"], userinfo)
         auth_response = self.oidc_backend.start_auth(context, None)
         auth_params = dict(parse_qsl(urlparse(auth_response.message).query))
 
@@ -198,7 +196,7 @@ def test_entire_flow(self, context, backend_config):
         self.oidc_backend.response_endpoint(context)
         assert self.oidc_backend.name not in context.state
         args = self.oidc_backend.auth_callback_func.call_args[0]
-        self.assert_expected_attributes(args[1].attributes)
+        self.assert_expected_attributes(internal_attributes, userinfo, args[1].attributes)
 
 
 class TestCreateClient(object):

tests/satosa/frontends/test_openid_connect.py

@@ -43,14 +43,17 @@ def frontend_config(self, signing_key_path):
 
         return config
 
-    @pytest.fixture
-    def frontend(self, frontend_config):
+    def create_frontend(self, frontend_config):
         # will use in-memory storage
         instance = OpenIDConnectFrontend(lambda ctx, req: None, INTERNAL_ATTRIBUTES,
                                          frontend_config, BASE_URL, "oidc_frontend")
         instance.register_endpoints(["foo_backend"])
         return instance
 
+    @pytest.fixture
+    def frontend(self, frontend_config):
+        return self.create_frontend(frontend_config)
+
     @pytest.fixture
     def authn_req(self):
         state = "my_state"
@@ -210,7 +213,7 @@ def test_register_endpoints_token_and_userinfo_endpoint_is_published_if_necessar
     def test_register_endpoints_token_and_userinfo_endpoint_is_not_published_if_only_implicit_flow(
             self, frontend_config, context):
         frontend_config["provider"]["response_types_supported"] = ["id_token", "id_token token"]
-        frontend = self.frontend(frontend_config)
+        frontend = self.create_frontend(frontend_config)
 
         urls = frontend.register_endpoints(["test"])
         assert ("^{}/{}".format("test", TokenEndpoint.url), frontend.token_endpoint) not in urls
@@ -227,7 +230,7 @@ def test_register_endpoints_token_and_userinfo_endpoint_is_not_published_if_only
     def test_register_endpoints_dynamic_client_registration_is_configurable(
             self, frontend_config, client_registration_enabled):
         frontend_config["provider"]["client_registration_supported"] = client_registration_enabled
-        frontend = self.frontend(frontend_config)
+        frontend = self.create_frontend(frontend_config)
 
         urls = frontend.register_endpoints(["test"])
         assert (("^{}/{}".format(frontend.name, RegistrationEndpoint.url),
@@ -238,7 +241,7 @@ def test_register_endpoints_dynamic_client_registration_is_configurable(
     def test_token_endpoint(self, context, frontend_config, authn_req):
         token_lifetime = 60 * 60 * 24
         frontend_config["provider"]["access_token_lifetime"] = token_lifetime
-        frontend = self.frontend(frontend_config)
+        frontend = self.create_frontend(frontend_config)
 
         user_id = "test_user"
         self.insert_client_in_client_db(frontend, authn_req["redirect_uri"])