66 lines
2 KiB
Python
66 lines
2 KiB
Python
import asyncio
|
|
import pytest
|
|
from aiohttp import web
|
|
from src.browser.manager import CamoufoxManager
|
|
from src.extractor.client import CurlClient
|
|
|
|
# Global to store received headers for verification
|
|
received_headers = []
|
|
|
|
async def handle_request(request):
|
|
"""
|
|
Mock endpoint that logs headers.
|
|
"""
|
|
headers = dict(request.headers)
|
|
received_headers.append(headers)
|
|
return web.json_response({"status": "ok", "headers": headers})
|
|
|
|
async def start_mock_server(port=8080):
|
|
app = web.Application()
|
|
app.router.add_get('/', handle_request)
|
|
runner = web.AppRunner(app)
|
|
await runner.setup()
|
|
site = web.TCPSite(runner, 'localhost', port)
|
|
await site.start()
|
|
return runner
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handover_e2e():
|
|
"""
|
|
Test the full loop:
|
|
1. Browser authenticates (hits mock) and extracts state.
|
|
2. Extractor initializes with state and hits mock.
|
|
3. Verify User-Agent consistency.
|
|
"""
|
|
port = 8888
|
|
base_url = f"http://localhost:{port}/"
|
|
runner = await start_mock_server(port)
|
|
|
|
received_headers.clear()
|
|
|
|
try:
|
|
# 1. Browser Phase
|
|
async with CamoufoxManager() as browser:
|
|
await browser.navigate(base_url)
|
|
session_state = await browser.extract_session_state()
|
|
|
|
assert len(received_headers) == 1
|
|
browser_headers = received_headers[0]
|
|
|
|
# 2. Extractor Phase
|
|
async with CurlClient(session_state) as extractor:
|
|
await extractor.fetch(base_url)
|
|
|
|
assert len(received_headers) == 2
|
|
extractor_headers = received_headers[1]
|
|
|
|
# 3. Verification
|
|
print(f"Browser UA: {browser_headers.get('User-Agent')}")
|
|
print(f"Extractor UA: {extractor_headers.get('User-Agent')}")
|
|
|
|
assert browser_headers.get('User-Agent') == extractor_headers.get('User-Agent')
|
|
# Note: Other headers might vary slightly due to browser vs curl defaults,
|
|
# but UA must be exact.
|
|
|
|
finally:
|
|
await runner.cleanup()
|