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

Fix for job urls that should have been within context of a service.

Browse Source 
The service id path parameter is captured in the url prefix for the
job blueprint.
This commit is contained in:
Adam Shimali
2016-01-18 09:57:04 +00:00
parent 56f455fe3a
commit f4a0a9b8d9
5 changed files with 63 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/__init__.py
View File

@@ -36,7 +36,7 @@ def create_app(config_name):
application.register_blueprint(user_blueprint, url_prefix='/user')
application.register_blueprint(template_blueprint, url_prefix="/template")
application.register_blueprint(status_blueprint, url_prefix='/status')
application.register_blueprint(job_blueprint, url_prefix='/job')
application.register_blueprint(job_blueprint)
return application

6
app/dao/jobs_dao.py
View File

@@ -7,13 +7,13 @@ def save_job(job):
db.session.commit()
def get_job_by_id(job_id):
return Job.query.filter_by(id=job_id).one()
def get_job(service_id, job_id):
return Job.query.filter_by(service_id=service_id, id=job_id).one()
def get_jobs_by_service(service_id):
return Job.query.filter_by(service_id=service_id).all()
def get_jobs():
def _get_jobs():
return Job.query.all()

14
app/job/rest.py
View File

@@ -9,8 +9,8 @@ from sqlalchemy.orm.exc import NoResultFound
from app.dao.jobs_dao import (
save_job,
get_job_by_id,
get_jobs
get_job,
get_jobs_by_service
)
from app.schemas import (
@@ -18,15 +18,15 @@ from app.schemas import (
jobs_schema
)
job = Blueprint('job', __name__)
job = Blueprint('job', __name__, url_prefix='/service/<int:service_id>/job')
@job.route('/<job_id>', methods=['GET'])
@job.route('', methods=['GET'])
def get_job(job_id=None):
def get_job_for_service(service_id, job_id=None):
if job_id:
try:
job = get_job_by_id(job_id)
job = get_job(service_id, job_id)
data, errors = job_schema.dump(job)
return jsonify(data=data)
except DataError:
@@ -34,13 +34,13 @@ def get_job(job_id=None):
except NoResultFound:
return jsonify(result="error", message="Job not found"), 404
else:
jobs = get_jobs()
jobs = get_jobs_by_service(service_id)
data, errors = jobs_schema.dump(jobs)
return jsonify(data=data)
@job.route('', methods=['POST'])
def create_job():
def create_job(service_id):
job, errors = job_schema.load(request.get_json())
if errors:
return jsonify(result="error", message=errors), 400

40
tests/app/dao/test_jobs_dao.py
View File

@@ -2,9 +2,9 @@ import uuid
from app.dao.jobs_dao import (
save_job,
get_job_by_id,
get_job,
get_jobs_by_service,
get_jobs
_get_jobs
)
from app.models import Job
@@ -35,17 +35,39 @@ def test_save_job(notify_db, notify_db_session, sample_template):
def test_get_job_by_id(notify_db, notify_db_session, sample_job):
job_from_db = get_job_by_id(sample_job.id)
job_from_db = get_job(sample_job.service.id, sample_job.id)
assert sample_job == job_from_db
def test_get_jobs_for_service(notify_db, notify_db_session, sample_job):
def test_get_jobs_for_service(notify_db, notify_db_session, sample_template):
service_id = sample_job.service_id
job_from_db = get_jobs_by_service(service_id)
from tests.app.conftest import sample_job as create_job
from tests.app.conftest import sample_service as create_service
from tests.app.conftest import sample_template as create_template
from tests.app.conftest import sample_user as create_user
assert len(job_from_db) == 1
assert sample_job == job_from_db[0]
one_job = create_job(notify_db, notify_db_session, sample_template.service,
sample_template)
other_user = create_user(notify_db, notify_db_session,
email="test@digital.cabinet-office.gov.uk")
other_service = create_service(notify_db, notify_db_session,
user=other_user)
other_template = create_template(notify_db, notify_db_session,
service=other_service)
other_job = create_job(notify_db, notify_db_session, service=other_service,
template=other_template)
one_job_from_db = get_jobs_by_service(one_job.service_id)
other_job_from_db = get_jobs_by_service(other_job.service_id)
assert len(one_job_from_db) == 1
assert one_job == one_job_from_db[0]
assert len(other_job_from_db) == 1
assert other_job == other_job_from_db[0]
assert one_job_from_db != other_job_from_db
def test_get_all_jobs(notify_db, notify_db_session, sample_template):
@@ -55,5 +77,5 @@ def test_get_all_jobs(notify_db, notify_db_session, sample_template):
notify_db_session,
sample_template.service,
sample_template)
jobs_from_db = get_jobs()
jobs_from_db = _get_jobs()
assert len(jobs_from_db) == 5

