LiuFan
/
PrivacyScanData


			
				
					
						
						
							123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208
							# Copyright (c) 2017 crocoite contributors
# 
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# 
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
# 
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

import asyncio
import pytest

from aiohttp import web
import websockets

from .devtools import Browser, Tab, MethodNotFound, Crashed, \
        InvalidParameter, Process, Passthrough

@pytest.fixture
async def browser ():
    async with Process () as url:
        yield Browser (url)

@pytest.fixture
async def tab (browser):
    async with browser as tab:
        yield tab
        # make sure there are no transactions left over (i.e. no unawaited requests)
        assert not tab.transactions

docBody = "<html><body><p>Hello, world</p></body></html>"
async def hello(request):
    return web.Response(text=docBody, content_type='text/html')

@pytest.fixture
async def server ():
    """ Simple HTTP server for testing notifications """
    app = web.Application()
    app.add_routes([web.get('/', hello)])
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()
    yield app
    await runner.cleanup ()

@pytest.mark.asyncio
async def test_tab_create (tab):
    """ Creating tabs works """
    assert isinstance (tab, Tab)
    version = await tab.Browser.getVersion ()
    assert 'protocolVersion' in version
    assert tab.pending == 0

@pytest.mark.asyncio
async def test_tab_close (browser):
    """ Tabs are closed after using them """
    async with browser as tab:
        tid = tab.id
    # give the browser some time to close the tab
    await asyncio.sleep (0.5)
    tabs = [t['id'] async for t in browser]
    assert tid not in tabs

@pytest.mark.asyncio
async def test_tab_notify_enable_disable (tab):
    """ Make sure enabling/disabling notifications works for all known
    namespaces """
    for name in ('Debugger', 'DOM', 'Log', 'Network', 'Page', 'Performance',
            'Profiler', 'Runtime', 'Security'):
        f = getattr (tab, name)
        await f.enable ()
        await f.disable ()

@pytest.mark.asyncio
async def test_tab_unknown_method (tab):
    with pytest.raises (MethodNotFound):
        await tab.Nonexistent.foobar ()

@pytest.mark.asyncio
async def test_tab_invalid_argument (tab):
    # should be string
    with pytest.raises (InvalidParameter):
        await tab.Page.captureScreenshot (format=123)

    with pytest.raises (InvalidParameter):
        await tab.Page.captureScreenshot (format=[123])

    with pytest.raises (InvalidParameter):
        await tab.Page.captureScreenshot (format={123: '456'})

@pytest.mark.asyncio
async def test_tab_crash (tab):
    with pytest.raises (Crashed):
        await tab.Page.crash ()

    # caling anything else now should fail as well
    with pytest.raises (Crashed):
        await tab.Browser.getVersion ()

@pytest.mark.asyncio
async def test_load (tab, server):
    await tab.Network.enable ()
    await tab.Page.navigate (url='http://localhost:8080')

    haveRequest = False
    haveResponse = False
    haveData = False
    haveFinished = False
    haveBody = False
    req = None
    resp = None
    while not haveBody:
        method, data = await tab.get ()

        # it can be either of those two in no specified order
        if method in (tab.Network.requestWillBeSent, tab.Network.requestWillBeSentExtraInfo) and not haveResponse:
            if req is None:
                req = data
            assert data['requestId'] == req['requestId']
            haveRequest = True
        elif method in (tab.Network.responseReceived, tab.Network.responseReceivedExtraInfo) and haveRequest:
            if resp is None:
                resp = data
            assert data['requestId'] == resp['requestId']
            haveResponse = True
        elif haveRequest and haveResponse and method == tab.Network.dataReceived:
            assert data['dataLength'] == len (docBody)
            assert data['requestId'] == req['requestId']
            haveData = True
        elif haveData:
            assert method == tab.Network.loadingFinished
            assert data['requestId'] == req['requestId']
            haveBody = True
        elif haveFinished:
            body = await tab.Network.getResponseBody (requestId=req['requestId'])
            assert body['body'] == docBody
            haveBody = True
        else:
            assert False, (method, req)

    await tab.Network.disable ()
    assert tab.pending == 0

@pytest.mark.asyncio
async def test_recv_failure(browser):
    """ Inject failure into receiver process and crash it """
    async with browser as tab:
        await tab.ws.close ()
        with pytest.raises (Crashed):
            await tab.Browser.getVersion ()

    async with browser as tab:
        await tab.ws.close ()
        with pytest.raises (Crashed):
            await tab.get ()

    async with browser as tab:
        handle = asyncio.ensure_future (tab.get ())
        await tab.ws.close ()
        with pytest.raises (Crashed):
            await handle

@pytest.mark.asyncio
async def test_tab_function (tab):
    assert tab.Network.enable.name == 'Network.enable'
    assert tab.Network.disable == tab.Network.disable
    assert tab.Network.enable != tab.Network.disable
    assert tab.Network != tab.Network.enable
    assert callable (tab.Network.enable)
    assert not callable (tab.Network.enable.name)
    assert 'Network.enable' in repr (tab.Network.enable)

@pytest.mark.asyncio
async def test_tab_function_hash (tab):
    d = {tab.Network.enable: 1, tab.Network.disable: 2, tab.Page: 3,
            tab.Page.enable: 4}
    assert len (d) == 4

@pytest.mark.asyncio
async def test_ws_ping(tab):
    """
    Chrome does not like websocket pings and closes the connection if it
    receives one. Not sure why.
    """
    with pytest.raises (Crashed):
        await tab.ws.ping ()
        await tab.Browser.getVersion ()

@pytest.mark.asyncio
async def test_passthrough ():
    """ Null service returns the url as is """

    url = 'http://localhost:12345'
    async with Passthrough (url) as u:
        assert str (u) == url