Produits Entr'ouvert » Passerelle

            'interventionData': data,

            'geom': {

                'type': 'Point',

                'coordinates': [post_data['lon'], post_data['lat']],

                'crs': 'EPSG:4326',

            },

            'interventionTypeId': intervention_type['id'],

            'notificationUrl': '%s?uuid=%s' % (endpoint_url['update-intervention'], wcs_request.uuid),

            'add_media_url': '%s?uuid=%s' % (endpoint_url['add-media'], wcs_request.uuid),

        }

        for label in 'checkDuplicated', 'onPrivateLand', 'safeguardRequired':

            if str(post_data.get(label)).lower() in ['true', 'oui', '1']:

                wcs_request.payload[label] = 'true'

        wcs_request.save()

        if not wcs_request.push():

            self.add_job(

    'submitterType': 'usager',

    'external_number': '42-2',

    'external_status': 'statut-1-wcs',

    'address': 'https://wcs.example.com/backoffice/management/foo/2/',

    'interventionData': {'FIELD1': 'Candélabre', 'FIELD2': 42},

    'geom': {'type': 'Point', 'coordinates': [2.323349, 48.833708], 'crs': 'EPSG:4326'},

    'interventionTypeId': '1234',

    'notificationUrl': 'http://testserver/toulouse-smart/test/update-intervention?uuid=%s' % str(UUID),

    'onPrivateLand': 'true',

    'safeguardRequired': 'true',

}

@mock_response(

    ['/v1/type-intervention', None, INTERVENTION_TYPES],

    ['/v1/intervention', CREATE_INTERVENTION_QUERY, get_json_file('create_intervention')],

)

@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)

def test_create_intervention(mocked_uuid4, app, smart):

    with pytest.raises(WcsRequest.DoesNotExist):

        smart.wcs_requests.get(uuid=UUID)

    resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)

    assert not resp.json['err']

    assert resp.json['data']['uuid'] == str(UUID)

    assert resp.json['data']['wcs_form_api_url'] == 'https://wcs.example.com/api/forms/foo/2/'

    wcs_request = smart.wcs_requests.get(uuid=UUID)

    assert wcs_request.wcs_form_api_url == 'https://wcs.example.com/api/forms/foo/2/'

    assert wcs_request.wcs_form_number == '42-2'

    assert wcs_request.payload == CREATE_INTERVENTION_QUERY

    assert wcs_request.result == json.loads(get_json_file('create_intervention'))

    ['/v1/intervention', CREATE_INTERVENTION_QUERY, get_json_file('create_intervention')],

    ['/api/forms/foo/2/hooks/update_intervention/', UPDATE_INTERVENTION_QUERY, WCS_RESPONSE_SUCCESS],

)

@mock.patch("django.db.models.fields.UUIDField.get_default", return_value=UUID)

def test_update_intervention(mocked_uuid, app, smart, wcs_service):

    resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)

    assert not resp.json['err']

    assert CREATE_INTERVENTION_QUERY[

    ] == 'http://testserver/toulouse-smart/test/update-intervention?uuid=%s' % str(UUID)

    wcs_request = smart.wcs_requests.get(uuid=UUID)

    mocked_push = mock.patch(

        "passerelle.contrib.toulouse_smart.models.SmartRequest.push",

        return_value=False,

    )

    mocked_push.start()

    )

    # smart and wcs are down

    mocked_wcs_request_push.start()

    mocked_smart_request_push.start()

    assert Job.objects.count() == 0

    resp = app.post_json(URL + 'create-intervention/', params=CREATE_INTERVENTION_PAYLOAD)

    assert not resp.json['err']

    assert resp.json['data']['uuid'] == str(UUID)

    assert resp.json['data']['wcs_form_api_url'] == 'https://wcs.example.com/api/forms/foo/2/'

    wcs_request = smart.wcs_requests.get(uuid=UUID)

    assert wcs_request.wcs_form_api_url == 'https://wcs.example.com/api/forms/foo/2/'

    assert wcs_request.wcs_form_number == '42-2'

    assert wcs_request.payload == CREATE_INTERVENTION_QUERY

    assert wcs_request.status == 'registered'