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

code review feedback

Browse Source
This commit is contained in:
Kenneth Kehl
2025-06-25 08:54:44 -07:00
parent 77b3c1afa6
commit 5b2d8e0648
2 changed files with 16 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
app/job/rest.py
View File

@@ -68,8 +68,7 @@ def check_suspicious_id(*args):
@job_blueprint.route("/<job_id>", methods=["GET"])
def get_job_by_service_and_job_id(service_id, job_id):
check_suspicious_id(service_id)
check_suspicious_id(job_id)
check_suspicious_id(service_id, job_id)
job = dao_get_job_by_service_id_and_job_id(service_id, job_id)
statistics = dao_get_notification_outcomes_for_job(service_id, job_id)
data = JobSchema(session=db.session).dump(job)
@@ -83,8 +82,7 @@ def get_job_by_service_and_job_id(service_id, job_id):
@job_blueprint.route("/<job_id>/cancel", methods=["POST"])
def cancel_job(service_id, job_id):
check_suspicious_id(service_id)
check_suspicious_id(job_id)
check_suspicious_id(service_id, job_id)
job = dao_get_future_scheduled_job_by_id_and_service_id(job_id, service_id)
job.job_status = JobStatus.CANCELLED
@@ -95,8 +93,7 @@ def cancel_job(service_id, job_id):
@job_blueprint.route("/<job_id>/notifications", methods=["GET"])
def get_all_notifications_for_service_job(service_id, job_id):
check_suspicious_id(service_id)
check_suspicious_id(job_id)
check_suspicious_id(service_id, job_id)
data = notifications_filter_schema.load(request.args)
page = data["page"] if "page" in data else 1
@@ -159,8 +156,7 @@ def get_all_notifications_for_service_job(service_id, job_id):
@job_blueprint.route("/<job_id>/recent_notifications", methods=["GET"])
def get_recent_notifications_for_service_job(service_id, job_id):
check_suspicious_id(service_id)
check_suspicious_id(job_id)
check_suspicious_id(service_id, job_id)
data = notifications_filter_schema.load(request.args)
page = data["page"] if "page" in data else 1
@@ -227,8 +223,7 @@ def get_recent_notifications_for_service_job(service_id, job_id):
@job_blueprint.route("/<job_id>/notification_count", methods=["GET"])
def get_notification_count_for_job_id(service_id, job_id):
check_suspicious_id(service_id)
check_suspicious_id(job_id)
check_suspicious_id(service_id, job_id)
dao_get_job_by_service_id_and_job_id(service_id, job_id)
count = dao_get_notification_count_for_job_id(job_id=job_id)

12
tests/app/job/test_rest.py
View File

@@ -5,6 +5,7 @@ from unittest.mock import ANY
from zoneinfo import ZoneInfo
import pytest
import werkzeug
from freezegun import freeze_time
import app.celery.tasks
@@ -16,7 +17,7 @@ from app.enums import (
NotificationType,
TemplateType,
)
from app.job.rest import is_suspicious_input, is_valid_id
from app.job.rest import check_suspicious_id, is_suspicious_input, is_valid_id
from app.utils import utc_now
from tests import create_admin_authorization_header
from tests.app.db import (
@@ -595,6 +596,15 @@ def test_is_valid_id(sample_job):
assert returnVal is False
def test_check_suspicious_id(sample_job):
# This should be good
check_suspicious_id(sample_job.id, sample_job.service_id)
# This should be bad
with pytest.raises(werkzeug.exceptions.Forbidden):
check_suspicious_id(sample_job.id, "what is this???")
def test_is_suspicious_input(sample_job):
returnVal = is_suspicious_input(sample_job.id)
assert returnVal is False