Review request page #159038263
This commit is contained in:
dandds 2018-09-07 16:07:52 -04:00 committed by GitHub
commit afeb6729ee
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 303 additions and 253 deletions

View File

@ -25,6 +25,16 @@ class Authorization(object):
return False
@classmethod
def check_can_approve_request(cls, user):
if (
Permissions.REVIEW_AND_APPROVE_JEDI_WORKSPACE_REQUEST
in user.atat_permissions
):
return True
else:
raise UnauthorizedError(user, "cannot review and approve requests")
@classmethod
def check_workspace_permission(cls, user, workspace, permission, message):
if not Authorization.has_workspace_permission(user, workspace, permission):

View File

@ -30,6 +30,8 @@ def readableInteger(value):
def getOptionLabel(value, options):
if hasattr(value, "value"):
value = value.value
try:
return next(tup[1] for tup in options if tup[0] == value)
except StopIteration:
@ -50,6 +52,19 @@ def mixedContentToJson(value):
return app.jinja_env.filters["tojson"](value)
def findFilter(value, filter_name, filter_args=[]):
if not filter_name:
return value
elif filter_name in app.jinja_env.filters:
return app.jinja_env.filters[filter_name](value, *filter_args)
else:
raise ValueError("filter name {} not found".format(filter_name))
def renderList(value):
return app.jinja_env.filters["safe"]("<br>".join(value))
def register_filters(app):
app.jinja_env.filters["iconSvg"] = iconSvg
app.jinja_env.filters["dollars"] = dollars
@ -57,3 +72,5 @@ def register_filters(app):
app.jinja_env.filters["readableInteger"] = readableInteger
app.jinja_env.filters["getOptionLabel"] = getOptionLabel
app.jinja_env.filters["mixedContentToJson"] = mixedContentToJson
app.jinja_env.filters["findFilter"] = findFilter
app.jinja_env.filters["renderList"] = renderList

View File

@ -140,3 +140,13 @@ WORKSPACE_ROLES = [
},
),
]
FUNDING_TYPES = [
("", "- Select -"),
("RDTE", "Research, Development, Testing & Evaluation (RDT&E)"),
("OM", "Operations & Maintenance (O&M)"),
("PROC", "Procurement (PROC)"),
("OTHER", "Other"),
]
TASK_ORDER_SOURCES = [("MANUAL", "Manual"), ("EDA", "EDA")]

View File

@ -10,6 +10,7 @@ from atst.domain.task_orders import TaskOrders
from .fields import NewlineListField, SelectField
from .forms import ValidatedForm
from .data import FUNDING_TYPES
PE_REGEX = re.compile(
@ -161,13 +162,7 @@ class ExtendedFinancialForm(BaseFinancialForm):
funding_type = SelectField(
description="What is the source of funding?",
choices=[
("", "- Select -"),
("RDTE", "Research, Development, Testing & Evaluation (RDT&E)"),
("OM", "Operations & Maintenance (O&M)"),
("PROC", "Procurement (PROC)"),
("OTHER", "Other"),
],
choices=FUNDING_TYPES,
validators=[Required()],
render_kw={"required": False},
)

View File

@ -8,7 +8,7 @@ from atst.models import Base
class Source(Enum):
MANUAL = "Manual"
EDA = "eda"
EDA = "EDA"
class FundingType(Enum):

View File

@ -1,9 +1,52 @@
from flask import render_template
from flask import render_template, g, Response
from flask import current_app as app
from . import requests_bp
from atst.forms.data import SERVICE_BRANCHES
from atst.domain.requests import Requests
from atst.domain.exceptions import NotFoundError
from atst.domain.authz import Authorization
@requests_bp.route("/request_approval", methods=["GET"])
def requests_approval():
return render_template("request_approval.html", service_branches=SERVICE_BRANCHES)
def task_order_dictionary(task_order):
return {
c.name: getattr(task_order, c.name)
for c in task_order.__table__.columns
if c.name not in ["id", "attachment_id"]
}
@requests_bp.route("/requests/approval/<string:request_id>", methods=["GET"])
def approval(request_id):
request = Requests.get(g.current_user, request_id)
Authorization.check_can_approve_request(g.current_user)
data = request.body
if request.task_order:
data["task_order"] = task_order_dictionary(request.task_order)
return render_template(
"requests/approval.html",
data=data,
request_id=request_id,
status=request.status.value,
financial_review=True,
pdf_available=request.task_order and request.task_order.pdf,
)
@requests_bp.route("/requests/task_order_download/<string:request_id>", methods=["GET"])
def task_order_pdf_download(request_id):
request = Requests.get(g.current_user, request_id)
if request.task_order and request.task_order.pdf:
pdf = request.task_order.pdf
generator = app.uploader.download_stream(pdf.object_name)
return Response(
generator,
headers={
"Content-Disposition": "attachment; filename={}".format(pdf.filename)
},
mimetype="application/pdf",
)
else:
raise NotFoundError("task_order pdf")

View File

@ -70,7 +70,9 @@ class RequestsIndex(object):
else "-"
)
if Requests.is_pending_financial_verification(request):
if viewing_role == "ccpo":
edit_link = url_for("requests.approval", request_id=request.id)
elif Requests.is_pending_financial_verification(request):
edit_link = url_for(
"requests.financial_verification", request_id=request.id
)