32
tests/app/job/test_job_rest.py
View File

@@ -9,12 +9,15 @@ from tests.app.conftest import sample_job as create_job
def test_get_jobs(notify_api, notify_db, notify_db_session, sample_template):
_setup_jobs(notify_db, notify_db_session, sample_template)
service_id = sample_template.service.id
with notify_api.test_request_context():
with notify_api.test_client() as client:
auth_header = create_authorization_header(service_id=sample_template.service.id,
path=url_for('job.get_job'),
path = url_for('job.get_job_for_service', service_id=service_id)
auth_header = create_authorization_header(service_id=service_id,
path=path,
method='GET')
response = client.get(url_for('job.get_job'), headers=[auth_header])
response = client.get(path, headers=[auth_header])
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
assert len(resp_json['data']) == 5
@@ -23,12 +26,14 @@ def test_get_jobs(notify_api, notify_db, notify_db_session, sample_template):
def test_get_job_with_invalid_id_returns400(notify_api, notify_db,
notify_db_session,
sample_template):
service_id = sample_template.service.id
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = url_for('job.get_job_for_service', job_id='invalid_id', service_id=service_id)
auth_header = create_authorization_header(service_id=sample_template.service.id,
path=url_for('job.get_job', job_id='invalid_id'),
path=path,
method='GET')
response = client.get(url_for('job.get_job', job_id='invalid_id'), headers=[auth_header])
response = client.get(path, headers=[auth_header])
assert response.status_code == 400
resp_json = json.loads(response.get_data(as_text=True))
assert resp_json == {'message': 'Invalid job id',
@@ -39,12 +44,14 @@ def test_get_job_with_unknown_id_returns404(notify_api, notify_db,
notify_db_session,
sample_template):
random_id = str(uuid.uuid4())
service_id = sample_template.service.id
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = url_for('job.get_job_for_service', job_id=random_id, service_id=service_id)
auth_header = create_authorization_header(service_id=sample_template.service.id,
path=url_for('job.get_job', job_id=random_id),
path=path,
method='GET')
response = client.get(url_for('job.get_job', job_id=random_id), headers=[auth_header])
response = client.get(path, headers=[auth_header])
assert response.status_code == 404
resp_json = json.loads(response.get_data(as_text=True))
assert resp_json == {'message': 'Job not found', 'result': 'error'}
@@ -53,12 +60,14 @@ def test_get_job_with_unknown_id_returns404(notify_api, notify_db,
def test_get_job_by_id(notify_api, notify_db, notify_db_session,
sample_job):
job_id = str(sample_job.id)
service_id = sample_job.service.id
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = url_for('job.get_job_for_service', job_id=job_id, service_id=service_id)
auth_header = create_authorization_header(service_id=sample_job.service.id,
path=url_for('job.get_job', job_id=job_id),
path=path,
method='GET')
response = client.get(url_for('job.get_job', job_id=job_id), headers=[auth_header])
response = client.get(path, headers=[auth_header])
assert response.status_code == 200
resp_json = json.loads(response.get_data(as_text=True))
assert resp_json['data']['id'] == job_id
@@ -81,13 +90,14 @@ def test_post_job(notify_api, notify_db, notify_db_session, sample_template):
}
with notify_api.test_request_context():
with notify_api.test_client() as client:
path = url_for('job.create_job', service_id=service_id)
auth_header = create_authorization_header(service_id=sample_template.service.id,
path=url_for('job.create_job'),
path=path,
method='POST',
request_body=json.dumps(data))
headers = [('Content-Type', 'application/json'), auth_header]
response = client.post(
url_for('job.create_job'),
path,
data=json.dumps(data),
headers=headers)
assert response.status_code == 201