path: root/tests/conftest.py

import copy
from datetime import datetime, timezone, timedelta
from typing import Optional
from uuid import uuid4

import pytest
from dateutil.tz import tzlocal
from mypy_boto3_mturk.type_defs import (
    GetHITResponseTypeDef,
    CreateHITTypeResponseTypeDef,
    CreateHITWithHITTypeResponseTypeDef,
    GetAssignmentResponseTypeDef,
)

from jb.decorators import HQM, HTM, HM, AM
from jb.managers.amt import AMTManager, APPROVAL_MESSAGE, NO_WORK_APPROVAL_MESSAGE
from jb.models.assignment import AssignmentStub, Assignment
from jb.models.currency import USDCent
from jb.models.definitions import HitStatus, HitReviewStatus, AssignmentStatus
from jb.models.hit import HitType, HitQuestion, Hit
from tests import generate_amt_id


@pytest.fixture
def amt_hit_type_id():
    return generate_amt_id()


@pytest.fixture
def amt_hit_id():
    return generate_amt_id()


@pytest.fixture
def amt_assignment_id():
    return generate_amt_id()


@pytest.fixture
def amt_worker_id():
    return generate_amt_id(length=21)


@pytest.fixture
def amt_group_id():
    return generate_amt_id()


@pytest.fixture
def tsid():
    return uuid4().hex


@pytest.fixture
def tsid1():
    return uuid4().hex


@pytest.fixture
def tsid2():
    return uuid4().hex


@pytest.fixture
def pe_id():
    # payout event / cashout request UUID
    return uuid4().hex


@pytest.fixture
def hit_type() -> HitType:
    return HitType(
        title="Awesome Surveys!",
        description="Give us your opinion",
        reward=USDCent(5),
        keywords="market,research,amazing",
        min_active=10,
    )


from jb.models.hit import HitType


@pytest.fixture
def hit_type_with_amt_id(hit_type: HitType) -> HitType:
    # This is a real hit type I've previously registered with amt (sandbox).
    # It will always exist
    hit_type.amt_hit_type_id = "3217B3DC4P5YW9DRV9R3X8O56V041J"

    # Get or create our db
    HTM.get_or_create(hit_type)
    # this call adds the pk int id ---^

    return hit_type


@pytest.fixture
def question():
    return HitQuestion(url="https://jamesbillings67.com/work/", height=1200)


@pytest.fixture
def hit_in_amt(hit_type_with_amt_id: HitType, question: HitQuestion) -> Hit:
    # Actually create a new HIT in amt (sandbox)
    question = HQM.get_or_create(question)
    hit = AMTManager.create_hit_with_hit_type(
        hit_type=hit_type_with_amt_id, question=question
    )
    # Create it in the DB
    HM.create(hit)
    return hit


@pytest.fixture
def hit(amt_hit_id, amt_hit_type_id, amt_group_id, question):
    now = datetime.now(tz=timezone.utc)
    return Hit.model_validate(
        dict(
            amt_hit_id=amt_hit_id,
            amt_hit_type_id=amt_hit_type_id,
            amt_group_id=amt_group_id,
            status=HitStatus.Assignable,
            review_status=HitReviewStatus.NotReviewed,
            creation_time=now,
            expiration=now + timedelta(days=3),
            hit_question_xml=question.xml,
            qualification_requirements=[],
            max_assignments=1,
            assignment_pending_count=0,
            assignment_available_count=1,
            assignment_completed_count=0,
            description="Description",
            keywords="Keywords",
            reward=USDCent(5),
            title="Title",
            question_id=question.id,
            hit_type_id=None,
        )
    )


@pytest.fixture
def hit_in_db(
    hit_type: HitType, amt_hit_type_id, amt_hit_id, question: HitQuestion, hit: Hit
) -> Hit:
    """
    Returns a hit that exists in our db, but does not in amazon (the amt ids
        are random). The mtwerk_hittype and mtwerk_question records will also
        exist (in the db)
    """
    question = HQM.get_or_create(question)
    hit_type.amt_hit_type_id = amt_hit_type_id
    HTM.create(hit_type)
    hit.hit_type_id = hit_type.id
    hit.amt_hit_id = amt_hit_id
    hit.question_id = question.id
    HM.create(hit)
    return hit


@pytest.fixture
def assignment_stub(hit: Hit, amt_assignment_id, amt_worker_id):
    now = datetime.now(tz=timezone.utc)
    return AssignmentStub(
        amt_assignment_id=amt_assignment_id,
        amt_hit_id=hit.amt_hit_id,
        amt_worker_id=amt_worker_id,
        status=AssignmentStatus.Submitted,
        modified_at=now,
        created_at=now,
    )


