darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-08-11 09:27:56 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #723 from alphagov/active-service

Browse Source 
deactivate a service
This commit is contained in:
Leo Hemsted
2016-11-11 15:25:32 +00:00
committed by GitHub
parent 10f9dfde6e 195f3615e6
commit 95764af6ea
13 changed files with 312 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
app/authentication/auth.py
View File

@@ -53,10 +53,19 @@ def requires_auth():
return handle_admin_key(auth_token, current_app.config.get('ADMIN_CLIENT_SECRET'))
try:
api_keys = get_model_api_keys(client)
service = dao_fetch_service_by_id(client)
except DataError:
raise AuthError("Invalid token: service id is not the right data type", 403)
for api_key in api_keys:
except NoResultFound:
raise AuthError("Invalid token: service not found", 403)
if not service.api_keys:
raise AuthError("Invalid token: service has no API keys", 403)
if not service.active:
raise AuthError("Invalid token: service is archived", 403)
for api_key in service.api_keys:
try:
get_decode_errors(auth_token, api_key.unsigned_secret)
except TokenDecodeError:
@@ -67,15 +76,8 @@ def requires_auth():
_request_ctx_stack.top.api_user = api_key
return
try:
dao_fetch_service_by_id(client)
except NoResultFound:
raise AuthError("Invalid token: service not found", 403)
if not api_keys:
raise AuthError("Invalid token: service has no API keys", 403)
else:
# service has API keys, but none matching the one the user provided
raise AuthError("Invalid token: signature, api token is not valid", 403)

2
app/dao/dao_utils.py
View File

