73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
from dataclasses import asdict
|
|
|
|
from apps.parsers import tasks
|
|
from apps.parsers.clients.common.structured import MAX_FILE_SIZE_BYTES
|
|
from apps.parsers.serializers import ParserRunRequestSerializer
|
|
from apps.parsers.source_registry import PARSER_SOURCES
|
|
from apps.parsers.views import TASKS_BY_NAME
|
|
from django.conf import settings
|
|
from django.test import SimpleTestCase
|
|
|
|
|
|
class ParserSourceRegistryEisTest(SimpleTestCase):
|
|
def test_procurement_urls_select_the_expected_law(self):
|
|
common_search_url = (
|
|
"https://zakupki.gov.ru/epz/order/extendedsearch/results.html"
|
|
)
|
|
|
|
self.assertEqual(
|
|
PARSER_SOURCES["procurements_44fz"].upstream_url,
|
|
f"{common_search_url}?fz44=on",
|
|
)
|
|
self.assertEqual(
|
|
PARSER_SOURCES["procurements_223fz"].upstream_url,
|
|
f"{common_search_url}?fz223=on",
|
|
)
|
|
|
|
|
|
class ParserSourceRegistryFNSTest(SimpleTestCase):
|
|
def test_request_body_limit_allows_maximum_registry_archive(self):
|
|
self.assertGreater(settings.DATA_UPLOAD_MAX_MEMORY_SIZE, MAX_FILE_SIZE_BYTES)
|
|
|
|
def test_fns_refresh_uses_bounded_api_task_and_preserves_file_upload(self):
|
|
source = PARSER_SOURCES["fns_financial"]
|
|
|
|
self.assertEqual(
|
|
source.task_name,
|
|
"apps.parsers.tasks.sync_fns_financial_reports",
|
|
)
|
|
self.assertEqual(source.parser_strategy, "fns_bfo_bounded_registry_sync")
|
|
self.assertTrue(source.supports_file_upload)
|
|
self.assertNotIn("query=", source.upstream_url)
|
|
self.assertIs(
|
|
TASKS_BY_NAME[source.task_name],
|
|
tasks.sync_fns_financial_reports,
|
|
)
|
|
|
|
|
|
class ParserSourceRegistryPresentationTest(SimpleTestCase):
|
|
def test_public_source_metadata_does_not_name_external_provider(self):
|
|
public_metadata = " ".join(
|
|
str(value)
|
|
for descriptor in PARSER_SOURCES.values()
|
|
for value in asdict(descriptor).values()
|
|
)
|
|
|
|
self.assertNotIn("checko", public_metadata.lower())
|
|
|
|
def test_vacancy_source_metadata_names_only_trudvsem(self):
|
|
source = PARSER_SOURCES["trudvsem"]
|
|
|
|
self.assertEqual(source.agency, "Работа России")
|
|
self.assertEqual(source.parser_strategy, "incremental_trudvsem_api")
|
|
self.assertNotIn("HH", source.data_scope)
|
|
self.assertNotIn("SuperJob", source.data_scope)
|
|
|
|
def test_vacancy_run_request_rejects_disabled_sources(self):
|
|
serializer = ParserRunRequestSerializer(
|
|
data={"vacancy_sources": ["hh", "superjob"]}
|
|
)
|
|
|
|
self.assertFalse(serializer.is_valid())
|
|
self.assertIn("vacancy_sources", serializer.errors)
|