darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-02-01 15:46:07 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1933 from alphagov/delete-service-stuff

Browse Source 
Delete service apis (callback + inbound)
This commit is contained in:
Leo Hemsted
2018-07-06 15:19:01 +01:00
committed by GitHub
parent f4cb3ab9b5 32415b3b14
commit 6705f9d209
6 changed files with 160 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
tests/app/conftest.py
View File

@@ -1106,7 +1106,7 @@ def admin_request(client):
url_for(endpoint, **(endpoint_kwargs or {})),
headers=[create_authorization_header()]
)
json_resp = json.loads(resp.get_data(as_text=True))
json_resp = resp.json
assert resp.status_code == _expected_status
return json_resp
@@ -1118,7 +1118,7 @@ def admin_request(client):
headers=[('Content-Type', 'application/json'), create_authorization_header()]
)
if resp.get_data():
json_resp = json.loads(resp.get_data(as_text=True))
json_resp = resp.json
else:
json_resp = None
assert resp.status_code == _expected_status
@@ -1130,7 +1130,10 @@ def admin_request(client):
url_for(endpoint, **(endpoint_kwargs or {})),
headers=[create_authorization_header()]
)
json_resp = json.loads(resp.get_data(as_text=True))
if resp.get_data():
json_resp = resp.json
else:
json_resp = None
assert resp.status_code == _expected_status, json_resp
return json_resp

175
tests/app/service/test_callback_rest.py
View File