@@ -9,7 +9,6 @@ def transactional(func):
@wraps(func)
def commit_or_rollback(*args, **kwargs):
from flask import current_app
from app import db
try:
res = func(*args, **kwargs)
db.session.commit()
@@ -27,7 +26,6 @@ def version_class(model_class, history_cls=None):
def versioned(func):
@wraps(func)
def record_version(*args, **kwargs):
from app import db
func(*args, **kwargs)
history_objects = [create_hist(obj) for obj in
itertools.chain(db.session.new, db.session.dirty)

59
app/dao/services_dao.py
View File

@@ -1,5 +1,5 @@
import uuid
from datetime import date
from datetime import date, datetime
from sqlalchemy import asc, func
from sqlalchemy.orm import joinedload
@@ -28,30 +28,68 @@ from app.models import (
from app.statsd_decorators import statsd
def dao_fetch_all_services():
return Service.query.order_by(
def dao_fetch_all_services(only_active=False):
query = Service.query.order_by(
asc(Service.created_at)
).options(
joinedload('users')
).all()
)
if only_active:
query = query.filter(Service.active)
return query.all()
def dao_fetch_service_by_id(service_id):
return Service.query.filter_by(
def dao_fetch_service_by_id(service_id, only_active=False):
query = Service.query.filter_by(
id=service_id
).options(
joinedload('users')
).one()
)
if only_active:
query = query.filter(Service.active)
return query.one()
def dao_fetch_all_services_by_user(user_id):
return Service.query.filter(
def dao_fetch_all_services_by_user(user_id, only_active=False):
query = Service.query.filter(
Service.users.any(id=user_id)
).order_by(
asc(Service.created_at)
).options(
joinedload('users')
).all()
)
if only_active:
query = query.filter(Service.active)
return query.all()
@transactional
@version_class(Service)
@version_class(Template, TemplateHistory)
@version_class(ApiKey)
def dao_deactive_service(service_id):
# have to eager load templates and api keys so that we don't flush when we loop through them
# to ensure that db.session still contains the models when it comes to creating history objects
service = Service.query.options(
joinedload('templates'),
joinedload('api_keys'),
).filter(Service.id == service_id).one()
service.active = False
service.name = '_archived_' + service.name
service.email_from = '_archived_' + service.email_from
for template in service.templates:
template.archived = True
for api_key in service.api_keys:
api_key.expiry_date = datetime.utcnow()
def dao_fetch_service_by_id_and_user(service_id, user_id):
@@ -70,6 +108,7 @@ def dao_create_service(service, user):
service.users.append(user)
permission_dao.add_default_service_permissions_for_user(user, service)
service.id = uuid.uuid4() # must be set now so version history model can use same id
service.active = True
service.research_mode = False
db.session.add(service)

6
app/models.py
View File

@@ -115,7 +115,7 @@ class Service(db.Model, Versioned):
unique=False,
nullable=True,
onupdate=datetime.datetime.utcnow)
active = db.Column(db.Boolean, index=False, unique=False, nullable=False)
active = db.Column(db.Boolean, index=False, unique=False, nullable=False, default=True)
message_limit = db.Column(db.BigInteger, index=False, unique=False, nullable=False)
users = db.relationship(
'User',
@@ -188,7 +188,7 @@ class ApiKey(db.Model, Versioned):
name = db.Column(db.String(255), nullable=False)
secret = db.Column(db.String(255), unique=True, nullable=False)
service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'), index=True, nullable=False)
service = db.relationship('Service', backref=db.backref('api_keys', lazy='dynamic'))
service = db.relationship('Service', backref='api_keys')
key_type = db.Column(db.String(255), db.ForeignKey('key_types.name'), index=True, nullable=False)
expiry_date = db.Column(db.DateTime)
created_at = db.Column(
@@ -275,7 +275,7 @@ class Template(db.Model):
content = db.Column(db.Text, index=False, unique=False, nullable=False)
archived = db.Column(db.Boolean, index=False, nullable=False, default=False)
service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'), index=True, unique=False, nullable=False)
service = db.relationship('Service', backref=db.backref('templates', lazy='dynamic'))
service = db.relationship('Service', backref='templates')
subject = db.Column(db.Text, index=False, unique=False, nullable=True)
created_by_id = db.Column(UUID(as_uuid=True), db.ForeignKey('users.id'), index=True, nullable=False)
created_by = db.relationship('User')

31
app/service/rest.py
View File

@@ -25,7 +25,8 @@ from app.dao.services_dao import (
dao_fetch_stats_for_service,
dao_fetch_todays_stats_for_service,
dao_fetch_weekly_historical_stats_for_service,
dao_fetch_todays_stats_for_all_services
dao_fetch_todays_stats_for_all_services,
dao_deactive_service
)
from app.dao.service_whitelist_dao import (
dao_fetch_service_whitelist,
@@ -58,13 +59,16 @@ register_errors(service_blueprint)
@service_blueprint.route('', methods=['GET'])
def get_services():
only_active = request.args.get('only_active') == 'True'
detailed = request.args.get('detailed') == 'True'
user_id = request.args.get('user_id', None)
if user_id:
services = dao_fetch_all_services_by_user(user_id)
elif request.args.get('detailed') == 'True':
return jsonify(data=get_detailed_services())
services = dao_fetch_all_services_by_user(user_id, only_active)
elif detailed:
return jsonify(data=get_detailed_services(only_active))
else:
services = dao_fetch_all_services()
services = dao_fetch_all_services(only_active)
data = service_schema.dump(services, many=True).data
return jsonify(data=data)
@@ -264,8 +268,8 @@ def get_detailed_service(service_id, today_only=False):
return detailed_service_schema.dump(service).data
def get_detailed_services():
services = {service.id: service for service in dao_fetch_all_services()}
def get_detailed_services(only_active=False):
services = {service.id: service for service in dao_fetch_all_services(only_active)}
stats = dao_fetch_todays_stats_for_all_services()
for service_id, rows in itertools.groupby(stats, lambda x: x.service_id):
@@ -312,6 +316,19 @@ def update_whitelist(service_id):
return '', 204
@service_blueprint.route('/<uuid:service_id>/deactivate', methods=['POST'])
def deactivate_service(service_id):
service = dao_fetch_service_by_id(service_id)
if not service.active:
# assume already inactive, don't change service name
return '', 204
dao_deactive_service(service.id)
return '', 204
@service_blueprint.route('/<uuid:service_id>/billable-units')
def get_billable_unit_count(service_id):
try:

23
migrations/versions/0059_set_services_to_active.py Normal file
View File

@@ -0,0 +1,23 @@
"""
we weren't previously using the services.active column , and by default it was set to false. lets set all services to
active, so that in the future we can turn it off to signify deactivating a service
Revision ID: 0059_set_services_to_active
Revises: 0058_add_letters_flag
Create Date: 2016-10-31 15:17:16.716450
"""
# revision identifiers, used by Alembic.
revision = '0059_set_services_to_active'
down_revision = '0058_add_letters_flag'
from alembic import op
def upgrade():
op.execute('UPDATE services SET active = TRUE')
def downgrade():
op.execute('UPDATE services SET active = FALSE')

9
tests/__init__.py
View File

@@ -30,3 +30,12 @@ def create_authorization_header(service_id=None, key_type=KEY_TYPE_NORMAL):
token = create_jwt_token(secret=secret, client_id=client_id)
return 'Authorization', 'Bearer {}'.format(token)
def unwrap_function(fn):
"""
Given a function, returns its undecorated original.
"""
while hasattr(fn, '__wrapped__'):
fn = fn.__wrapped__
return fn

11
tests/app/authentication/test_authentication.py
View File

@@ -266,6 +266,17 @@ def test_authentication_returns_error_when_service_doesnt_exit(
assert error_message['message'] == {'token': ['Invalid token: service not found']}
def test_authentication_returns_error_when_service_inactive(client, sample_api_key):
sample_api_key.service.active = False
token = create_jwt_token(secret=str(sample_api_key.id), client_id=str(sample_api_key.service_id))
response = client.get('/service', headers={'Authorization': 'Bearer {}'.format(token)})
assert response.status_code == 403
error_message = json.loads(response.get_data())
assert error_message['message'] == {'token': ['Invalid token: service is archived']}
def test_authentication_returns_error_when_service_has_no_secrets(notify_api,
sample_service,
fake_uuid):

3
tests/app/conftest.py
View File

@@ -129,6 +129,7 @@ def sample_service(notify_db,
notify_db_session,
service_name="Sample service",
user=None,
active=True,
restricted=False,
limit=1000,
email_from=None):
@@ -139,7 +140,7 @@ def sample_service(notify_db,
data = {
'name': service_name,
'message_limit': limit,
'active': False,
'active': active,
'restricted': restricted,
'email_from': email_from,
'created_by': user

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

@@ -58,7 +58,6 @@ def test_create_service(sample_user):
service = Service(name="service_name",
email_from="email_from",
message_limit=1000,
active=True,
restricted=False,
created_by=sample_user)
dao_create_service(service, sample_user)
@@ -68,7 +67,8 @@ def test_create_service(sample_user):
assert service_db.name == "service_name"
assert service_db.id == service.id
assert service_db.branding == BRANDING_GOVUK
assert not service_db.research_mode
assert service_db.research_mode is False
assert service.active is True
assert sample_user in service_db.users
@@ -77,14 +77,12 @@ def test_cannot_create_two_services_with_same_name(sample_user):
service1 = Service(name="service_name",
email_from="email_from1",
message_limit=1000,
active=True,
restricted=False,
created_by=sample_user)
service2 = Service(name="service_name",
email_from="email_from2",
message_limit=1000,
active=True,
restricted=False,
created_by=sample_user)
with pytest.raises(IntegrityError) as excinfo:
@@ -98,13 +96,11 @@ def test_cannot_create_two_services_with_same_email_from(sample_user):
service1 = Service(name="service_name1",
email_from="email_from",
message_limit=1000,
active=True,
restricted=False,
created_by=sample_user)
service2 = Service(name="service_name2",
email_from="email_from",
message_limit=1000,
active=True,
restricted=False,
created_by=sample_user)
with pytest.raises(IntegrityError) as excinfo:
@@ -118,7 +114,6 @@ def test_cannot_create_service_with_no_user(notify_db_session, sample_user):
service = Service(name="service_name",
email_from="email_from",
message_limit=1000,
active=True,
restricted=False,
created_by=sample_user)
with pytest.raises(FlushError) as excinfo:
@@ -130,7 +125,6 @@ def test_should_add_user_to_service(sample_user):
service = Service(name="service_name",
email_from="email_from",
message_limit=1000,
active=True,
restricted=False,
created_by=sample_user)
dao_create_service(service, sample_user)
@@ -150,7 +144,6 @@ def test_should_remove_user_from_service(sample_user):
service = Service(name="service_name",
email_from="email_from",
message_limit=1000,
active=True,
restricted=False,
created_by=sample_user)
dao_create_service(service, sample_user)
@@ -244,7 +237,6 @@ def test_create_service_creates_a_history_record_with_current_data(sample_user):
service = Service(name="service_name",
email_from="email_from",
message_limit=1000,
active=True,
restricted=False,
created_by=sample_user)
dao_create_service(service, sample_user)
@@ -270,7 +262,6 @@ def test_update_service_creates_a_history_record_with_current_data(sample_user):
service = Service(name="service_name",
email_from="email_from",
message_limit=1000,
active=True,
restricted=False,
created_by=sample_user)
dao_create_service(service, sample_user)
@@ -299,7 +290,6 @@ def test_create_service_and_history_is_transactional(sample_user):
service = Service(name=None,
email_from="email_from",
message_limit=1000,
active=True,
restricted=False,
created_by=sample_user)
@@ -348,7 +338,6 @@ def test_add_existing_user_to_another_service_doesnot_change_old_permissions(sam
service_one = Service(name="service_one",
email_from="service_one",
message_limit=1000,
active=True,
restricted=False,
created_by=sample_user)
@@ -367,7 +356,6 @@ def test_add_existing_user_to_another_service_doesnot_change_old_permissions(sam
service_two = Service(name="service_two",
email_from="service_two",
message_limit=1000,
active=True,
restricted=False,
created_by=other_user)
dao_create_service(service_two, other_user)
@@ -397,7 +385,6 @@ def test_fetch_stats_filters_on_service(sample_notification):
service_two = Service(name="service_two",
created_by=sample_notification.service.created_by,
email_from="hello",
active=False,
restricted=False,
message_limit=1000)
dao_create_service(service_two, sample_notification.service.created_by)

89
tests/app/service/test_deactivate.py Normal file
View File

@@ -0,0 +1,89 @@
import uuid
from unittest import mock
import pytest
from app import db
from app.models import Service, TemplateHistory, ApiKey
from app.dao.services_dao import dao_deactive_service
from tests import create_authorization_header, unwrap_function
from tests.app.conftest import (
sample_template as create_template,
sample_api_key as create_api_key
)
def test_deactivate_only_allows_post(client):
auth_header = create_authorization_header()
response = client.get('/service/{}/deactivate'.format(uuid.uuid4()), headers=[auth_header])
assert response.status_code == 405
def test_deactivate_service_errors_with_bad_service_id(client):
auth_header = create_authorization_header()
response = client.post('/service/{}/deactivate'.format(uuid.uuid4()), headers=[auth_header])
assert response.status_code == 404
def test_deactivating_inactive_service_does_nothing(client, sample_service):
auth_header = create_authorization_header()
sample_service.active = False
response = client.post('/service/{}/deactivate'.format(sample_service.id), headers=[auth_header])
assert response.status_code == 204
assert sample_service.name == 'Sample service'
@pytest.fixture
def deactivated_service(client, notify_db, notify_db_session, sample_service):
create_template(notify_db, notify_db_session, template_name='a')
create_template(notify_db, notify_db_session, template_name='b')
create_api_key(notify_db, notify_db_session)
create_api_key(notify_db, notify_db_session)
auth_header = create_authorization_header()
response = client.post('/service/{}/deactivate'.format(sample_service.id), headers=[auth_header])
assert response.status_code == 204
assert response.data == b''
return sample_service
def test_deactivating_service_changes_name_and_email(deactivated_service):
assert deactivated_service.name == '_archived_Sample service'
assert deactivated_service.email_from == '_archived_sample.service'
def test_deactivating_service_revokes_api_keys(deactivated_service):
assert len(deactivated_service.api_keys) == 2
for key in deactivated_service.api_keys:
assert key.expiry_date is not None
assert key.version == 2
def test_deactivating_service_archives_templates(deactivated_service):
assert len(deactivated_service.templates) == 2
for template in deactivated_service.templates:
assert template.archived is True
assert template.version == 2
def test_deactivating_service_creates_history(deactivated_service):
ServiceHistory = Service.get_history_model()
history = ServiceHistory.query.filter_by(
id=deactivated_service.id
).order_by(
ServiceHistory.version.desc()
).first()
assert history.version == 2
assert history.active is False
def test_deactivating_service_rolls_back_everything_on_error(sample_service, sample_api_key, sample_template):
unwrapped_deactive_service = unwrap_function(dao_deactive_service)
unwrapped_deactive_service(sample_service.id)
assert sample_service in db.session.dirty
assert sample_api_key in db.session.dirty
assert sample_template in db.session.dirty

141
tests/app/service/test_rest.py
View File

@@ -11,8 +11,8 @@ from app.dao.services_dao import dao_remove_user_from_service
from app.models import User, Organisation
from tests import create_authorization_header
from tests.app.conftest import (
sample_service as create_sample_service,
sample_service_permission as create_sample_service_permission,
sample_service as create_service,
sample_service_permission as create_service_permission,
sample_user as create_sample_user,
sample_notification as create_sample_notification,
sample_notification_with_job)
@@ -22,9 +22,9 @@ from app.models import KEY_TYPE_TEST
def test_get_service_list(notify_api, service_factory):
with notify_api.test_request_context():
with notify_api.test_client() as client:
service_factory.get('one', email_from='one')
service_factory.get('two', email_from='two')
service_factory.get('three', email_from='three')
service_factory.get('one')
service_factory.get('two')
service_factory.get('three')
auth_header = create_authorization_header()
response = client.get(
'/service',
@@ -38,50 +38,84 @@ def test_get_service_list(notify_api, service_factory):
assert json_resp['data'][2]['name'] == 'three'
def test_get_service_list_by_user(notify_api, sample_user, service_factory):
def test_get_service_list_with_only_active_flag(client, service_factory):
inactive = service_factory.get('one')
active = service_factory.get('two')
with notify_api.test_request_context():
with notify_api.test_client() as client:
service_factory.get('one', sample_user, email_from='one')
service_factory.get('two', sample_user, email_from='two')
service_factory.get('three', sample_user, email_from='three')
inactive.active = False
auth_header = create_authorization_header()
response = client.get(
'/service?user_id='.format(sample_user.id),
headers=[auth_header]
)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert len(json_resp['data']) == 3
assert json_resp['data'][0]['name'] == 'one'
assert json_resp['data'][1]['name'] == 'two'
assert json_resp['data'][2]['name'] == 'three'
auth_header = create_authorization_header()
response = client.get(
'/service?only_active=True',
headers=[auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 1
assert json_resp['data'][0]['id'] == str(active.id)
def test_get_service_list_by_user_should_return_empty_list_if_no_services(notify_api, service_factory, sample_user):
with notify_api.test_request_context():
with notify_api.test_client() as client:
new_user = User(
name='Test User',
email_address='new_user@digital.cabinet-office.gov.uk',
password='password',
mobile_number='+447700900986'
)
save_model_user(new_user)
def test_get_service_list_with_user_id_and_only_active_flag(
notify_db,
notify_db_session,
client,
sample_user,
service_factory
):
other_user = create_sample_user(notify_db, notify_db_session, email='foo@bar.gov.uk')
service_factory.get('one', sample_user, email_from='one')
service_factory.get('two', sample_user, email_from='two')
service_factory.get('three', sample_user, email_from='three')
inactive = service_factory.get('one', user=sample_user)
active = service_factory.get('two', user=sample_user)
from_other_user = service_factory.get('three', user=other_user)
auth_header = create_authorization_header()
response = client.get(
'/service?user_id={}'.format(new_user.id),
headers=[auth_header]
)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert len(json_resp['data']) == 0
inactive.active = False
auth_header = create_authorization_header()
response = client.get(
'/service?user_id={}&only_active=True'.format(sample_user.id),
headers=[auth_header]
)
assert response.status_code == 200
json_resp = json.loads(response.get_data(as_text=True))
assert len(json_resp['data']) == 1
assert json_resp['data'][0]['id'] == str(active.id)
def test_get_service_list_by_user(notify_db, notify_db_session, client, sample_user, service_factory):
other_user = create_sample_user(notify_db, notify_db_session, email='foo@bar.gov.uk')
service_factory.get('one', sample_user)
service_factory.get('two', sample_user)
service_factory.get('three', other_user)
auth_header = create_authorization_header()
response = client.get(
'/service?user_id={}'.format(sample_user.id),
headers=[auth_header]
)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert len(json_resp['data']) == 2
assert json_resp['data'][0]['name'] == 'one'
assert json_resp['data'][1]['name'] == 'two'
def test_get_service_list_by_user_should_return_empty_list_if_no_services(
notify_db,
notify_db_session,
client,
sample_service
):
# service is already created by sample user
new_user = create_sample_user(notify_db, notify_db_session, email='foo@bar.gov.uk')
auth_header = create_authorization_header()
response = client.get(
'/service?user_id={}'.format(new_user.id),
headers=[auth_header]
)
json_resp = json.loads(response.get_data(as_text=True))
assert response.status_code == 200
assert len(json_resp['data']) == 0
def test_get_service_list_should_return_empty_list_if_no_services(notify_api, notify_db, notify_db_session):
@@ -132,7 +166,7 @@ def test_get_service_by_id_should_404_if_no_service(notify_api, notify_db):
def test_get_service_by_id_and_user(notify_api, service_factory, sample_user):
with notify_api.test_request_context():
with notify_api.test_client() as client:
service = service_factory.get('new service', sample_user, email_from='new.service')
service = service_factory.get('new.service', sample_user)
auth_header = create_authorization_header()
resp = client.get(
'/service/{}?user_id={}'.format(service.id, sample_user.id),
@@ -285,7 +319,6 @@ def test_should_not_create_service_if_missing_data(notify_api, sample_user):
assert resp.status_code == 400
assert json_resp['result'] == 'error'
assert 'Missing data for required field.' in json_resp['message']['name']
assert 'Missing data for required field.' in json_resp['message']['active']
assert 'Missing data for required field.' in json_resp['message']['message_limit']
assert 'Missing data for required field.' in json_resp['message']['restricted']
@@ -446,7 +479,7 @@ def test_should_not_update_service_with_duplicate_name(notify_api,
with notify_api.test_request_context():
with notify_api.test_client() as client:
service_name = "another name"
service = create_sample_service(
service = create_service(
notify_db,
notify_db_session,
service_name=service_name,
@@ -479,7 +512,7 @@ def test_should_not_update_service_with_duplicate_email_from(notify_api,
with notify_api.test_client() as client:
email_from = "duplicate.name"
service_name = "duplicate name"
service = create_sample_service(
service = create_service(
notify_db,
notify_db_session,
service_name=service_name,
@@ -915,7 +948,7 @@ def test_remove_user_from_service(notify_api, notify_db, notify_db_session, samp
notify_db_session,
email="new@digital.cabinet-office.gov.uk")
# Simulates successfully adding a user to the service
second_permission = create_sample_service_permission(
second_permission = create_service_permission(
notify_db,
notify_db_session,
user=second_user)
@@ -1019,8 +1052,8 @@ def test_set_reply_to_email_for_service(notify_api, sample_service):
def test_get_all_notifications_for_service_in_order(notify_api, notify_db, notify_db_session):
with notify_api.test_request_context(), notify_api.test_client() as client:
service_1 = create_sample_service(notify_db, notify_db_session, service_name="1", email_from='1')
service_2 = create_sample_service(notify_db, notify_db_session, service_name="2", email_from='2')
service_1 = create_service(notify_db, notify_db_session, service_name="1", email_from='1')
service_2 = create_service(notify_db, notify_db_session, service_name="2", email_from='2')
create_sample_notification(notify_db, notify_db_session, service=service_2)
@@ -1240,8 +1273,8 @@ def test_get_services_with_detailed_flag(notify_api, notify_db, notify_db_sessio
def test_get_detailed_services_groups_by_service(notify_db, notify_db_session):
from app.service.rest import get_detailed_services
service_1 = create_sample_service(notify_db, notify_db_session, service_name="1", email_from='1')
service_2 = create_sample_service(notify_db, notify_db_session, service_name="2", email_from='2')
service_1 = create_service(notify_db, notify_db_session, service_name="1", email_from='1')
service_2 = create_service(notify_db, notify_db_session, service_name="2", email_from='2')
create_sample_notification(notify_db, notify_db_session, service=service_1, status='created')
create_sample_notification(notify_db, notify_db_session, service=service_2, status='created')
@@ -1267,8 +1300,8 @@ def test_get_detailed_services_groups_by_service(notify_db, notify_db_session):
def test_get_detailed_services_includes_services_with_no_notifications(notify_db, notify_db_session):
from app.service.rest import get_detailed_services
service_1 = create_sample_service(notify_db, notify_db_session, service_name="1", email_from='1')
service_2 = create_sample_service(notify_db, notify_db_session, service_name="2", email_from='2')
service_1 = create_service(notify_db, notify_db_session, service_name="1", email_from='1')
service_2 = create_service(notify_db, notify_db_session, service_name="2", email_from='2')
create_sample_notification(notify_db, notify_db_session, service=service_1)

1
tests/conftest.py
View File

@@ -47,6 +47,7 @@ def client(notify_api):
@pytest.fixture(scope='session')
def notify_db(notify_api):
assert db.engine.url.database != 'notification_api', 'dont run tests against main db'
Migrate(notify_api, db)
Manager(db, MigrateCommand)
BASE_DIR = os.path.dirname(os.path.dirname(__file__))