Merge remote-tracking branch 'origin/master' into remove_alpha_client_from_api

2026-02-01 23:55:58 -05:00 · 2016-02-09 15:54:21 +00:00
parent 2fda7ee59b 3fd333ef94
commit bd947d0743
10 changed files with 419 additions and 5 deletions
--- a/tests/app/conftest.py
+++ b/tests/app/conftest.py
@@ -1,12 +1,13 @@
 import pytest
 from flask import jsonify

-from app.models import (User, Service, Template, ApiKey, Job, VerifyCode)
+from app.models import (User, Service, Template, ApiKey, Job, VerifyCode, Notification)
 from app.dao.users_dao import (save_model_user, create_user_code, create_secret_code)
 from app.dao.services_dao import save_model_service
 from app.dao.templates_dao import save_model_template
 from app.dao.api_key_dao import save_model_api_key
 from app.dao.jobs_dao import save_job
+from app.dao.notifications_dao import save_notification
 import uuid


@@ -162,3 +163,31 @@ def mock_secret_code(mocker):

    mock_class = mocker.patch('app.dao.users_dao.create_secret_code', side_effect=_create)
    return mock_class
+
+
+@pytest.fixture(scope='function')
+def sample_notification(notify_db,
+                        notify_db_session,
+                        service=None,
+                        template=None,
+                        job=None):
+    if service is None:
+        service = sample_service(notify_db, notify_db_session)
+    if template is None:
+        template = sample_template(notify_db, notify_db_session, service=service)
+    if job is None:
+        job = sample_job(notify_db, notify_db_session, service=service, template=template)
+
+    notificaton_id = uuid.uuid4()
+    to = '+44709123456'
+
+    data = {
+        'id': notificaton_id,
+        'to': to,
+        'job': job,
+        'service': service,
+        'template': template
+    }
+    notification = Notification(**data)
+    save_notification(notification)
+    return notification
--- a/tests/app/dao/test_notification_dao.py
+++ b/tests/app/dao/test_notification_dao.py
@@ -0,0 +1,75 @@
+import uuid
+
+from app.models import Notification
+
+from app.dao.notifications_dao import (
+    save_notification,
+    get_notification,
+    get_notifications
+)
+
+
+def test_save_notification(notify_db, notify_db_session, sample_template, sample_job):
+
+    assert Notification.query.count() == 0
+
+    notification_id = uuid.uuid4()
+    to = '+44709123456'
+    data = {
+        'id': notification_id,
+        'to': to,
+        'job': sample_job,
+        'service': sample_template.service,
+        'template': sample_template
+    }
+
+    notification = Notification(**data)
+    save_notification(notification)
+
+    assert Notification.query.count() == 1
+
+    notification_from_db = Notification.query.get(notification_id)
+
+    assert data['id'] == notification_from_db.id
+    assert data['to'] == notification_from_db.to
+    assert data['job'] == notification_from_db.job
+    assert data['service'] == notification_from_db.service
+    assert data['template'] == notification_from_db.template
+    assert 'sent' == notification_from_db.status
+
+
+def test_get_notification_for_job(notify_db, notify_db_session, sample_notification):
+    notifcation_from_db = get_notification(sample_notification.service.id,
+                                           sample_notification.job_id,
+                                           sample_notification.id)
+    assert sample_notification == notifcation_from_db
+
+
+def test_get_all_notifications_for_job(notify_db, notify_db_session, sample_job):
+
+    from tests.app.conftest import sample_notification
+    for i in range(0, 5):
+        sample_notification(notify_db,
+                            notify_db_session,
+                            service=sample_job.service,
+                            template=sample_job.template,
+                            job=sample_job)
+
+    notifcations_from_db = get_notifications(sample_job.service.id, sample_job.id)
+    assert len(notifcations_from_db) == 5
+
+
+def test_update_notification(notify_db, notify_db_session, sample_notification):
+    assert sample_notification.status == 'sent'
+
+    update_dict = {
+        'id': str(sample_notification.id),
+        'service': str(sample_notification.service.id),
+        'template': sample_notification.template.id,
+        'job': str(sample_notification.job.id),
+        'status': 'failed'
+    }
+
+    save_notification(sample_notification, update_dict=update_dict)
+    notification_from_db = Notification.query.get(sample_notification.id)
+    assert notification_from_db.status == 'failed'
--- a/tests/app/job/test_job_rest.py
+++ b/tests/app/job/test_job_rest.py
@@ -163,6 +163,130 @@ def test_get_update_job_status(notify_api,
            assert resp_json['data']['status'] == 'in progress'


+def test_get_notification(notify_api, notify_db, notify_db_session, sample_notification):
+
+    with notify_api.test_request_context():
+        with notify_api.test_client() as client:
+            path = url_for('job.get_notification_for_job',
+                           service_id=sample_notification.service.id,
+                           job_id=sample_notification.job.id,
+                           notification_id=sample_notification.id)
+
+            auth_header = create_authorization_header(service_id=sample_notification.service.id,
+                                                      path=path,
+                                                      method='GET')
+
+            headers = [('Content-Type', 'application/json'), auth_header]
+            response = client.get(path, headers=headers)
+            resp_json = json.loads(response.get_data(as_text=True))
+
+            assert str(sample_notification.id) == resp_json['data']['id']
+            assert str(sample_notification.service.id) == resp_json['data']['service']
+            assert sample_notification.template.id == resp_json['data']['template']
+            assert str(sample_notification.job.id) == resp_json['data']['job']
+            assert sample_notification.status == resp_json['data']['status']
+
+
+def test_get_notifications(notify_api, notify_db, notify_db_session, sample_job):
+
+    from tests.app.conftest import sample_notification
+    for i in range(0, 5):
+        sample_notification(notify_db,
+                            notify_db_session,
+                            service=sample_job.service,
+                            template=sample_job.template,
+                            job=sample_job)
+
+    service_id = str(sample_job.service.id)
+    job_id = str(sample_job.id)
+
+    with notify_api.test_request_context():
+        with notify_api.test_client() as client:
+            path = url_for('job.get_notification_for_job',
+                           service_id=service_id,
+                           job_id=job_id)
+
+            auth_header = create_authorization_header(service_id=service_id,
+                                                      path=path,
+                                                      method='GET')
+
+            headers = [('Content-Type', 'application/json'), auth_header]
+            response = client.get(path, headers=headers)
+            resp_json = json.loads(response.get_data(as_text=True))
+
+            assert len(resp_json['data']) == 5
+
+
+def test_add_notification(notify_api, notify_db, notify_db_session, sample_job):
+
+    notificaton_id = uuid.uuid4()
+    to = '+44709123456'
+    data = {
+        'id': str(notificaton_id),
+        'to': to,
+        'job': str(sample_job.id),
+        'service': str(sample_job.service.id),
+        'template': sample_job.template.id
+    }
+    with notify_api.test_request_context():
+        with notify_api.test_client() as client:
+            path = url_for('job.create_notification_for_job',
+                           service_id=sample_job.service.id,
+                           job_id=sample_job.id)
+
+            auth_header = create_authorization_header(service_id=sample_job.service.id,
+                                                      path=path,
+                                                      method='POST',
+                                                      request_body=json.dumps(data))
+
+            headers = [('Content-Type', 'application/json'), auth_header]
+
+            response = client.post(path, headers=headers, data=json.dumps(data))
+
+            resp_json = json.loads(response.get_data(as_text=True))
+
+            assert data['id'] == resp_json['data']['id']
+            assert data['to'] == resp_json['data']['to']
+            assert data['service'] == resp_json['data']['service']
+            assert data['template'] == resp_json['data']['template']
+            assert data['job'] == resp_json['data']['job']
+            assert 'sent' == resp_json['data']['status']
+
+
+def test_update_notification(notify_api, notify_db, notify_db_session, sample_notification):
+
+    assert sample_notification.status == 'sent'
+
+    update_data = {
+        'id': str(sample_notification.id),
+        'to': sample_notification.to,
+        'job': str(sample_notification.job.id),
+        'service': str(sample_notification.service.id),
+        'template': sample_notification.template.id,
+        'status': 'failed'
+    }
+    with notify_api.test_request_context():
+        with notify_api.test_client() as client:
+            path = url_for('job.update_notification_for_job',
+                           service_id=sample_notification.service.id,
+                           job_id=sample_notification.job.id,
+                           notification_id=sample_notification.id)
+
+            auth_header = create_authorization_header(service_id=sample_notification.service.id,
+                                                      path=path,
+                                                      method='PUT',
+                                                      request_body=json.dumps(update_data))
+
+            headers = [('Content-Type', 'application/json'), auth_header]
+
+            response = client.put(path, headers=headers, data=json.dumps(update_data))
+
+            resp_json = json.loads(response.get_data(as_text=True))
+
+            assert update_data['id'] == resp_json['data']['id']
+            assert 'failed' == resp_json['data']['status']
+
+
 def _setup_jobs(notify_db, notify_db_session, template, number_of_jobs=5):
    for i in range(number_of_jobs):
        create_job(notify_db, notify_db_session, service=template.service,