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

Service and User API added, working with tests. Still need to polish the edges and add more tests.

Browse Source
This commit is contained in:
Nicholas Staples
2016-01-11 15:07:13 +00:00
parent 0bc4d02713
commit 49e98c21e7
14 changed files with 282 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
tests/app/conftest.py
View File

@@ -1,14 +1,16 @@
import pytest
from app.dao.users_dao import (create_user, get_users)
from app.dao.services_dao import (create_service, get_services)
from app.models import (User, Service)
from app.dao.users_dao import (create_model_user, get_model_users)
from app.dao.services_dao import create_model_service
@pytest.fixture(scope='function')
def sample_user(notify_db,
notify_db_session,
email="notify@digital.cabinet-office.gov.uk"):
user_id = create_user(email)
return get_users(user_id=user_id)
user = User(**{'email_address': email})
create_model_user(user)
return user
@pytest.fixture(scope='function')
@@ -18,5 +20,12 @@ def sample_service(notify_db,
user=None):
if user is None:
user = sample_user(notify_db, notify_db_session)
service_id = create_service(service_name, user)
return get_services(service_id=service_id)
data = {
'name': service_name,
'users': [user],
'limit': 1000,
'active': False,
'restricted': False}
service = Service(**data)
create_model_service(service)
return service

38
tests/app/dao/test_services_dao.py
View File

@@ -1,4 +1,6 @@
from app.dao.services_dao import (create_service, get_services)
import pytest
from app.dao.services_dao import (
create_model_service, get_model_services, DAOException)
from tests.app.conftest import sample_service as create_sample_service
from app.models import Service
@@ -6,10 +8,17 @@ from app.models import Service
def test_create_service(notify_api, notify_db, notify_db_session, sample_user):
assert Service.query.count() == 0
service_name = 'Sample Service'
service_id = create_service(service_name, sample_user)
data = {
'name': service_name,
'users': [sample_user],
'limit': 1000,
'active': False,
'restricted': False}
service = Service(**data)
create_model_service(service)
assert Service.query.count() == 1
assert Service.query.first().name == service_name
assert Service.query.first().id == service_id
assert Service.query.first().id == service.id
def test_get_services(notify_api, notify_db, notify_db_session, sample_user):
@@ -17,14 +26,14 @@ def test_get_services(notify_api, notify_db, notify_db_session, sample_user):
notify_db_session,
user=sample_user)
assert Service.query.count() == 1
assert len(get_services()) == 1
assert len(get_model_services()) == 1
service_name = "Another service"
sample_service = create_sample_service(notify_db,
notify_db_session,
service_name=service_name,
user=sample_user)
assert Service.query.count() == 2
assert len(get_services()) == 2
assert len(get_model_services()) == 2
def test_get_user_service(notify_api, notify_db, notify_db_session, sample_user):
@@ -34,5 +43,22 @@ def test_get_user_service(notify_api, notify_db, notify_db_session, sample_user)
notify_db_session,
service_name=service_name,
user=sample_user)
assert get_services(service_id=sample_service.id).name == service_name
assert get_model_services(service_id=sample_service.id).name == service_name
assert Service.query.count() == 1
def test_missing_user_attribute(notify_api, notify_db, notify_db_session):
assert Service.query.count() == 0
try:
service_name = 'Sample Service'
data = {
'name': service_name,
'limit': 1000,
'active': False,
'restricted': False}
service = Service(**data)
create_model_service(service)
pytest.fail("DAOException not thrown")
except DAOException as e:
assert "Missing data for required attribute" in str(e)

17
tests/app/dao/test_users_dao.py
View File