View File

@ -9,6 +9,8 @@ from atst.forms.data import (
ASSISTANCE_ORG_TYPES,
DATA_TRANSFER_AMOUNTS,
COMPLETION_DATE_RANGES,
FUNDING_TYPES,
TASK_ORDER_SOURCES,
)
@ -19,6 +21,8 @@ def option_data():
"assistance_org_types": ASSISTANCE_ORG_TYPES,
"data_transfer_amounts": DATA_TRANSFER_AMOUNTS,
"completion_date_ranges": COMPLETION_DATE_RANGES,
"funding_types": FUNDING_TYPES,
"task_order_sources": TASK_ORDER_SOURCES,
}

View File

@ -37,8 +37,11 @@ class Uploader:
)
return (fyle.filename, object_name)
def download(self, path):
pass
def download_stream(self, object_name):
obj = self.container.get_object(object_name=object_name)
with NamedTemporaryFile() as tempfile:
obj.download(tempfile.name, overwrite_existing=True)
return open(tempfile.name, "rb")
def _get_container(self, provider, container, key, secret):
if provider == "LOCAL":

View File

@ -2,6 +2,20 @@
<span class='label label--error'>Response Required</span>
{%- endmacro %}
{% macro DefinitionReviewField(title, section, item_name, filter=None, filter_args=[]) -%}
<div>
<dt>{{ title | safe }}</dt>
<dd>
{% if data[section] and data[section][item_name] %}
{{ data[section][item_name] | findFilter(filter, filter_args) }}
{% else %}
{{ RequiredLabel() }}
{% endif %}
</dd>
</div>
{% endmacro %}
<hr>
<h2>
Details of Use
@ -14,161 +28,51 @@
</h2>
<dl>
<div>
<dt>DoD Component</dt>
<dd>
{% if data['details_of_use']['dod_component'] %}
{{ data['details_of_use']['dod_component'] | getOptionLabel(service_branches) }}
{% else %}
{{ RequiredLabel() }}
{% endif %}
</dd>
</div>
<div>
<dt>JEDI Usage</dt>
<dd>{{ data['details_of_use']['jedi_usage'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("DoD Component", "details_of_use", "dod_component", filter="getOptionLabel", filter_args=[service_branches]) }}
<div>
<dt>Number of software systems</dt>
<dd>
{% if data['details_of_use']['num_software_systems'] %}
{{ data['details_of_use']['num_software_systems'] | readableInteger }}
{% else %}
{{ RequiredLabel() }}
{% endif %}
</dd>
</div>
{{ DefinitionReviewField("JEDI Usage", "details_of_use", "jedi_usage") }}
<div>
<dt>JEDI Migration</dt>
<dd>{{ data['details_of_use']['jedi_migration'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("Number of software systems", "details_of_use", "num_software_systems", filter="readableInteger") }}
{{ DefinitionReviewField("JEDI Migration", "details_of_use", "jedi_migration") }}
{% if data['details_of_use']['jedi_migration'] == 'yes' %}
<div>
<dt>Rationalization of Software Systems</dt>
<dd>{{ data['details_of_use']['rationalization_software_systems'] or RequiredLabel() }}</dd>
</div>
<div>
<dt>Technical Support Team</dt>
<dd>{{ data['details_of_use']['technical_support_team'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("Rationalization of Software Systems", "details_of_use", "rationalization_software_systems") }}
{{ DefinitionReviewField("Technical Support Team", "details_of_use", "technical_support_team") }}
{% if data['details_of_use']['technical_support_team'] == 'yes' %}
<div>
<dt>Organization Providing Assistance</dt>
<dd>
{% if data['details_of_use']['organization_providing_assistance'] %}
{{ data['details_of_use']['organization_providing_assistance'] | getOptionLabel(assistance_org_types) }}
{% else %}
{{ RequiredLabel() }}
{% endif %}
</dd>
</div>
{{ DefinitionReviewField("Organization Providing Assistance", "details_of_use", "organization_providing_assistance", filter="getOptionLabel", filter_args=[assistance_org_types]) }}
{% endif %}
<div>
<dt>Engineering Assessment</dt>
<dd>{{ data['details_of_use']['engineering_assessment'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("Engineering Assessment", "details_of_use", "engineering_assessment") }}
<div>
<dt>Data Transfers</dt>
<dd>
{% if data['details_of_use']['data_transfers'] %}
{{ data['details_of_use']['data_transfers'] | getOptionLabel(data_transfer_amounts) }}
{% else %}
{{ RequiredLabel() }}
{% endif %}
</dd>
</div>
{{ DefinitionReviewField("Data Transfers", "details_of_use", "data_transfers", filter="getOptionLabel", filter_args=[data_transfer_amounts]) }}
<div>
<dt>Expected Completion Date</dt>
<dd>
{% if data['details_of_use']['expected_completion_date'] %}
{{ data['details_of_use']['expected_completion_date'] | getOptionLabel(completion_date_ranges) }}
{% else %}
{{ RequiredLabel() }}
{% endif %}
</dd>
</div>
{{ DefinitionReviewField("Expected Completion Date", "details_of_use", "expected_completion_date", filter="getOptionLabel", filter_args=[completion_date_ranges]) }}
{% else %}
<div>
<dt>Cloud Native</dt>
<dd>{{ data['details_of_use']['cloud_native'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("Cloud Native", "details_of_use", "cloud_native") }}
{% endif %}
<div>
<dt>Estimated Monthly Spend</dt>
<dd>
{% if data['details_of_use']['estimated_monthly_spend'] %}
{{ data['details_of_use']['estimated_monthly_spend'] | dollars }}
{% else %}
{{ RequiredLabel() }}
{% endif %}
</dd>
</div>
{{ DefinitionReviewField("Estimated Monthly Spend", "details_of_use", "estimated_monthly_spend", filter="dollars") }}
{% if jedi_request and jedi_request.annual_spend > annual_spend_threshold %}
<div>
<dt>Number of User Sessions</dt>
<dd>
{% if data['details_of_use']['number_user_sessions'] %}
{{ data['details_of_use']['number_user_sessions'] | readableInteger }}
{% else %}
{{ RequiredLabel() }}
{% endif %}
</dd>
</div>
{{ DefinitionReviewField("Number of User Sessions", "details_of_use", "number_user_sessions", filter="readableInteger") }}
<div>
<dt>Average Daily Traffic (Number of Requests)</dt>
<dd>
{% if data['details_of_use']['average_daily_traffic'] %}
{{ data['details_of_use']['average_daily_traffic'] | readableInteger }}
{% else %}
{{ RequiredLabel() }}
{% endif %}
</dd>
</div>
{{ DefinitionReviewField("Average Daily Traffic (Number of Requests)", "details_of_use", "average_daily_traffic", filter="readableInteger") }}
{{ DefinitionReviewField("Average Daily Traffic (GB)", "details_of_use", "average_daily_traffic_gb", filter="readableInteger") }}
<div>
<dt>Average Daily Traffic (GB)</dt>
<dd>
{% if data['details_of_use']['average_daily_traffic_gb'] %}
{{ data['details_of_use']['average_daily_traffic_gb'] | readableInteger }} GB
{% else %}
{{ RequiredLabel() }}
{% endif %}
</dd>
</div>
{% endif %}
<div>
<dt>Total Spend</dt>
<dd>
{% if data['details_of_use']['dollar_value'] %}
{{ data['details_of_use']['dollar_value'] | dollars }}
{% else %}
{{ RequiredLabel() }}
{% endif %}
</dd>
</div>
{{ DefinitionReviewField("Total Spend", "details_of_use", "dollar_value", filter="dollars") }}
<div>
<dt>Start Date</dt>
<dd>{{ data['details_of_use']['start_date'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("Start Date", "details_of_use", "start_date") }}
</dl>
<hr>
@ -183,63 +87,21 @@
</h2>
<dl>
<div>
<dt>First Name</dt>
<dd>{{ data['information_about_you']['fname_request'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("First Name", "information_about_you", "fname_request") }}
<div>
<dt>Last Name</dt>
<dd>{{ data['information_about_you']['lname_request'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("Last Name", "information_about_you", "lname_request") }}
<div>
<dt>Email Address</dt>
<dd>{{ data['information_about_you']['email_request'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("Email Address", "information_about_you", "email_request") }}
<div>
<dt>Phone Number</dt>
<dd>
{% if data['information_about_you']['phone_number'] %}
{{ data['information_about_you']['phone_number'] | usPhone }}
{% else %}
{{ RequiredLabel() }}
{% endif %}
</dd>
</div>
{{ DefinitionReviewField("Phone Number", "information_about_you", "phone_number", filter="usPhone") }}
<div>
<dt>Service Branch or Agency</dt>
<dd>
{% if data['information_about_you']['service_branch'] %}
{{ data['information_about_you']['service_branch'] | getOptionLabel(service_branches) }}
{% else %}
{{ RequiredLabel() }}
{% endif %}
</dd>
</div>
{{ DefinitionReviewField("Service Branch or Agency", "information_about_you", "service_branch", filter="getOptionLabel", filter_args=[service_branches]) }}
<div>
<dt>Citizenship</dt>
<dd>{{ data['information_about_you']['citizenship'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("Citizenship", "information_about_you", "citizenship") }}
<div>
<dt>Designation of Person</dt>
<dd>
{% if data['information_about_you']['designation'] %}
{{ data['information_about_you']['designation'] | capitalize }}
{% else %}
{{ RequiredLabel() }}
{% endif %}
</dd>
</div>
{{ DefinitionReviewField("Designation of Person", "information_about_you", "designation", filter="capitalize") }}
<div>
<dt>Latest Information Assurance (IA) Training completion date</dt>
<dd>{{ data['information_about_you']['date_latest_training'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("Latest Information Assurance (IA) Training completion date", "information_about_you", "date_latest_training") }}
</dl>
<hr>
@ -254,25 +116,74 @@
</h2>
<dl>
<div>
<dt>POC First Name</dt>
<dd>{{ data['primary_poc']['fname_poc'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("POC First Name", "primary_poc", "fname_poc") }}
<div>
<dt>POC Last Name</dt>
<dd>{{ data['primary_poc']['lname_poc'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("POC Last Name", "primary_poc", "lname_poc") }}
<div>
<dt>POC Email Address</dt>
<dd>{{ data['primary_poc']['email_poc'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("POC Email Address", "primary_poc", "email_poc") }}
<div>
<dt>DOD ID</dt>
<dd>{{ data['primary_poc']['dodid_poc'] or RequiredLabel() }}</dd>
</div>
{{ DefinitionReviewField("DOD ID", "primary_poc", "dodid_poc") }}
</dl>
{% if financial_review %}
<hr>
<h2>
Financial Verification
</h2>
{% if pdf_available %}
<div>
<a href="{{ url_for("requests.task_order_pdf_download", request_id=request_id)}}" download>
Download the Task Order PDF
</a>
</div>
{% endif %}
<dl>
{{ DefinitionReviewField("Task Order Information Source", "task_order", "source", filter="getOptionLabel", filter_args=[task_order_sources]) }}
{{ DefinitionReviewField("Task Order Number", "task_order", "number") }}
{{ DefinitionReviewField("What is the source of funding?", "task_order", "funding_type", filter="getOptionLabel", filter_args=[funding_types]) }}
{% if data["task_order"] and data["task_order"]["funding_type"].value == "OTHER" %}
{{ DefinitionReviewField("If other, please specify", "task_order", "funding_type_other") }}
{% endif %}
{{ DefinitionReviewField("<dl><dt>CLIN 0001</dt> - <dd>Unclassified IaaS and PaaS Amount</dd></dl>", "task_order", "clin_0001", filter="dollars") }}
{{ DefinitionReviewField("<dl><dt>CLIN 0003</dt> - <dd>Unclassified Cloud Support Package</dd></dl>", "task_order", "clin_0003", filter="dollars") }}
{{ DefinitionReviewField("<dl><dt>CLIN 1001</dt> - <dd>Unclassified IaaS and PaaS Amount <br> OPTION PERIOD 1</dd></dl>", "task_order", "clin_1001", filter="dollars") }}
{{ DefinitionReviewField("<dl><dt>CLIN 1003</dt> - <dd>Unclassified Cloud Support Package <br> OPTION PERIOD 1</dd></dl>", "task_order", "clin_1003", filter="dollars") }}
{{ DefinitionReviewField("<dl><dt>CLIN 2001</dt> - <dd>Unclassified IaaS and PaaS Amount <br> OPTION PERIOD 2</dd></dl>", "task_order", "clin_2001", filter="dollars") }}
{{ DefinitionReviewField("<dl><dt>CLIN 2003</dt> - <dd>Unclassified Cloud Support Package <br> OPTION PERIOD 2</dd></dl>", "task_order", "clin_2003", filter="dollars") }}
{{ DefinitionReviewField("Unique Item Identifier (UII)s related to your application(s) if you already have them", "financial_verification", "uii_ids", filter="renderList") }}
{{ DefinitionReviewField("Program Element (PE) Number related to your request", "financial_verification", "pe_id") }}
{{ DefinitionReviewField("Program Treasury Code", "financial_verification", "treasury_code") }}
{{ DefinitionReviewField("Program Budget Activity (BA) Code", "financial_verification", "ba_code") }}
{{ DefinitionReviewField("Contracting Officer First Name", "financial_verification", "fname_co") }}
{{ DefinitionReviewField("Contracting Officer Last Name", "financial_verification", "lname_co") }}
{{ DefinitionReviewField("Contracting Officer Email", "financial_verification", "email_co") }}
{{ DefinitionReviewField("Contracting Officer Office", "financial_verification", "office_co") }}
{{ DefinitionReviewField("Contracting Officer Representative (COR) First Name", "financial_verification", "fname_cor") }}
{{ DefinitionReviewField("Contracting Officer Representative (COR) Last Name", "financial_verification", "lname_cor") }}
{{ DefinitionReviewField("Contracting Officer Representative (COR) Email", "financial_verification", "email_cor") }}
{{ DefinitionReviewField("Contracting Officer Representative (COR) Office", "financial_verification", "office_cor") }}
</dl>
{% endif %}

View File

@ -10,52 +10,13 @@
<form>
<section class='panel'>
<header class='panel__heading request-approval__heading'>
<h1 class='h2'>Request #1234567890</h1>
<span class='label label--info'>Pending</span>
<h1 class='h2'>Request #{{ request_id }}</h1>
<span class='label label--info'>{{ status }}</span>
</header>
<div class='panel__content'>
<p>Ongoing maintainence for Death Star (a moon-sized Imperial military battlestation armed with a planet-destroying superlaser). Its definitely hasn't been sabotaged from the start.</p>
{% with data = {
"primary_poc": {
"am_poc": False,
"dodid_poc": "1234567890",
"email_poc": "fake@email.com",
"fname_poc": "Amanda",
"lname_poc": "Adamson",
},
"details_of_use": {
"jedi_usage": "adf",
"start_date": "2018-08-08",
"cloud_native": "yes",
"dollar_value": 500000,
"dod_component": "Air Force, Department of the",
"data_transfers": "Less than 100GB",
"expected_completion_date": "Less than 1 month",
"jedi_migration": "yes",
"num_software_systems": 1,
"number_user_sessions": 2,
"average_daily_traffic": 1,
"engineering_assessment": "yes",
"technical_support_team": "yes",
"estimated_monthly_spend": 100,
"average_daily_traffic_gb": 4,
"rationalization_software_systems": "yes",
"organization_providing_assistance": "In-house staff",
},
"information_about_you": {
"citizenship": "United States",
"designation": "military",
"phone_number": "1234567890",
"email_request": "fake@email.mil",
"fname_request": "Amanda",
"lname_request": "Adamson",
"service_branch": "Air Force, Department of the",
"date_latest_training": "2018-08-06",
}
}, service_branches=service_branches %}
{% with data=data, request_id=request_id %}
{% include "requests/_review.html" %}
{% endwith %}

View File

@ -0,0 +1,61 @@
import os
from flask import url_for
from atst.models.attachment import Attachment
from atst.domain.roles import Roles
from tests.factories import RequestFactory, TaskOrderFactory, UserFactory
def test_ccpo_can_view_approval(user_session, client):
ccpo = Roles.get("ccpo")
user = UserFactory.create(atat_role=ccpo)
user_session(user)
request = RequestFactory.create()
response = client.get(url_for("requests.approval", request_id=request.id))
assert response.status_code == 200
def test_non_ccpo_cannot_view_approval(user_session, client):
user = UserFactory.create()
user_session(user)
request = RequestFactory.create(creator=user)
response = client.get(url_for("requests.approval", request_id=request.id))
assert response.status_code == 404
def test_task_order_download(app, client, user_session, pdf_upload):
user = UserFactory.create()
user_session(user)
attachment = Attachment.attach(pdf_upload)
task_order = TaskOrderFactory.create(number="abc123", pdf=attachment)
request = RequestFactory.create(task_order=task_order, creator=user)
# ensure that real data for pdf upload has been flushed to disk
pdf_upload.seek(0)
pdf_content = pdf_upload.read()
pdf_upload.close()
full_path = os.path.join(
app.config.get("STORAGE_CONTAINER"), attachment.object_name
)
with open(full_path, "wb") as output_file:
output_file.write(pdf_content)
output_file.flush()
response = client.get(
url_for("requests.task_order_pdf_download", request_id=request.id)
)
assert response.data == pdf_content
def test_task_order_download_does_not_exist(client, user_session):
user = UserFactory.create()
user_session(user)
request = RequestFactory.create(creator=user)
response = client.get(
url_for("requests.task_order_pdf_download", request_id=request.id)
)
assert response.status_code == 404

View File

@ -1,3 +1,5 @@
from flask import url_for
from atst.routes.requests.index import RequestsIndex
from tests.factories import RequestFactory, UserFactory
from atst.domain.requests import Requests
@ -21,3 +23,18 @@ def test_action_required_ccpo():
context = RequestsIndex(ccpo).execute()
assert context["num_action_required"] == 1
def test_ccpo_sees_approval_screen():
ccpo = UserFactory.from_atat_role("ccpo")
request = RequestFactory.create()
Requests.submit(request)
ccpo_context = RequestsIndex(ccpo).execute()
assert ccpo_context["requests"][0]["edit_link"] == url_for(
"requests.approval", request_id=request.id
)
mo_context = RequestsIndex(request.creator).execute()
assert mo_context["requests"][0]["edit_link"] != url_for(
"requests.approval", request_id=request.id
)

View File

@ -31,3 +31,19 @@ def test_upload_fails_for_non_pdfs(uploader):
fs = FileStorage(fp, content_type="text/plain")
with pytest.raises(UploadError):
uploader.upload(fs)
def test_download_stream(upload_dir, uploader, pdf_upload):
# write pdf content to upload file storage and make sure it is flushed to
# disk
pdf_upload.seek(0)
pdf_content = pdf_upload.read()
pdf_upload.close()
full_path = os.path.join(upload_dir, "abc")
with open(full_path, "wb") as output_file:
output_file.write(pdf_content)
output_file.flush()
stream = uploader.download_stream("abc")
stream_content = b"".join([b for b in stream])
assert pdf_content == stream_content