passerelle/tests/api_impot_particulier/test_misc.py

# Passerelle

import datetime
import re
from unittest import mock

import pytest
import requests
import responses

from passerelle.apps.api_impot_particulier.models import Resource
from passerelle.utils.jsonresponse import APIError

from ..utils import make_resource

last_year = str(datetime.date.today().year - 1)


@responses.activate
def test_get_access_token_with_scope():
    responses.add(responses.POST, 'https://dgfip/token', json={'access_token': '1234'})

    assert (
        Resource.get_access_token(requests, 'https://dgfip/impotparticulier/1.0', 'user', 'pass', 'scope')
        == '1234'
    )
    assert len(responses.calls) == 1
    request = responses.calls[0].request
    assert request.body == 'grant_type=client_credentials&scope=scope'
    assert request.headers['authorization'] == 'Basic dXNlcjpwYXNz'


@responses.activate
def test_get_access_token_without_scope():
    responses.add(responses.POST, 'https://dgfip/token', json={'access_token': '1234'})
    assert (
        Resource.get_access_token(requests, 'https://dgfip/impotparticulier/1.0', 'user', 'pass', '')
        == '1234'
    )
    assert len(responses.calls) == 1
    assert responses.calls[0].request.body == 'grant_type=client_credentials'


@responses.activate
def test_get_access_token_connection_timeout():
    responses.add(responses.POST, 'https://dgfip/impotparticulier/1.0', body=requests.ConnectTimeout())
    with pytest.raises(APIError, match=r'.*is unavailable'):
        Resource.get_access_token(requests, 'https://dgfip/', 'user', 'pass', 'scope')


@responses.activate
def test_get_tax_data():
    responses.add(responses.GET, 'https://dgfip/impotparticulier/1.0/123/truc/xyz/', json={'data': 1})

    assert Resource.get_tax_data(
        requests,
        'https://dgfip/impotparticulier/1.0',
        access_token='1234',
        correlation_id='abcd',
        endpoint_template='{var1}/truc/{var2}/',
        var1='123',
        var2='xyz',
        accept='application/prs.dgfip.part.situations.ir.assiettes.v1+json',
    ) == {'data': 1}

    assert len(responses.calls) == 1
    request = responses.calls[0].request
    assert request.headers['authorization'] == 'Bearer 1234'
    assert request.headers['x-correlation-id'] == 'abcd'
    assert request.headers['Accept'] == 'application/prs.dgfip.part.situations.ir.assiettes.v1+json'


@responses.activate
def test_get_tax_data_connection_timeout():
    responses.add(
        responses.GET, 'https://dgfip/impotparticulier/1.0/123/truc/xyz/', body=requests.ConnectTimeout()
    )

    with pytest.raises(APIError, match=r'.*is unavailable'):
        assert Resource.get_tax_data(
            requests,
            'https://dgfip/impotparticulier/1.0',
            access_token='1234',
            correlation_id='abcd',
            endpoint_template='{var1}/truc/{var2}/',
            var1='123',
            var2='xyz',
            accept='application/prs.dgfip.part.situations.ir.assiettes.v1+json',
        ) == {'data': 1}


@responses.activate
@pytest.mark.parametrize(
    'responses_kwargs,response',
    [
        ({'json': {'erreur': {'message': 'foobar'}}, 'status': 403}, {'message': 'foobar'}),
        ({'json': {'erreur': {'message': 'foobar'}}, 'status': 204}, {'message': 'foobar'}),
        ({'headers': {'codeapp': 'foobar'}, 'status': 403}, {'codeapp': 'foobar'}),
        ({'headers': {'codeapp': 'foobar'}, 'status': 204}, {'codeapp': 'foobar'}),
    ],
)
def test_get_tax_data_errors(responses_kwargs, response):
    responses.add(
        responses.GET,
        'https://dgfip/impotparticulier/1.0/123/truc/xyz/',
        **responses_kwargs,
    )

    with pytest.raises(APIError) as ei:
        Resource.get_tax_data(
            requests,
            'https://dgfip/impotparticulier/1.0',
            access_token='1234',
            correlation_id='abcd',
            endpoint_template='{var1}/truc/{var2}/',
            var1='123',
            var2='xyz',
            accept='application/prs.dgfip.part.situations.ir.assiettes.v1+json',
        )
    assert ei.value.data == response


@pytest.fixture
def resource(db):
    return make_resource(
        Resource,
        slug='test',
        title='API Impot Particulier',
        description='API Impot Particulier',
        oauth_username='user',
        oauth_password='pass',
        oauth_scopes='scope',
    )