@pytest.fixture
def assignment_factory(hit: Hit):
    def inner(amt_worker_id: str = None):
        now = datetime.now(tz=timezone.utc)
        amt_assignment_id = generate_amt_id()
        amt_worker_id = amt_worker_id or generate_amt_id()
        return Assignment(
            amt_assignment_id=amt_assignment_id,
            amt_hit_id=hit.amt_hit_id,
            amt_worker_id=amt_worker_id,
            status=AssignmentStatus.Submitted,
            modified_at=now,
            created_at=now,
            accept_time=now,
            auto_approval_time=now,
            submit_time=now,
        )

    return inner


@pytest.fixture
def assignment_in_db_factory(assignment_factory):
    def inner(hit_id: int, amt_worker_id: Optional[str] = None):
        a = assignment_factory(amt_worker_id=amt_worker_id)
        a.hit_id = hit_id
        AM.create_stub(a)
        AM.update_answer(a)
        return a

    return inner


@pytest.fixture
def assignment_stub_in_db(hit_in_db, assignment_stub) -> AssignmentStub:
    """
    Returns an AssignmentStub that exists in our db, but does not in amazon (the amt ids are random).
    The mtwerk_hit, mtwerk_hittype, and mtwerk_question records will also exist (in the db)
    """
    assignment_stub.hit_id = hit_in_db.id
    AM.create_stub(assignment_stub)
    return assignment_stub


@pytest.fixture
def amt_response_metadata():
    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,
    }


@pytest.fixture
def create_hit_type_response(
    amt_hit_type_id, amt_response_metadata
) -> CreateHITTypeResponseTypeDef:
    return {
        "HITTypeId": amt_hit_type_id,
        "ResponseMetadata": amt_response_metadata,
    }


@pytest.fixture
def create_hit_with_hit_type_response(
    amt_hit_type_id, amt_hit_id, amt_response_metadata
) -> 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 get_hit_response(
    amt_hit_type_id, amt_hit_id, amt_response_metadata
) -> 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 get_hit_response_reviewing(get_hit_response):
    res = copy.deepcopy(get_hit_response)
    res["HIT"]["NumberOfAssignmentsAvailable"] = 0
    res["HIT"]["NumberOfAssignmentsCompleted"] = 1
    res["HIT"]["HITStatus"] = "Reviewing"
    return res


@pytest.fixture
def get_assignment_response(
    amt_hit_type_id,
    amt_hit_id,
    amt_assignment_id,
    amt_worker_id,
    get_hit_response,
    amt_response_metadata,
    tsid,
) -> GetAssignmentResponseTypeDef:
    hit_response = get_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 get_assignment_response_no_tsid(
    get_assignment_response, amt_worker_id, amt_assignment_id
):
    res = copy.deepcopy(get_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 get_assignment_response_approved(
    get_assignment_response: GetAssignmentResponseTypeDef,
):
    def inner(feedback: str = APPROVAL_MESSAGE) -> GetAssignmentResponseTypeDef:
        res = copy.deepcopy(get_assignment_response)
        res["Assignment"]["AssignmentStatus"] = "Approved"
        res["Assignment"]["RequesterFeedback"] = feedback
        res["Assignment"]["ApprovalTime"] = res["Assignment"]["SubmitTime"]
        return res

    return inner


@pytest.fixture
def get_assignment_response_rejected(
    get_assignment_response: GetAssignmentResponseTypeDef,
):

    def inner(reject_reason: str = "reject reason") -> GetAssignmentResponseTypeDef:
        res = copy.deepcopy(get_assignment_response)
        res["Assignment"]["AssignmentStatus"] = "Rejected"
        res["Assignment"]["RequesterFeedback"] = reject_reason
        res["Assignment"]["RejectionTime"] = res["Assignment"]["SubmitTime"]
        return res

    return inner


@pytest.fixture
def get_assignment_response_rejected_no_tsid(
    get_assignment_response_no_tsid: GetAssignmentResponseTypeDef,
):

    def inner(reject_reason: str = "reject reason") -> GetAssignmentResponseTypeDef:
        res = copy.deepcopy(get_assignment_response_no_tsid)
        res["Assignment"]["AssignmentStatus"] = "Rejected"
        res["Assignment"]["RequesterFeedback"] = reject_reason
        res["Assignment"]["RejectionTime"] = res["Assignment"]["SubmitTime"]
        return res

    return inner


@pytest.fixture
def get_assignment_response_approved_no_tsid(
    get_assignment_response_no_tsid: GetAssignmentResponseTypeDef,
):
    res = copy.deepcopy(get_assignment_response_no_tsid)
    res["Assignment"]["AssignmentStatus"] = "Approved"
    res["Assignment"]["RequesterFeedback"] = NO_WORK_APPROVAL_MESSAGE
    res["Assignment"]["ApprovalTime"] = res["Assignment"]["SubmitTime"]
    return res