56 lines
3.2 KiB
Python
56 lines
3.2 KiB
Python
import json
|
|
import unittest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from backend.app import db
|
|
from backend.app.services import issue_resolution as service
|
|
from backend.tests.test_backend_quality import TempDatabaseMixin
|
|
|
|
|
|
class IssueAcceptanceTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase):
|
|
def issue(self):
|
|
item = db.create_portal_item(kind="issue", title="Broken <movie>", description="Repair",
|
|
created_by_username="reporter", created_by_id=None, status="in_progress", issue_type="broken_media")
|
|
self.start(item["id"])
|
|
return item
|
|
|
|
def start(self, item_id):
|
|
db.add_portal_item_activity(item_id, event_type="replacement_started", actor_username="reporter",
|
|
actor_role="user", message="New repair", metadata_json=json.dumps({"repairTracking": {"requestId": "12", "actionId": "replace_media"}}))
|
|
|
|
async def test_importing_or_unverified_media_does_not_email_reporter(self):
|
|
self.issue()
|
|
for phase in ["collecting", "indexing", "unavailable"]:
|
|
with patch.object(service, "_media_repair_evidence", new=AsyncMock(return_value={"complete": False, "phase": phase})), patch.object(service, "begin_issue_confirmation", new=AsyncMock()) as begin:
|
|
await service.process_active_media_repairs()
|
|
begin.assert_not_awaited()
|
|
|
|
async def test_verified_repair_emails_once_and_no_requires_a_new_repair(self):
|
|
item = self.issue()
|
|
with (
|
|
patch.object(service, "_media_repair_evidence", new=AsyncMock(return_value={"complete": True, "phase": "complete"})),
|
|
patch.object(service, "_workflow_settings", return_value=(3, 2, "days")),
|
|
patch.object(service, "get_user_by_username", return_value={"username": "reporter"}),
|
|
patch.object(service, "resolve_user_delivery_email", return_value="reporter@example.test"),
|
|
patch.object(service, "send_generic_email", new=AsyncMock()) as email,
|
|
):
|
|
await service.process_active_media_repairs()
|
|
await service.process_active_media_repairs()
|
|
self.assertEqual(email.await_count, 1)
|
|
self.assertEqual(db.get_portal_item(item["id"])["status"], "awaiting_confirmation")
|
|
content = email.await_args.kwargs
|
|
self.assertIn("YES — it works", content["body_html"])
|
|
self.assertIn("NO — still broken", content["body_html"])
|
|
self.assertIn(f"/issues/confirm/{item['id']}#yes", content["body_html"])
|
|
self.assertIn("Broken <movie>", content["body_html"])
|
|
self.assertNotIn("<movie>", content["body_html"])
|
|
self.assertIn("Confirm your answer in Magent", content["body_text"])
|
|
service.respond_to_issue_confirmation(item["id"], resolved=False, actor_username="reporter", actor_role="user")
|
|
await service.process_active_media_repairs()
|
|
await service.process_due_issue_confirmations()
|
|
self.assertEqual(email.await_count, 1)
|
|
self.assertEqual(db.get_portal_item(item["id"])["status"], "in_progress")
|
|
self.start(item["id"])
|
|
await service.process_active_media_repairs()
|
|
self.assertEqual(email.await_count, 2)
|