15157215686

Committed 21 May 2025 08:22AM UTC coverage: 91.329% (+0.09%) from 91.241%

Build # 15157215686

Build Type

Pull #126

github

Committed by

web-flow

Commit Message

Merge 838ffcd0c into b2c6cfc2f

Pull Request Pull Request #126: Okfn updates

Run Details

128 of 141 new or added lines in 11 files covered. (90.78%)

10 existing lines in 4 files now uncovered.

969 of 1061 relevant lines covered (91.33%)

3.65 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.7

/ckanext/saml2auth/tests/test_interface.py

"""
Copyright (c) 2020 Keitaro AB

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as
published by the Free Software Foundation, either version 3 of the
License, or (at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU Affero General Public License for more details.

You should have received a copy of the GNU Affero General Public License
along with this program.  If not, see <https://www.gnu.org/licenses/>.
"""

import os
from collections import defaultdict

import pytest

import ckan.model as model
import ckan.plugins as plugins
from ckan.tests import factories

from ckanext.saml2auth.interfaces import ISaml2Auth
from ckanext.saml2auth.tests.test_blueprint_get_request import (
    _prepare_unsigned_response
)

here = os.path.dirname(os.path.abspath(__file__))
extras_folder = os.path.join(here, 'extras')


class ExampleISaml2AuthPlugin(plugins.SingletonPlugin):

    plugins.implements(ISaml2Auth, inherit=True)

    def __init__(self, *args, **kwargs):

        self.calls = defaultdict(int)
        super().__init__(*args, **kwargs)

    def before_saml2_user_update(self, user_dict, saml_attributes):

        self.calls['before_saml2_user_update'] += 1

        user_dict['fullname'] += ' TEST UPDATE'

        user_dict['plugin_extras']['my_plugin'] = {}
        user_dict['plugin_extras']['my_plugin']['key1'] = 'value1'

    def before_saml2_user_create(self, user_dict, saml_attributes):

        self.calls['before_saml2_user_create'] += 1

        user_dict['fullname'] += ' TEST CREATE'

        user_dict['plugin_extras']['my_plugin'] = {}
        user_dict['plugin_extras']['my_plugin']['key2'] = 'value2'

    def after_saml2_login(self, resp, saml_attributes):

        self.calls['after_saml2_login'] += 1

        resp.headers['X-Custom-header'] = 'test'

        return resp


@pytest.mark.usefixtures(u'clean_db', u'with_plugins')
@pytest.mark.ckan_config(u'ckan.plugins', u'saml2auth')
@pytest.mark.ckan_config(u'ckanext.saml2auth.entity_id', u'urn:gov:gsa:SAML:2.0.profiles:sp:sso:test:entity')
@pytest.mark.ckan_config(u'ckanext.saml2auth.idp_metadata.location', u'local')
@pytest.mark.ckan_config(u'ckanext.saml2auth.idp_metadata.local_path', os.path.join(extras_folder, 'provider0', 'idp.xml'))
@pytest.mark.ckan_config(u'ckanext.saml2auth.want_response_signed', u'False')
@pytest.mark.ckan_config(u'ckanext.saml2auth.want_assertions_signed', u'False')
@pytest.mark.ckan_config(u'ckanext.saml2auth.want_assertions_or_response_signed', u'False')
class TestInterface(object):

    def test_after_login_is_called(self, app):

        encoded_response = _prepare_unsigned_response()
        url = '/acs'

        data = {
            'SAMLResponse': encoded_response
        }

        with plugins.use_plugin("test_saml2auth") as plugin:
            response = app.post(url=url, params=data, follow_redirects=False)
            assert 302 == response.status_code

            assert plugin.calls["after_saml2_login"] == 1, plugin.calls

            assert response.headers['X-Custom-header'] == 'test'

    def test_before_create_is_called(self, app):

        encoded_response = _prepare_unsigned_response()
        url = '/acs'

        data = {
            'SAMLResponse': encoded_response
        }

        with plugins.use_plugin("test_saml2auth") as plugin:
            response = app.post(url=url, params=data, follow_redirects=False)
            assert 302 == response.status_code

            assert plugin.calls["before_saml2_user_create"] == 1, plugin.calls

        user = model.User.by_email('test@example.com')
        if isinstance(user, list):
            user = user[0]

        assert user.fullname.endswith('TEST CREATE')

        assert user.plugin_extras['my_plugin']['key2'] == 'value2'

        assert 'saml_id' in user.plugin_extras['saml2auth']

    def test_before_update_is_called_on_saml_user(self, app):

        # From unsigned0.xml
        saml_id = '_ce3d2948b4cf20146dee0a0b3dd6f69b6cf86f62d7'

        user = factories.User(
            email='test@example.com',
            plugin_extras={
                'saml2auth': {
                    'saml_id': saml_id,
                }
            }
        )

        encoded_response = _prepare_unsigned_response()
        url = '/acs'

        data = {
            'SAMLResponse': encoded_response
        }

        with plugins.use_plugin("test_saml2auth") as plugin:
            response = app.post(url=url, params=data, follow_redirects=False)
            assert 302 == response.status_code

            assert plugin.calls["before_saml2_user_update"] == 1, plugin.calls

        user = model.User.by_email('test@example.com')
        if isinstance(user, list):
            user = user[0]

        assert user.fullname.endswith('TEST UPDATE')

        assert user.plugin_extras['my_plugin']['key1'] == 'value1'

        assert user.plugin_extras['saml2auth']['saml_id'] == saml_id

    def test_before_update_is_called_on_ckan_user(self, app):

        user = factories.User(
            email='test@example.com',
        )

        encoded_response = _prepare_unsigned_response()
        url = '/acs'

        data = {
            'SAMLResponse': encoded_response
        }

        with plugins.use_plugin("test_saml2auth") as plugin:
            response = app.post(url=url, params=data, follow_redirects=False)
            assert 302 == response.status_code

            assert plugin.calls["before_saml2_user_update"] == 1, plugin.calls

        user = model.User.by_email('test@example.com')
        if isinstance(user, list):
            user = user[0]

        assert user.fullname.endswith('TEST UPDATE')

        assert 'saml_id' in user.plugin_extras['saml2auth']

