Skip to content
This repository has been archived by the owner on Aug 10, 2024. It is now read-only.

Commit

Permalink
Add test_poll_state
Browse files Browse the repository at this point in the history
  • Loading branch information
aaron-lane committed Dec 28, 2023
1 parent abee7a3 commit 3fc50c7
Showing 1 changed file with 31 additions and 11 deletions.
42 changes: 31 additions & 11 deletions app/views/test_views.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
download_receipt,
export_csv,
import_csv,
poll_state,
)


Expand Down Expand Up @@ -69,30 +70,49 @@ def test_export_csv_post(self) -> None:
self.assertEqual(first=response.status_code, second=302)

def test_download_receipt(self) -> None:
request = self.request_factory.post(path="")
request.user = self.user
request.queryset = {}
download_receipt_request = self.request_factory.post(path="")
download_receipt_request.user = self.user
download_receipt_request.queryset = {}

post_response = download_receipt(request=request)
download_receipt_response = download_receipt(request=download_receipt_request)

response = self.client.get(path=post_response.url)
get_receipt_response = self.client.get(path=download_receipt_response.url)

self.assertContains(response=response,
self.assertContains(response=get_receipt_response,
text="""<div class="progress">
<div class="bar" role="progressbar"></div>
</div>""",
status_code=200, html=True)

location = post_response.get(header="Location")
location = download_receipt_response.get(header="Location")
parsed_url = urlparse(url=location)
query = parse_qs(qs=parsed_url.query)
task_id = query["job"]
request = self.request_factory.get(path="", data={"task_id": task_id})
request.user = self.user
poll_state_request = self.request_factory.post(path="", data={"task_id": task_id})
poll_state_request.user = self.user

poll_state_response = poll_state(request=poll_state_request)
self.assertContains(response=poll_state_response, text="SUCCESS", status_code=200, html=True)

response = download_file(request=request)
content_type = response["Content-Type"]
download_file_request = self.request_factory.get(path="", data={"task_id": task_id})
download_file_request.user = self.user

download_file_response = download_file(request=download_file_request)
content_type = download_file_response["Content-Type"]

self.assertEqual(first=content_type, second="application/zip", msg=(
"The content type of the receipt response was unexpected. "
"Is Celery running with a results backend enabled?"))

def test_poll_state(self) -> None:
request = self.request_factory.post(path="", data={"task_id": "test-task-id"})
request.user = self.user

response = poll_state(request=request)

print(response.content)
self.assertEqual(first=response.status_code, second=200)
self.assertJSONEqual(
raw=response.content,
expected_data={"state": "PENDING", "process_percent": 0, "status": 200},
)

0 comments on commit 3fc50c7

Please sign in to comment.