Your Name
108 lines
3.7 KiB
Python
108 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from vk_parser_app.source_adapters import SiteParserClient, validate_source_config
|
|
|
|
|
|
class FakeResponse:
|
|
status = 200
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_args):
|
|
return None
|
|
|
|
async def json(self, **_kwargs):
|
|
return {
|
|
"items": [{
|
|
"external_id": "post-1",
|
|
"url": "https://example.test/1",
|
|
"title": "Title",
|
|
"text": "Body",
|
|
"published_at": "2026-08-10T10:00:00+00:00",
|
|
"media": [{"type": "photo", "url": "https://example.test/1.jpg"}],
|
|
}],
|
|
"browser_state": {"cookies": [{"name": "cf_clearance"}]},
|
|
}
|
|
|
|
|
|
class FollowedPageResponse(FakeResponse):
|
|
async def json(self, **_kwargs):
|
|
data = await super().json(**_kwargs)
|
|
data["items"][0]["text"] = "Title\nAuthor"
|
|
return data
|
|
|
|
|
|
class EmptyBodyResponse(FakeResponse):
|
|
async def json(self, **_kwargs):
|
|
data = await super().json(**_kwargs)
|
|
data["items"][0]["text"] = ""
|
|
return data
|
|
|
|
|
|
class FakeSession:
|
|
def post(self, *_args, **_kwargs):
|
|
return FakeResponse()
|
|
|
|
|
|
class FollowedPageSession(FakeSession):
|
|
def post(self, *_args, **_kwargs):
|
|
return FollowedPageResponse()
|
|
|
|
|
|
class EmptyBodySession(FakeSession):
|
|
def post(self, *_args, **_kwargs):
|
|
return EmptyBodyResponse()
|
|
|
|
|
|
class SourceAdapterTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_worker_response_is_normalized(self) -> None:
|
|
client = SiteParserClient(FakeSession(), "http://worker", "token", "captcha", 30)
|
|
items, state = await client.fetch({
|
|
"url": "https://example.test/rss.xml",
|
|
"settings_json": '{"format":"rss","access":"auto"}',
|
|
"runtime_state_json": '{}',
|
|
})
|
|
|
|
self.assertEqual(items[0].text, "Title\n\nBody")
|
|
self.assertEqual(items[0].media[0].url, "https://example.test/1.jpg")
|
|
self.assertEqual(state["browser_state"]["cookies"][0]["name"], "cf_clearance")
|
|
|
|
async def test_followed_page_does_not_repeat_title(self) -> None:
|
|
client = SiteParserClient(FollowedPageSession(), "http://worker", "token", "captcha", 30)
|
|
items, _ = await client.fetch({
|
|
"url": "https://example.test/rss.xml",
|
|
"settings_json": '{"format":"rss","follow_links":true,"content_selector":"article.full"}',
|
|
"runtime_state_json": "{}",
|
|
})
|
|
self.assertEqual(items[0].text, "Title\nAuthor")
|
|
self.assertEqual(items[0].body_text, "Title\nAuthor")
|
|
|
|
async def test_title_is_not_counted_as_site_body(self) -> None:
|
|
client = SiteParserClient(EmptyBodySession(), "http://worker", "token", "captcha", 30)
|
|
items, _ = await client.fetch({
|
|
"url": "https://example.test/rss.xml",
|
|
"settings_json": '{"format":"rss"}',
|
|
"runtime_state_json": "{}",
|
|
})
|
|
self.assertEqual(items[0].text, "Title")
|
|
self.assertEqual(items[0].body_text, "")
|
|
|
|
def test_site_config_is_required(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "нужен конфиг"):
|
|
validate_source_config("site", {})
|
|
|
|
def test_follow_links_requires_selector(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "content_selector"):
|
|
validate_source_config("site", {"format": "rss", "follow_links": True})
|
|
|
|
def test_site_min_text_length_is_validated(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "min_text_length"):
|
|
validate_source_config("site", {"format": "rss", "min_text_length": "many"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|