64 lines
1.8 KiB
Python
64 lines
1.8 KiB
Python
import asyncio
|
|
from pathlib import Path
|
|
|
|
from vk_parser_app.workers.media_uploader import MediaUploader, video_provider
|
|
|
|
|
|
class FakeContent:
|
|
async def iter_chunked(self, _size):
|
|
yield b"video"
|
|
|
|
|
|
class FakeResponse:
|
|
status = 200
|
|
content = FakeContent()
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_args):
|
|
return None
|
|
|
|
|
|
class FakeSession:
|
|
def get(self, *_args, **_kwargs):
|
|
return FakeResponse()
|
|
|
|
|
|
def test_video_provider() -> None:
|
|
assert video_provider("https://vk.com/video-1_2") == "vk"
|
|
assert video_provider("https://youtu.be/dQw4w9WgXcQ") == "youtube"
|
|
assert video_provider("https://www.youtube.com/watch?v=dQw4w9WgXcQ") == "youtube"
|
|
assert video_provider("file:///etc/passwd") is None
|
|
assert video_provider("http://127.0.0.1/video.mp4") is None
|
|
|
|
|
|
def test_site_request_options() -> None:
|
|
options = MediaUploader().site_request_options({
|
|
"platform": "site",
|
|
"source_url": "https://example.test/rss.xml",
|
|
"source_runtime_state": {
|
|
"browser_user_agent": "Test Browser",
|
|
"browser_state": {"cookies": [{"name": "cf_clearance", "value": "secret"}]},
|
|
},
|
|
})
|
|
assert options["headers"] == {
|
|
"User-Agent": "Test Browser",
|
|
"Referer": "https://example.test/rss.xml",
|
|
}
|
|
assert options["cookies"] == {"cf_clearance": "secret"}
|
|
|
|
|
|
async def test_http_video_download() -> None:
|
|
path = "/tmp/vkparser_tg_media_http_test.mp4"
|
|
result = await MediaUploader().download_http_video(FakeSession(), "https://example.test/a.mp4", path)
|
|
assert result["size_bytes"] == 5
|
|
assert Path(path).read_bytes() == b"video"
|
|
Path(path).unlink()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_video_provider()
|
|
test_site_request_options()
|
|
asyncio.run(test_http_video_download())
|