98 lines
2.8 KiB
Python
98 lines
2.8 KiB
Python
|
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 "<html>this_is_a_fake_html_string</html>"
|
||
|
|
|
||
|
|
|
||
|
|
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
|