diff --git a/tests/capturer_test.py b/tests/capturer_test.py new file mode 100644 index 0000000..da60050 --- /dev/null +++ b/tests/capturer_test.py @@ -0,0 +1,170 @@ +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" + ) diff --git a/tests/capturer_tests.py b/tests/capturer_tests.py deleted file mode 100644 index 772b1f5..0000000 --- a/tests/capturer_tests.py +++ /dev/null @@ -1,236 +0,0 @@ -# -*- coding: utf-8 -*- -import sys - -sys.path.append("..") -from capturer.capturer import CapturingTask, Capturer, AdHtmlParser -from db_layer.capturas_interface import capturas_interface - - -def test_CapturingTask(): - parameters = { - "uuid": "testie test", - "ad_url": "https://www.idealista.com/inmueble/28252032", - "fk_uuid_exploring": None, - "status": "Pending", - } - - task = CapturingTask(parameters) - - task.capture() - print(task.get_ad_data()) - capturas_interface.insert_captura(task.get_ad_data()) - - -def test_Capturer(): - capturer = Capturer() - capturer.start() - - -def test_AdHtmlParser(): - - html = """ - - - - - - - - - - - - - - - Alquiler de Garaje en calle de Balmes, 138, La Dreta de l'Eixample, Barcelona

Alquiler de Garaje en calle de Balmes, 138 La Dreta de l'Eixample, Barcelona Ver mapa

30 €/mes
1 m²

Comentario del anunciante

Disponible en: Español English

Características básicas

  • 1 m²

¿Hay algún error en este anuncio?

Infórmanos para corregirlo y ayudar a otros usuarios.

Cuéntanos qué error has visto

¿Cuánto vale este inmueble?

Te enviamos un informe con la estimación de precio para este inmueble y con información de la zona.

Comprar estimación de precio

Ubicación

  • Calle de Balmes, 138
  • Urb. Eixample esquerra
  • Barrio La Dreta de l'Eixample
  • Distrito Eixample
  • Barcelona
  • Área de Barcelona, Barcelona

Estadísticas

- - - - """ - - parser = AdHtmlParser(html) - - parser.parse() - parser._validate() - - -# test_AdHtmlParser() - -test_CapturingTask() - -# test_Capturer() diff --git a/tests/mock_classes.py b/tests/mock_classes.py new file mode 100644 index 0000000..57bbbcf --- /dev/null +++ b/tests/mock_classes.py @@ -0,0 +1,97 @@ +from collections import namedtuple +from typing import Dict + +from bs4 import BeautifulSoup + +from db_layer.capturing_tasks_interface import CapturingTasksInterface +from core.parsing_utils import ParsingFlow +from core.scrapping_utils import UrlAttack + + +class MockCapturingInterface(CapturingTasksInterface): + + task_state_record = namedtuple( + "TaskStateRecord", ["uuid", "uuid_exploring", "status", "ad_url"] + ) + + def __init__(self): + self.tasks = {} + + def update_capturing_task(self, uuid, uuid_exploring, status, ad_url): + if uuid not in self.tasks: + self.tasks[uuid] = [] + + self.tasks[uuid].append( + MockCapturingInterface.task_state_record( + uuid=uuid, uuid_exploring=uuid_exploring, status=status, ad_url=ad_url + ) + ) + + +class MockParsingFlow(ParsingFlow): + def __init__( + self, + issues_to_return: Dict[str, dict] = None, + mock_all_found_fields_are_valid: bool = True, + mock_all_non_optional_fields_were_found: bool = True, + mock_field_values_to_return: Dict[str, dict] = None, + ): + args_with_empty_dict_as_default = [ + issues_to_return, + mock_field_values_to_return, + ] + for arg in args_with_empty_dict_as_default: + if arg is None: + arg = dict() + + self._issues = issues_to_return + self._mock_all_found_fields_are_valid = mock_all_found_fields_are_valid + self._mock_field_values_to_return = mock_field_values_to_return + self._mock_all_non_optional_fields_were_found = ( + mock_all_non_optional_fields_were_found + ) + + def execute_flow(self, soup: BeautifulSoup) -> None: + pass + + @property + def issues(self) -> Dict[str, dict]: + return self._issues + + @property + def all_found_fields_are_valid(self) -> bool: + return self._mock_all_found_fields_are_valid + + @property + def all_non_optional_fields_were_found(self) -> bool: + return self._mock_all_non_optional_fields_were_found + + @property + def field_values(self) -> Dict: + return self._mock_field_values_to_return + + +class MockUrlAttack(UrlAttack): + def __init__(self, url: str) -> None: + super().__init__(url=url) + + def get_text(self) -> str: + return "this_is_a_fake_html_string" + + +class MockUrlAttackReturnsSuccess(MockUrlAttack): + def __init__(self, url: str) -> None: + super().__init__(url=url) + + def attack(self) -> None: + self.success = True + self.has_been_attacked = True + + +class MockUrlAttackReturnsFailure(MockUrlAttack): + def __init__(self, url: str) -> None: + super().__init__(url=url) + + def attack(self) -> None: + self.success = False + self.has_been_attacked = True