@@ -1,28 +1,27 @@
import json
import uuid
from tests import create_authorization_header
from tests.app.db import (
create_service_inbound_api,
create_service_callback_api
)
from app.models import ServiceInboundApi, ServiceCallbackApi
def test_create_service_inbound_api(client, sample_service):
def test_create_service_inbound_api(admin_request, sample_service):
data = {
"url": "https://some_service/inbound-sms",
"bearer_token": "some-unique-string",
"updated_by_id": str(sample_service.users[0].id)
}
response = client.post(
'/service/{}/inbound-api'.format(sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
resp_json = admin_request.post(
'service_callback.create_service_inbound_api',
service_id=sample_service.id,
_data=data,
_expected_status=201
)
assert response.status_code == 201
resp_json = json.loads(response.get_data(as_text=True))["data"]
resp_json = resp_json["data"]
assert resp_json["id"]
assert resp_json["service_id"] == str(sample_service.id)
assert resp_json["url"] == "https://some_service/inbound-sms"
@@ -31,22 +30,22 @@ def test_create_service_inbound_api(client, sample_service):
assert not resp_json["updated_at"]
def test_set_service_inbound_api_raises_404_when_service_does_not_exist(client):
def test_set_service_inbound_api_raises_404_when_service_does_not_exist(admin_request):
data = {
"url": "https://some_service/inbound-sms",
"bearer_token": "some-unique-string",
"updated_by_id": str(uuid.uuid4())
}
response = client.post(
'/service/{}/inbound-api'.format(uuid.uuid4()),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
response = admin_request.post(
'service_callback.create_service_inbound_api',
service_id=uuid.uuid4(),
_data=data,
_expected_status=404
)
assert response.status_code == 404
assert json.loads(response.get_data(as_text=True))['message'] == 'No result found'
assert response['message'] == 'No result found'
def test_update_service_inbound_api_updates_url(client, sample_service):
def test_update_service_inbound_api_updates_url(admin_request, sample_service):
service_inbound_api = create_service_inbound_api(service=sample_service,
url="https://original_url.com")
@@ -54,53 +53,74 @@ def test_update_service_inbound_api_updates_url(client, sample_service):
"url": "https://another_url.com",
"updated_by_id": str(sample_service.users[0].id)
}
response = client.post("/service/{}/inbound-api/{}".format(sample_service.id, service_inbound_api.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()])
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))["data"]
assert resp_json["url"] == "https://another_url.com"
response = admin_request.post(
'service_callback.update_service_inbound_api',
service_id=sample_service.id,
inbound_api_id=service_inbound_api.id,
_data=data
)
assert response["data"]["url"] == "https://another_url.com"
assert service_inbound_api.url == "https://another_url.com"
def test_update_service_inbound_api_updates_bearer_token(client, sample_service):
def test_update_service_inbound_api_updates_bearer_token(admin_request, sample_service):
service_inbound_api = create_service_inbound_api(service=sample_service,
bearer_token="some_super_secret")
data = {
"bearer_token": "different_token",
"updated_by_id": str(sample_service.users[0].id)
}
response = client.post("/service/{}/inbound-api/{}".format(sample_service.id, service_inbound_api.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()])
assert response.status_code == 200
admin_request.post(
'service_callback.update_service_inbound_api',
service_id=sample_service.id,
inbound_api_id=service_inbound_api.id,
_data=data
)
assert service_inbound_api.bearer_token == "different_token"
def test_fetch_service_inbound_api(client, sample_service):
def test_fetch_service_inbound_api(admin_request, sample_service):
service_inbound_api = create_service_inbound_api(service=sample_service)
response = client.get("/service/{}/inbound-api/{}".format(sample_service.id, service_inbound_api.id),
headers=[create_authorization_header()])
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True))["data"] == service_inbound_api.serialize()
response = admin_request.get(
'service_callback.fetch_service_inbound_api',
service_id=sample_service.id,
inbound_api_id=service_inbound_api.id,
)
assert response["data"] == service_inbound_api.serialize()
def test_create_service_callback_api(client, sample_service):
def test_delete_service_inbound_api(admin_request, sample_service):
service_inbound_api = create_service_inbound_api(sample_service)
response = admin_request.delete(
'service_callback.remove_service_inbound_api',
service_id=sample_service.id,
inbound_api_id=service_inbound_api.id,
)
assert response is None
assert ServiceInboundApi.query.count() == 0
def test_create_service_callback_api(admin_request, sample_service):
data = {
"url": "https://some_service/delivery-receipt-endpoint",
"bearer_token": "some-unique-string",
"updated_by_id": str(sample_service.users[0].id)
}
response = client.post(
'/service/{}/delivery-receipt-api'.format(sample_service.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
)
assert response.status_code == 201
resp_json = json.loads(response.get_data(as_text=True))["data"]
resp_json = admin_request.post(
'service_callback.create_service_callback_api',
service_id=sample_service.id,
_data=data,
_expected_status=201
)
resp_json = resp_json["data"]
assert resp_json["id"]
assert resp_json["service_id"] == str(sample_service.id)
assert resp_json["url"] == "https://some_service/delivery-receipt-endpoint"
@@ -109,22 +129,23 @@ def test_create_service_callback_api(client, sample_service):
assert not resp_json["updated_at"]
def test_set_service_callback_api_raises_404_when_service_does_not_exist(client, notify_db_session):
def test_set_service_callback_api_raises_404_when_service_does_not_exist(admin_request, notify_db_session):
data = {
"url": "https://some_service/delivery-receipt-endpoint",
"bearer_token": "some-unique-string",
"updated_by_id": str(uuid.uuid4())
}
response = client.post(
'/service/{}/delivery-receipt-api'.format(uuid.uuid4()),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()]
resp_json = admin_request.post(
'service_callback.create_service_callback_api',
service_id=uuid.uuid4(),
_data=data,
_expected_status=404
)
assert response.status_code == 404
assert json.loads(response.get_data(as_text=True))['message'] == 'No result found'
assert resp_json['message'] == 'No result found'
def test_update_service_callback_api_updates_url(client, sample_service):
def test_update_service_callback_api_updates_url(admin_request, sample_service):
service_callback_api = create_service_callback_api(service=sample_service,
url="https://original_url.com")
@@ -132,34 +153,54 @@ def test_update_service_callback_api_updates_url(client, sample_service):
"url": "https://another_url.com",
"updated_by_id": str(sample_service.users[0].id)
}
response = client.post("/service/{}/delivery-receipt-api/{}".format(sample_service.id, service_callback_api.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()])
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))["data"]
assert resp_json["url"] == "https://another_url.com"
resp_json = admin_request.post(
'service_callback.update_service_callback_api',
service_id=sample_service.id,
callback_api_id=service_callback_api.id,
_data=data
)
assert resp_json["data"]["url"] == "https://another_url.com"
assert service_callback_api.url == "https://another_url.com"
def test_update_service_callback_api_updates_bearer_token(client, sample_service):
def test_update_service_callback_api_updates_bearer_token(admin_request, sample_service):
service_callback_api = create_service_callback_api(service=sample_service,
bearer_token="some_super_secret")
data = {
"bearer_token": "different_token",
"updated_by_id": str(sample_service.users[0].id)
}
response = client.post("/service/{}/delivery-receipt-api/{}".format(sample_service.id, service_callback_api.id),
data=json.dumps(data),
headers=[('Content-Type', 'application/json'), create_authorization_header()])
assert response.status_code == 200
admin_request.post(
'service_callback.update_service_callback_api',
service_id=sample_service.id,
callback_api_id=service_callback_api.id,
_data=data
)
assert service_callback_api.bearer_token == "different_token"
def test_fetch_service_callback_api(client, sample_service):
def test_fetch_service_callback_api(admin_request, sample_service):
service_callback_api = create_service_callback_api(service=sample_service)
response = client.get("/service/{}/delivery-receipt-api/{}".format(sample_service.id, service_callback_api.id),
headers=[create_authorization_header()])
response = admin_request.get(
'service_callback.fetch_service_callback_api',
service_id=sample_service.id,
callback_api_id=service_callback_api.id,
)
assert response.status_code == 200
assert json.loads(response.get_data(as_text=True))["data"] == service_callback_api.serialize()
assert response["data"] == service_callback_api.serialize()
def test_delete_service_callback_api(admin_request, sample_service):
service_callback_api = create_service_callback_api(sample_service)
response = admin_request.delete(
'service_callback.remove_service_callback_api',
service_id=sample_service.id,
callback_api_id=service_callback_api.id,
)
assert response is None
assert ServiceCallbackApi.query.count() == 0