drogon/tests/capturer_test.py

from tests.mock_classes import (
    MockCapturingInterface,
    MockParsingFlow,
    MockUrlAttackReturnsSuccess,
    MockUrlAttackReturnsFailure,
)
from capturer.capturer import CapturingTask


def test_capturing_task_successful_task_flow():

    the_task_parameters = dict()
    the_task_parameters["uuid"] = "test_uuid"
    the_task_parameters["ad_url"] = "test_url"
    the_task_parameters["fk_uuid_exploring"] = "test_exploring_uuid"
    the_task_parameters["status"] = "Pending"

    fake_resulting_field_values = {
        "a_field": {"a_value": 1},
        "another_field": {"another_value": 2},
    }
    mock_parsing_flow = MockParsingFlow(
        mock_all_found_fields_are_valid=True,
        mock_all_non_optional_fields_were_found=True,
        mock_field_values_to_return=fake_resulting_field_values,
    )

    mock_capturing_interface = MockCapturingInterface()

    task = CapturingTask(
        task_parameters=the_task_parameters,
        capturing_interface=mock_capturing_interface,
        new_parsing_flow=mock_parsing_flow,
        url_acquisition_object=MockUrlAttackReturnsSuccess,
        dead_ad_checker=lambda: False,
    )

    task.capture()

    final_data = task.get_ad_data()

    assert (
        len(mock_capturing_interface.tasks) == 1
        and mock_capturing_interface.tasks[the_task_parameters["uuid"]][-1].status
        == "Data ready"
        and fake_resulting_field_values == final_data
    )


def test_capturing_task_dead_ad_task_flow():
    the_task_parameters = dict()
    the_task_parameters["uuid"] = "test_uuid"
    the_task_parameters["ad_url"] = "test_url"
    the_task_parameters["fk_uuid_exploring"] = "test_exploring_uuid"
    the_task_parameters["status"] = "Pending"

    mock_parsing_flow = MockParsingFlow(
        mock_all_found_fields_are_valid=False,
        issues_to_return={"some_field": {"valid": False}},
    )

    mock_capturing_interface = MockCapturingInterface()

    task = CapturingTask(
        task_parameters=the_task_parameters,
        capturing_interface=mock_capturing_interface,
        new_parsing_flow=mock_parsing_flow,
        url_acquisition_object=MockUrlAttackReturnsFailure,
        dead_ad_checker=lambda x: True,
    )

    task.capture()

    assert (
        len(mock_capturing_interface.tasks) == 1
        and mock_capturing_interface.tasks[the_task_parameters["uuid"]][-1].status
        == "Dead ad"
    )


def test_capturing_task_invalid_fields_surrender_flow():
    the_task_parameters = dict()
    the_task_parameters["uuid"] = "test_uuid"
    the_task_parameters["ad_url"] = "test_url"
    the_task_parameters["fk_uuid_exploring"] = "test_exploring_uuid"
    the_task_parameters["status"] = "Pending"

    mock_parsing_flow = MockParsingFlow(
        mock_all_found_fields_are_valid=False,
        issues_to_return={"some_field": {"valid": False}},
    )

    mock_capturing_interface = MockCapturingInterface()

    task = CapturingTask(
        task_parameters=the_task_parameters,
        capturing_interface=mock_capturing_interface,
        new_parsing_flow=mock_parsing_flow,
        url_acquisition_object=MockUrlAttackReturnsSuccess,
        dead_ad_checker=lambda: False,
    )

    task.capture()

    assert (
        len(mock_capturing_interface.tasks) == 1
        and mock_capturing_interface.tasks[the_task_parameters["uuid"]][-1].status
        == "Invalid value fields"
    )


def test_capturing_task_missing_fields_surrender_flow():
    the_task_parameters = dict()
    the_task_parameters["uuid"] = "test_uuid"
    the_task_parameters["ad_url"] = "test_url"
    the_task_parameters["fk_uuid_exploring"] = "test_exploring_uuid"
    the_task_parameters["status"] = "Pending"

    mock_parsing_flow = MockParsingFlow(
        mock_all_non_optional_fields_were_found=False,
        issues_to_return={"some_field": {"found": False}},
    )

    mock_capturing_interface = MockCapturingInterface()

    task = CapturingTask(
        task_parameters=the_task_parameters,
        capturing_interface=mock_capturing_interface,
        new_parsing_flow=mock_parsing_flow,
        url_acquisition_object=MockUrlAttackReturnsSuccess,
        dead_ad_checker=lambda: False,
    )

    task.capture()

    assert (
        len(mock_capturing_interface.tasks) == 1
        and mock_capturing_interface.tasks[the_task_parameters["uuid"]][-1].status
        == "Fields missing"
    )


def test_capturing_task_unexpected_issue_surrender_flow():
    the_task_parameters = dict()
    the_task_parameters["uuid"] = "test_uuid"
    the_task_parameters["ad_url"] = "test_url"
    the_task_parameters["fk_uuid_exploring"] = "test_exploring_uuid"
    the_task_parameters["status"] = "Pending"

    mock_parsing_flow = MockParsingFlow()

    mock_capturing_interface = MockCapturingInterface()

    CapturingTask.sleep_time_failed_request = 0  # Override quite long sleep time

    task = CapturingTask(
        task_parameters=the_task_parameters,
        capturing_interface=mock_capturing_interface,
        new_parsing_flow=mock_parsing_flow,
        url_acquisition_object=MockUrlAttackReturnsFailure,
        dead_ad_checker=lambda x: False,
    )

    task.capture()

    assert (
        len(mock_capturing_interface.tasks) == 1
        and mock_capturing_interface.tasks[the_task_parameters["uuid"]][-1].status
        == "Surrender"
    )