@@ -1,27 +1,28 @@
from sqlalchemy.exc import DataError
from sqlalchemy.orm.exc import NoResultFound
from app.dao.users_dao import (create_user, get_users)
from app.dao.users_dao import (create_model_user, get_model_users)
from tests.app.conftest import sample_user as create_sample_user
from app.models import User
def test_create_user(notify_api, notify_db, notify_db_session):
email = 'notify@digital.cabinet-office.gov.uk'
user_id = create_user(email)
user = User(**{'email_address': email})
create_model_user(user)
assert User.query.count() == 1
assert User.query.first().email_address == email
assert User.query.filter_by(id=user_id).one()
assert User.query.first().id == user.id
def test_get_all_users(notify_api, notify_db, notify_db_session, sample_user):
assert User.query.count() == 1
assert len(get_users()) == 1
assert len(get_model_users()) == 1
email = "another.notify@digital.cabinet-office.gov.uk"
another_user = create_sample_user(notify_db,
notify_db_session,
email=email)
assert User.query.count() == 2
assert len(get_users()) == 2
assert len(get_model_users()) == 2
def test_get_user(notify_api, notify_db, notify_db_session):
@@ -29,12 +30,12 @@ def test_get_user(notify_api, notify_db, notify_db_session):
another_user = create_sample_user(notify_db,
notify_db_session,
email=email)
assert get_users(user_id=another_user.id).email_address == email
assert get_model_users(user_id=another_user.id).email_address == email
def test_get_user_not_exists(notify_api, notify_db, notify_db_session):
try:
get_users(user_id="12345")
get_model_users(user_id="12345")
pytest.fail("NoResultFound exception not thrown.")
except:
pass
@@ -42,7 +43,7 @@ def test_get_user_not_exists(notify_api, notify_db, notify_db_session):
def test_get_user_invalid_id(notify_api, notify_db, notify_db_session):
try:
get_users(user_id="blah")
get_model_users(user_id="blah")
pytest.fail("DataError exception not thrown.")
except DataError:
pass

10
tests/app/main/views/test_authentication.py
View File