1	"""
2	Copyright (c) 2020 Keitaro AB
3
4	This program is free software: you can redistribute it and/or modify
5	it under the terms of the GNU Affero General Public License as
6	published by the Free Software Foundation, either version 3 of the
7	License, or (at your option) any later version.
8
9	This program is distributed in the hope that it will be useful,
10	but WITHOUT ANY WARRANTY; without even the implied warranty of
11	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12	GNU Affero General Public License for more details.
13
14	You should have received a copy of the GNU Affero General Public License
15	along with this program. If not, see <https://www.gnu.org/licenses/>.
16	"""
17
18	import os	4✔
19	from collections import defaultdict	4✔
20
21	import pytest	4✔
22
23	import ckan.model as model	4✔
24	import ckan.plugins as plugins	4✔
25	from ckan.tests import factories	4✔
26
27	from ckanext.saml2auth.interfaces import ISaml2Auth	4✔
28	from ckanext.saml2auth.tests.test_blueprint_get_request import (	4✔
29	_prepare_unsigned_response
30	)
31
32	here = os.path.dirname(os.path.abspath(__file__))	4✔
33	extras_folder = os.path.join(here, 'extras')	4✔
34
35
36	class ExampleISaml2AuthPlugin(plugins.SingletonPlugin):	4✔
37
38	plugins.implements(ISaml2Auth, inherit=True)	4✔
39
40	def __init__(self, args, *kwargs):	4✔
41
42	self.calls = defaultdict(int)	4✔
43	super().__init__(args, *kwargs)	4✔
44
45	def before_saml2_user_update(self, user_dict, saml_attributes):	4✔
46
47	self.calls['before_saml2_user_update'] += 1	4✔
48
49	user_dict['fullname'] += ' TEST UPDATE'	4✔
50
51	user_dict['plugin_extras']['my_plugin'] = {}	4✔
52	user_dict['plugin_extras']['my_plugin']['key1'] = 'value1'	4✔
53
54	def before_saml2_user_create(self, user_dict, saml_attributes):	4✔
55
56	self.calls['before_saml2_user_create'] += 1	4✔
57
58	user_dict['fullname'] += ' TEST CREATE'	4✔
59
60	user_dict['plugin_extras']['my_plugin'] = {}	4✔
61	user_dict['plugin_extras']['my_plugin']['key2'] = 'value2'	4✔
62
63	def after_saml2_login(self, resp, saml_attributes):	4✔
64
65	self.calls['after_saml2_login'] += 1	4✔
66
67	resp.headers['X-Custom-header'] = 'test'	4✔
68
69	return resp	4✔
70
71
72	@pytest.mark.usefixtures(u'clean_db', u'with_plugins')	4✔
73	@pytest.mark.ckan_config(u'ckan.plugins', u'saml2auth')	4✔
74	@pytest.mark.ckan_config(u'ckanext.saml2auth.entity_id', u'urn:gov:gsa:SAML:2.0.profiles:sp:sso:test:entity')	4✔
75	@pytest.mark.ckan_config(u'ckanext.saml2auth.idp_metadata.location', u'local')	4✔
76	@pytest.mark.ckan_config(u'ckanext.saml2auth.idp_metadata.local_path', os.path.join(extras_folder, 'provider0', 'idp.xml'))	4✔
77	@pytest.mark.ckan_config(u'ckanext.saml2auth.want_response_signed', u'False')	4✔
78	@pytest.mark.ckan_config(u'ckanext.saml2auth.want_assertions_signed', u'False')	4✔
79	@pytest.mark.ckan_config(u'ckanext.saml2auth.want_assertions_or_response_signed', u'False')	4✔
80	class TestInterface(object):	4✔
81
82	def test_after_login_is_called(self, app):	4✔
83
84	encoded_response = _prepare_unsigned_response()	4✔
85	url = '/acs'	4✔
86
87	data = {	4✔
88	'SAMLResponse': encoded_response
89	}
90
91	with plugins.use_plugin("test_saml2auth") as plugin:	4✔
92	response = app.post(url=url, params=data, follow_redirects=False)	4✔
93	assert 302 == response.status_code	4✔
94
95	assert plugin.calls["after_saml2_login"] == 1, plugin.calls	4✔
96
97	assert response.headers['X-Custom-header'] == 'test'	4✔
98
99	def test_before_create_is_called(self, app):	4✔
100
101	encoded_response = _prepare_unsigned_response()	4✔
102	url = '/acs'	4✔
103
104	data = {	4✔
105	'SAMLResponse': encoded_response
106	}
107
108	with plugins.use_plugin("test_saml2auth") as plugin:	4✔
109	response = app.post(url=url, params=data, follow_redirects=False)	4✔
110	assert 302 == response.status_code	4✔
111
112	assert plugin.calls["before_saml2_user_create"] == 1, plugin.calls	4✔
113
114	user = model.User.by_email('test@example.com')	4✔
115	if isinstance(user, list):	4✔
UNCOV 116	user = user[0]	×
117
118	assert user.fullname.endswith('TEST CREATE')	4✔
119
120	assert user.plugin_extras['my_plugin']['key2'] == 'value2'	4✔
121
122	assert 'saml_id' in user.plugin_extras['saml2auth']	4✔
123
124	def test_before_update_is_called_on_saml_user(self, app):	4✔
125
126	# From unsigned0.xml
127	saml_id = '_ce3d2948b4cf20146dee0a0b3dd6f69b6cf86f62d7'	4✔
128
129	user = factories.User(	4✔
130	email='test@example.com',
131	plugin_extras={
132	'saml2auth': {
133	'saml_id': saml_id,
134	}
135	}
136	)
137
138	encoded_response = _prepare_unsigned_response()	4✔
139	url = '/acs'	4✔
140
141	data = {	4✔
142	'SAMLResponse': encoded_response
143	}
144
145	with plugins.use_plugin("test_saml2auth") as plugin:	4✔
146	response = app.post(url=url, params=data, follow_redirects=False)	4✔
147	assert 302 == response.status_code	4✔
148
149	assert plugin.calls["before_saml2_user_update"] == 1, plugin.calls	4✔
150
151	user = model.User.by_email('test@example.com')	4✔
152	if isinstance(user, list):	4✔
UNCOV 153	user = user[0]	×
154
155	assert user.fullname.endswith('TEST UPDATE')	4✔
156
157	assert user.plugin_extras['my_plugin']['key1'] == 'value1'	4✔
158
159	assert user.plugin_extras['saml2auth']['saml_id'] == saml_id	4✔
160
161	def test_before_update_is_called_on_ckan_user(self, app):	4✔
162
163	user = factories.User(	4✔
164	email='test@example.com',
165	)
166
167	encoded_response = _prepare_unsigned_response()	4✔
168	url = '/acs'	4✔
169
170	data = {	4✔
171	'SAMLResponse': encoded_response
172	}
173
174	with plugins.use_plugin("test_saml2auth") as plugin:	4✔
175	response = app.post(url=url, params=data, follow_redirects=False)	4✔
176	assert 302 == response.status_code	4✔
177
178	assert plugin.calls["before_saml2_user_update"] == 1, plugin.calls	4✔
179
180	user = model.User.by_email('test@example.com')	4✔
181	if isinstance(user, list):	4✔
UNCOV 182	user = user[0]	×
183
184	assert user.fullname.endswith('TEST UPDATE')	4✔
185
186	assert 'saml_id' in user.plugin_extras['saml2auth']	4✔

keitaroinc / ckanext-saml2auth / 15157215686

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous