60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
import uuid
|
|
|
|
import pytest
|
|
|
|
from proxy_pool.plugins.builtin.parsers.protocol_prefix import ProtocolPrefixParser
|
|
|
|
|
|
@pytest.fixture
|
|
def parser():
|
|
return ProtocolPrefixParser()
|
|
|
|
|
|
class TestParse:
|
|
async def test_parses_socks5(self, parser):
|
|
raw = b"socks5://1.2.3.4:1080\nsocks5://5.6.7.8:9050\n"
|
|
source_id = uuid.uuid4()
|
|
|
|
results = await parser.parse(raw, "http://example.com", source_id, "http")
|
|
|
|
assert len(results) == 2
|
|
assert results[0].ip == "1.2.3.4"
|
|
assert results[0].port == 1080
|
|
assert results[0].protocol == "socks5"
|
|
|
|
async def test_parses_mixed_protocols(self, parser):
|
|
raw = b"http://1.1.1.1:8080\nsocks4://2.2.2.2:1080\nsocks5://3.3.3.3:9050\nhttps://4.4.4.4:443\n"
|
|
|
|
results = await parser.parse(raw, "http://example.com", None, "http")
|
|
|
|
assert len(results) == 4
|
|
protocols = [r.protocol for r in results]
|
|
assert protocols == ["http", "socks4", "socks5", "https"]
|
|
|
|
async def test_handles_uppercase(self, parser):
|
|
raw = b"SOCKS5://1.2.3.4:1080\nHTTP://5.6.7.8:8080\n"
|
|
|
|
results = await parser.parse(raw, "http://example.com", None, "http")
|
|
|
|
assert len(results) == 2
|
|
assert results[0].protocol == "socks5"
|
|
assert results[1].protocol == "http"
|
|
|
|
async def test_skips_invalid_lines(self, parser):
|
|
raw = b"socks5://1.2.3.4:1080\nnot a proxy\nftp://1.2.3.4:21\n\n# comment\n"
|
|
|
|
results = await parser.parse(raw, "http://example.com", None, "http")
|
|
|
|
assert len(results) == 1
|
|
|
|
async def test_ignores_default_protocol(self, parser):
|
|
"""Protocol comes from the line, not from default_protocol."""
|
|
raw = b"socks5://1.2.3.4:1080\n"
|
|
|
|
results = await parser.parse(raw, "http://example.com", None, "http")
|
|
|
|
assert results[0].protocol == "socks5"
|
|
|
|
async def test_empty_input(self, parser):
|
|
results = await parser.parse(b"", "http://example.com", None, "http")
|
|
assert results == [] |