@@ -1,7 +1,9 @@
import pytest
from flask import json
from client.authentication import create_jwt_token
@pytest.mark.xfail(reason="Authentication to be added.")
def test_should_not_allow_request_with_no_token(notify_api):
response = notify_api.test_client().get("/")
assert response.status_code == 401
@@ -9,6 +11,7 @@ def test_should_not_allow_request_with_no_token(notify_api):
assert data['error'] == 'Unauthorized, authentication token must be provided'
@pytest.mark.xfail(reason="Authentication to be added.")
def test_should_not_allow_request_with_incorrect_header(notify_api):
response = notify_api.test_client().get(
"/",
@@ -21,6 +24,7 @@ def test_should_not_allow_request_with_incorrect_header(notify_api):
assert data['error'] == 'Unauthorized, authentication bearer scheme must be used'
@pytest.mark.xfail(reason="Authentication to be added.")
def test_should_not_allow_request_with_incorrect_token(notify_api):
response = notify_api.test_client().get(
"/",
@@ -33,6 +37,7 @@ def test_should_not_allow_request_with_incorrect_token(notify_api):
assert data['error'] == 'Invalid token: signature'
@pytest.mark.xfail(reason="Authentication to be added.")
def test_should_not_allow_incorrect_path(notify_api):
token = create_jwt_token(request_method="GET", request_path="/bad", secret="secret", client_id="client_id")
response = notify_api.test_client().get(
@@ -46,6 +51,7 @@ def test_should_not_allow_incorrect_path(notify_api):
assert data['error'] == 'Invalid token: request'
@pytest.mark.xfail(reason="Authentication to be added.")
def test_should_not_allow_incorrect_method(notify_api):
token = create_jwt_token(request_method="POST", request_path="/", secret="secret", client_id="client_id")
response = notify_api.test_client().get(
@@ -59,6 +65,7 @@ def test_should_not_allow_incorrect_method(notify_api):
assert data['error'] == 'Invalid token: request'
@pytest.mark.xfail(reason="Authentication to be added.")
def test_should_not_allow_invalid_secret(notify_api):
token = create_jwt_token(request_method="POST", request_path="/", secret="not-so-secret", client_id="client_id")
response = notify_api.test_client().get(
@@ -72,6 +79,7 @@ def test_should_not_allow_invalid_secret(notify_api):
assert data['error'] == 'Invalid token: signature'
@pytest.mark.xfail(reason="Authentication to be added.")
def test_should_allow_valid_token(notify_api):
token = create_jwt_token(request_method="GET", request_path="/", secret="secret", client_id="client_id")
response = notify_api.test_client().get(
@@ -83,6 +91,7 @@ def test_should_allow_valid_token(notify_api):
assert response.status_code == 200
@pytest.mark.xfail(reason="Authentication to be added.")
def test_should_allow_valid_token_with_post_body(notify_api):
json_body = json.dumps({
"key1": "value1",
@@ -106,6 +115,7 @@ def test_should_allow_valid_token_with_post_body(notify_api):
assert response.status_code == 200
@pytest.mark.xfail(reason="Authentication to be added.")
def test_should_not_allow_valid_token_with_invalid_post_body(notify_api):
json_body = json.dumps({
"key1": "value1",

34
tests/app/service/views/test_rest.py
View File

@@ -1,20 +1,27 @@
import json
from app.models import Service
from flask import url_for
def test_get_service_list(notify_api, notify_db, notify_db_session, sample_service):
"""
Tests GET endpoint '/' to retrieve entire service list.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
response = client.get(url_for('service.get_service'))
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
# TODO assert correct json returned
assert len(json_resp['data']) == 1
assert len(json_resp) == 1
assert json_resp['data'][0]['name'] == sample_service.name
assert json_resp['data'][0]['id'] == sample_service.id
def test_get_service(notify_api, notify_db, notify_db_session, sample_service):
"""
Tests GET endpoint '/<service_id>' to retrieve a single service.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
resp = client.get(url_for('service.get_service',
@@ -25,8 +32,29 @@ def test_get_service(notify_api, notify_db, notify_db_session, sample_service):
assert json_resp['data']['id'] == sample_service.id
def test_post_service(notify_api, notify_db, notify_db_session, sample_service):
pass
def test_post_service(notify_api, notify_db, notify_db_session, sample_user):
"""
Tests POST endpoint '/' to create a service.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert Service.query.count() == 0
data = {
'name': 'created service',
'users': [sample_user.id],
'limit': 1000,
'restricted': False,
'active': False}
headers = [('Content-Type', 'application/json')]
resp = client.post(
url_for('service.create_service'),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 201
service = Service.query.first()
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['name'] == service.name
assert json_resp['data']['limit'] == service.limit
def test_put_service(notify_api, notify_db, notify_db_session, sample_service):

57
tests/app/user/views/test_rest.py
View File

@@ -0,0 +1,57 @@
import json
from app.models import User
from flask import url_for
def test_get_user_list(notify_api, notify_db, notify_db_session, sample_user):
"""
Tests GET endpoint '/' to retrieve entire user list.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
response = client.get(url_for('user.get_user'))
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
# TODO assert correct json returned
assert len(json_resp['data']) == 1
assert json_resp['data'][0]['email_address'] == sample_user.email_address
assert json_resp['data'][0]['id'] == sample_user.id
def test_get_user(notify_api, notify_db, notify_db_session, sample_user):
"""
Tests GET endpoint '/<user_id>' to retrieve a single service.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
resp = client.get(url_for('user.get_user',
user_id=sample_user.id))
assert resp.status_code == 200
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['email_address'] == sample_user.email_address
assert json_resp['data']['id'] == sample_user.id
def test_post_user(notify_api, notify_db, notify_db_session):
"""
Tests POST endpoint '/' to create a user.
"""
with notify_api.test_request_context():
with notify_api.test_client() as client:
assert User.query.count() == 0
data = {
'email_address': 'user@digital.cabinet-office.gov.uk'}
headers = [('Content-Type', 'application/json')]
resp = client.post(
url_for('user.create_user'),
data=json.dumps(data),
headers=headers)
assert resp.status_code == 201
user = User.query.first()
json_resp = json.loads(resp.get_data(as_text=True))
assert json_resp['data']['email_address'] == user.email_address
assert json_resp['data']['id'] == user.id
def test_put_user(notify_api, notify_db, notify_db_session, sample_user):
pass