From 0bf32fadd85d5938ae29d489efdd82e2cd137300 Mon Sep 17 00:00:00 2001 From: Max Nanis Date: Thu, 26 Feb 2026 15:51:49 -0500 Subject: Passing Managers into flow tasks for better pytest usage. Conftests broken out into seperate fixture files. Extensive type hinting. --- tests/fixtures/amt.py | 263 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 263 insertions(+) create mode 100644 tests/fixtures/amt.py (limited to 'tests/fixtures/amt.py') diff --git a/tests/fixtures/amt.py b/tests/fixtures/amt.py new file mode 100644 index 0000000..65df125 --- /dev/null +++ b/tests/fixtures/amt.py @@ -0,0 +1,263 @@ +import pytest +import copy + +from datetime import datetime, timedelta +from typing import Callable +from uuid import uuid4 +from dateutil.tz import tzlocal +from mypy_boto3_mturk.type_defs import ( + GetHITResponseTypeDef, + CreateHITTypeResponseTypeDef, + ResponseMetadataTypeDef, + CreateHITWithHITTypeResponseTypeDef, + GetAssignmentResponseTypeDef, +) +from jb.managers.amt import APPROVAL_MESSAGE, NO_WORK_APPROVAL_MESSAGE +from tests import generate_amt_id + +# --- Generic Boto --- + + +@pytest.fixture +def amt_response_metadata() -> ResponseMetadataTypeDef: + req_id = str(uuid4()) + return { + "RequestId": req_id, + "HTTPStatusCode": 200, + "HTTPHeaders": { + "x-amzn-requestid": req_id, + "content-type": "application/x-amz-json-1.1", + "content-length": "46", + "date": "Wed, 15 Oct 2025 02:16:16 GMT", + }, + "RetryAttempts": 0, + } + + +# --- Assignment --- + + +@pytest.fixture +def assignment_response( + amt_hit_id: str, + amt_assignment_id: str, + amt_worker_id: str, + hit_response: GetHITResponseTypeDef, + amt_response_metadata: ResponseMetadataTypeDef, + tsid: str, +) -> GetAssignmentResponseTypeDef: + + hit_response = hit_response["HIT"] + local_now = datetime.now(tz=tzlocal()) + + return { + "Assignment": { + "AssignmentId": amt_assignment_id, + "WorkerId": amt_worker_id, + "HITId": amt_hit_id, + "AssignmentStatus": "Submitted", + "AutoApprovalTime": local_now + timedelta(days=7), + "AcceptTime": local_now - timedelta(minutes=10), + "SubmitTime": local_now, + "Deadline": local_now + timedelta(minutes=90), + "Answer": '\n' + '\n ' + "\n amt_worker_id\n " + f" {amt_worker_id}\n \n \n " + " amt_assignment_id\n " + f" {amt_assignment_id}\n \n \n " + f" tsid\n {tsid}\n " + " \n", + "RequesterFeedback": "Good work", + }, + "HIT": hit_response, + "ResponseMetadata": amt_response_metadata, + } + + +@pytest.fixture +def assignment_response_bad_tsid( + assignment_response: GetAssignmentResponseTypeDef, + amt_worker_id: str, + amt_assignment_id: str, +) -> GetAssignmentResponseTypeDef: + res = copy.deepcopy(assignment_response) + res["Assignment"]["Answer"] = ( + '\n' + '\n ' + "\n amt_worker_id\n " + f" {amt_worker_id}\n \n \n " + " amt_assignment_id\n " + f" {amt_assignment_id}\n \n " + f" \n tsid\n abc123 \n \n" + f"" + ) + return res + + +@pytest.fixture +def assignment_response_no_tsid( + assignment_response: GetAssignmentResponseTypeDef, + amt_worker_id: str, + amt_assignment_id: str, +) -> GetAssignmentResponseTypeDef: + res = copy.deepcopy(assignment_response) + res["Assignment"]["Answer"] = ( + '\n' + '\n ' + "\n amt_worker_id\n " + f" {amt_worker_id}\n \n \n " + " amt_assignment_id\n " + f" {amt_assignment_id}\n \n " + # f"\n tsid\n {tsid}\n \n" + f"" + ) + return res + + +@pytest.fixture +def assignment_response_approved_no_tsid( + assignment_response_no_tsid: GetAssignmentResponseTypeDef, +) -> GetAssignmentResponseTypeDef: + res = copy.deepcopy(assignment_response_no_tsid) + res["Assignment"]["AssignmentStatus"] = "Approved" + res["Assignment"]["RequesterFeedback"] = NO_WORK_APPROVAL_MESSAGE + res["Assignment"]["ApprovalTime"] = res["Assignment"]["SubmitTime"] + return res + + +@pytest.fixture +def assignment_response_factory_approved( + assignment_response: GetAssignmentResponseTypeDef, +) -> Callable[[str], GetAssignmentResponseTypeDef]: + + def inner(feedback: str = APPROVAL_MESSAGE) -> GetAssignmentResponseTypeDef: + res = copy.deepcopy(assignment_response) + res["Assignment"]["AssignmentStatus"] = "Approved" + res["Assignment"]["RequesterFeedback"] = feedback + res["Assignment"]["ApprovalTime"] = res["Assignment"]["SubmitTime"] + return res + + return inner + + +@pytest.fixture +def assignment_response_factory_rejected( + assignment_response: GetAssignmentResponseTypeDef, +) -> Callable[[str], GetAssignmentResponseTypeDef]: + + def inner(reject_reason: str = "reject reason") -> GetAssignmentResponseTypeDef: + res = copy.deepcopy(assignment_response) + res["Assignment"]["AssignmentStatus"] = "Rejected" + res["Assignment"]["RequesterFeedback"] = reject_reason + res["Assignment"]["RejectionTime"] = res["Assignment"]["SubmitTime"] + return res + + return inner + + +@pytest.fixture +def assignment_response_factory_rejected_no_tsid( + assignment_response_no_tsid: GetAssignmentResponseTypeDef, +) -> Callable[[str], GetAssignmentResponseTypeDef]: + + def inner(reject_reason: str = "reject reason") -> GetAssignmentResponseTypeDef: + res = copy.deepcopy(assignment_response_no_tsid) + res["Assignment"]["AssignmentStatus"] = "Rejected" + res["Assignment"]["RequesterFeedback"] = reject_reason + res["Assignment"]["RejectionTime"] = res["Assignment"]["SubmitTime"] + return res + + return inner + + +# --- HITType --- + + +@pytest.fixture +def create_hit_type_response( + amt_hit_type_id: str, amt_response_metadata: ResponseMetadataTypeDef +) -> CreateHITTypeResponseTypeDef: + return { + "HITTypeId": amt_hit_type_id, + "ResponseMetadata": amt_response_metadata, + } + + +# --- HIT --- + + +@pytest.fixture +def hit_response_with_hit_type( + amt_hit_type_id: str, + amt_hit_id: str, + amt_response_metadata: ResponseMetadataTypeDef, +) -> CreateHITWithHITTypeResponseTypeDef: + amt_group_id = generate_amt_id(length=30) + return { + "HIT": { + "HITId": amt_hit_id, + "HITTypeId": amt_hit_type_id, + "HITGroupId": amt_group_id, + "CreationTime": datetime(2025, 10, 14, 20, 22, tzinfo=tzlocal()), + "Title": "Test", + "Description": "test", + "Question": '\n\n https://jamesbillings67.com/work/\n 1200\n ', + "HITStatus": "Assignable", + "MaxAssignments": 1, + "Reward": "0.05", + "AutoApprovalDelayInSeconds": 2_592_000, + "Expiration": datetime(2025, 10, 14, 20, 24, 3, tzinfo=tzlocal()), + "AssignmentDurationInSeconds": 123, + "QualificationRequirements": [], + "HITReviewStatus": "NotReviewed", + "NumberOfAssignmentsPending": 0, + "NumberOfAssignmentsAvailable": 1, + "NumberOfAssignmentsCompleted": 0, + }, + "ResponseMetadata": amt_response_metadata, + } + + +@pytest.fixture +def hit_response( + amt_hit_type_id: str, + amt_hit_id: str, + amt_response_metadata: ResponseMetadataTypeDef, +) -> GetHITResponseTypeDef: + amt_group_id = generate_amt_id(length=30) + return { + "HIT": { + "HITId": amt_hit_id, + "HITTypeId": amt_hit_type_id, + "HITGroupId": amt_group_id, + "CreationTime": datetime(2025, 10, 13, 23, 0, 3, tzinfo=tzlocal()), + "Title": "Awesome Surveys!", + "Description": "Give us your opinion", + "Question": '\n\n https://jamesbillings67.com/work/\n 1200\n ', + "Keywords": "market,research,amazing", + "HITStatus": "Assignable", + "MaxAssignments": 1, + "Reward": "0.05", + "AutoApprovalDelayInSeconds": 604_800, + "Expiration": datetime(2025, 10, 27, 23, 0, 3, tzinfo=tzlocal()), + "AssignmentDurationInSeconds": 5_400, + "QualificationRequirements": [], + "HITReviewStatus": "NotReviewed", + "NumberOfAssignmentsPending": 0, + "NumberOfAssignmentsAvailable": 1, + "NumberOfAssignmentsCompleted": 0, + }, + "ResponseMetadata": amt_response_metadata, + } + + +@pytest.fixture +def hit_response_reviewing( + hit_response: GetHITResponseTypeDef, +) -> GetHITResponseTypeDef: + res = copy.deepcopy(hit_response) + res["HIT"]["NumberOfAssignmentsAvailable"] = 0 + res["HIT"]["NumberOfAssignmentsCompleted"] = 1 + res["HIT"]["HITStatus"] = "Reviewing" + return res -- cgit v1.2.3