Passing Managers into flow tasks for better pytest usage. Conftests broken out into seperate fixture files. Extensive type hinting.

1 files changed, 263 insertions, 0 deletions

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": '<?xml version="1.0" encoding="UTF-8"?>\n'
+            '<QuestionFormAnswers xmlns="http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2005-10-01/QuestionFormAnswers.xsd">\n '
+            "<Answer>\n    <QuestionIdentifier>amt_worker_id</QuestionIdentifier>\n   "
+            f" <FreeText>{amt_worker_id}</FreeText>\n  </Answer>\n  <Answer>\n   "
+            " <QuestionIdentifier>amt_assignment_id</QuestionIdentifier>\n "
+            f"   <FreeText>{amt_assignment_id}</FreeText>\n  </Answer>\n  <Answer>\n "
+            f"   <QuestionIdentifier>tsid</QuestionIdentifier>\n    <FreeText>{tsid}</FreeText>\n "
+            " </Answer>\n</QuestionFormAnswers>",
+            "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"] = (
+        '<?xml version="1.0" encoding="UTF-8"?>\n'
+        '<QuestionFormAnswers xmlns="http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2005-10-01/QuestionFormAnswers.xsd">\n '
+        "<Answer>\n    <QuestionIdentifier>amt_worker_id</QuestionIdentifier>\n   "
+        f" <FreeText>{amt_worker_id}</FreeText>\n  </Answer>\n  <Answer>\n   "
+        " <QuestionIdentifier>amt_assignment_id</QuestionIdentifier>\n "
+        f"   <FreeText>{amt_assignment_id}</FreeText>\n  </Answer>\n "
+        f" <Answer>\n  <QuestionIdentifier>tsid</QuestionIdentifier>\n  abc123 <FreeText></FreeText>\n </Answer>\n"
+        f"</QuestionFormAnswers>"
+    )
+    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"] = (
+        '<?xml version="1.0" encoding="UTF-8"?>\n'
+        '<QuestionFormAnswers xmlns="http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2005-10-01/QuestionFormAnswers.xsd">\n '
+        "<Answer>\n    <QuestionIdentifier>amt_worker_id</QuestionIdentifier>\n   "
+        f" <FreeText>{amt_worker_id}</FreeText>\n  </Answer>\n  <Answer>\n   "
+        " <QuestionIdentifier>amt_assignment_id</QuestionIdentifier>\n "
+        f"   <FreeText>{amt_assignment_id}</FreeText>\n  </Answer>\n  "
+        # f"<Answer>\n  <QuestionIdentifier>tsid</QuestionIdentifier>\n  <FreeText>  {tsid}</FreeText>\n </Answer>\n"
+        f"</QuestionFormAnswers>"
+    )
+    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": '<?xml version="1.0" encoding="UTF-8"?>\n<ExternalQuestion xmlns="http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2006-07-14/ExternalQuestion.xsd">\n                    <ExternalURL>https://jamesbillings67.com/work/</ExternalURL>\n                    <FrameHeight>1200</FrameHeight>\n                    </ExternalQuestion>',
+            "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": '<?xml version="1.0" encoding="UTF-8"?>\n<ExternalQuestion xmlns="http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2006-07-14/ExternalQuestion.xsd">\n                    <ExternalURL>https://jamesbillings67.com/work/</ExternalURL>\n                    <FrameHeight>1200</FrameHeight>\n                    </ExternalQuestion>',
+            "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