class TestGetMethods:
    @pytest.fixture(autouse=True)
    def setup(self):
        with responses._default_mock:
            with mock.patch('passerelle.apps.api_impot_particulier.models.uuid') as uuid_mock:
                uuid_mock.uuid4.return_value.hex = 'abcd'
                responses.add(
                    responses.POST, 'https://gw.dgfip.finances.gouv.fr/token', json={'access_token': '1234'}
                )
                yield None

    def test_spi_situations_ir_assiettes_annrev(self, resource):
        responses.add(
            responses.GET,
            'https://gw.dgfip.finances.gouv.fr/impotparticulier/1.0/spi/1234/situations/ir/assiettes/annrev/2023',
            json={'data': 2},
        )
        assert resource.get_spi_situations_ir_assiettes_annrev(
            numero_fiscal='1234', annee_de_revenu='2023'
        ) == {'data': 2}
        assert len(responses.calls) <= 2
        if len(responses.calls) == 2:
            request = responses.calls[0].request
            assert request.body == 'grant_type=client_credentials&scope=scope'
            assert request.headers['authorization'] == 'Basic dXNlcjpwYXNz'
        request = responses.calls[-1].request
        assert request.headers['authorization'] == 'Bearer 1234'
        assert request.headers['x-correlation-id'] == 'abcd'
        assert request.headers['Accept'] == 'application/prs.dgfip.part.situations.ir.assiettes.v1+json'

    def test_spi_situations_th_assiettes_principale_annrev(self, resource):
        responses.add(
            responses.GET,
            'https://gw.dgfip.finances.gouv.fr/impotparticulier/1.0/spi/1234/situations/th/assiettes/principale/annrev/2023',
            json={'data': 2},
        )
        assert resource.get_spi_situations_th_assiettes_principale_annrev(
            numero_fiscal='1234', annee_de_revenu='2023'
        ) == {'data': 2}
        assert len(responses.calls) <= 2
        if len(responses.calls) == 2:
            request = responses.calls[0].request
            assert request.body == 'grant_type=client_credentials&scope=scope'
            assert request.headers['authorization'] == 'Basic dXNlcjpwYXNz'
        request = responses.calls[-1].request
        assert request.headers['authorization'] == 'Bearer 1234'
        assert request.headers['x-correlation-id'] == 'abcd'
        assert request.headers['Accept'] == 'application/prs.dgfip.part.situations.th.assiettes.v1+json'

    def test_spi_situations_connection_error(self, resource):
        responses.add(
            responses.GET,
            re.compile('https://gw.dgfip.finances.gouv.fr/.*'),
            body=requests.ConnectTimeout(),
        )
        with pytest.raises(APIError, match=r'.*is unavailable'):
            resource.get_spi_situations_ir_assiettes_annrev(numero_fiscal='1234', annee_de_revenu='2023')


class TestEndpoints:
    @pytest.fixture(autouse=True)
    def setup(self):
        with responses._default_mock:
            with mock.patch('passerelle.apps.api_impot_particulier.models.uuid') as uuid_mock:
                uuid_mock.uuid4.return_value.hex = 'abcd'
                responses.add(
                    responses.POST, 'https://gw.dgfip.finances.gouv.fr/token', json={'access_token': '1234'}
                )
                yield None

    @mock.patch(
        'passerelle.apps.api_impot_particulier.models.Resource.get_spi_situations_ir_assiettes_annrev'
    )
    def test_spi_situations_ir_assiettes_annrev(self, mock_get, app, resource):
        mock_get.return_value = 'ok'
        response = app.get(
            '/api-impot-particulier/test/spi-situations-ir-assiettes-annrev',
            params={'numero_fiscal': '1234', 'annee_de_revenu': last_year},
        )
        assert mock_get.call_args[1]['numero_fiscal'] == '1234'
        assert response.json == {'data': 'ok', 'err': 0}

    @mock.patch(
        'passerelle.apps.api_impot_particulier.models.Resource.get_spi_situations_ir_assiettes_annrev'
    )
    def test_spi_situations_ir_assiettes_annrev_strip(self, mock_get, app, resource):
        mock_get.return_value = 'ok'
        response = app.get(
            '/api-impot-particulier/test/spi-situations-ir-assiettes-annrev',
            params={'numero_fiscal': ' 1 2 3 4 ', 'annee_de_revenu': last_year},
        )
        assert mock_get.call_args[1]['numero_fiscal'] == '1234'
        assert response.json == {'data': 'ok', 'err': 0}

    def test_spi_situations_ir_assiettes_annrev_invalid(self, app, resource):
        response = app.get(
            '/api-impot-particulier/test/spi-situations-ir-assiettes-annrev',
            params={'numero_fiscal': '1234', 'annee_de_revenu': '2000'},
        )
        assert response.json['err'] == 1

    @mock.patch(
        'passerelle.apps.api_impot_particulier.models.Resource.get_spi_situations_th_assiettes_principale_annrev'
    )
    def test_spi_situations_th_assiettes_principale_annrev(self, mock_get, app, resource):
        mock_get.return_value = 'ok'
        response = app.get(
            '/api-impot-particulier/test/spi-situations-th-assiettes-principale-annrev',
            params={'numero_fiscal': '1234', 'annee_de_revenu': last_year},
        )
        assert mock_get.call_args[1]['numero_fiscal'] == '1234'
        assert response.json == {'data': 'ok', 'err': 0}