servo/python/servo/devtools_tests.py

# Copyright 2013 The Servo Project Developers. See the COPYRIGHT
# file at the top-level directory of this distribution.
#
# Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
# http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
# <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
# option. This file may not be copied, modified, or distributed
# except according to those terms.

from concurrent.futures import Future
import logging
from geckordp.actors.root import RootActor
from geckordp.actors.descriptors.tab import TabActor
from geckordp.actors.watcher import WatcherActor
from geckordp.actors.resources import Resources
from geckordp.actors.events import Events
from geckordp.rdp_client import RDPClient
import http.server
import os.path
import socketserver
import subprocess
import time
from threading import Thread
from typing import Optional
import unittest


class DevtoolsTests(unittest.IsolatedAsyncioTestCase):
    # /path/to/servo/python/servo
    script_path = None

    def __init__(self, methodName="runTest"):
        super().__init__(methodName)
        self.servoshell = None
        self.base_url = None
        self.web_server = None
        self.web_server_thread = None

    def test_sources_list(self):
        self.run_servoshell(test_dir=os.path.join(DevtoolsTests.script_path, "devtools_tests/sources"))
        self.assert_sources_list([f"{self.base_url}/classic.js", f"{self.base_url}/test.html", "https://servo.org/js/load-table.js"])

    def test_sources_list_with_data_no_scripts(self):
        self.run_servoshell(url="data:text/html,")
        self.assert_sources_list([])

    def test_sources_list_with_data_empty_inline_script(self):
        self.run_servoshell(url="data:text/html,<script></script>")
        self.assert_sources_list([])

    def test_sources_list_with_data_inline_script(self):
        self.run_servoshell(url="data:text/html,<script>;</script>")
        self.assert_sources_list(["data:text/html,<script>;</script>"])

    def run_servoshell(self, *, test_dir=None, url=None):
        if test_dir is None:
            test_dir = os.path.join(DevtoolsTests.script_path, "devtools_tests")
        base_url = Future()

        class Handler(http.server.SimpleHTTPRequestHandler):
            def __init__(self, *args, **kwargs):
                super().__init__(*args, directory=test_dir, **kwargs)

            def log_message(self, format, *args):
                # Uncomment this to log requests.
                # return super().log_message(format, *args)
                pass

        def server_thread():
            self.web_server = socketserver.TCPServer(("0.0.0.0", 0), Handler)
            base_url.set_result(f"http://127.0.0.1:{self.web_server.server_address[1]}")
            self.web_server.serve_forever()

        # Start a web server for the test.
        self.web_server_thread = Thread(target=server_thread)
        self.web_server_thread.start()
        self.base_url = base_url.result(1)

        # Change this setting if you want to debug Servo.
        os.environ["RUST_LOG"] = "error,devtools=warn"

        # Run servoshell.
        if url is None:
            url = f"{self.base_url}/test.html"
        self.servoshell = subprocess.Popen(["target/release/servo", "--devtools=6080", url])

        # FIXME: Don’t do this
        time.sleep(1)

    def tearDown(self):
        # Terminate servoshell.
        self.servoshell.terminate()

        # Stop the web server.
        self.web_server.shutdown()
        self.web_server_thread.join()

    def assert_sources_list(self, expected_urls):
        client = RDPClient()
        client.connect("127.0.0.1", 6080)
        root = RootActor(client)
        tabs = root.list_tabs()
        tab_dict = tabs[0]
        tab = TabActor(client, tab_dict["actor"])
        watcher = tab.get_watcher()
        watcher = WatcherActor(client, watcher["actor"])

        target = Future()

        def on_target(data):
            if data["target"]["browsingContextID"] == tab_dict["browsingContextID"]:
                target.set_result(data["target"])

        client.add_event_listener(
            watcher.actor_id, Events.Watcher.TARGET_AVAILABLE_FORM, on_target,
        )
        watcher.watch_targets(WatcherActor.Targets.FRAME)

        done = Future()
        target = target.result(1)

        def on_source_resource(data):
            for [resource_type, sources] in data["array"]:
                try:
                    self.assertEqual(resource_type, "source")
                    self.assertEqual([source["url"] for source in sources], expected_urls)
                    done.set_result(None)
                except Exception as e:
                    # Raising here does nothing, for some reason.
                    # Send the exception back so it can be raised.
                    done.set_result(e)

        client.add_event_listener(
            target["actor"],
            Events.Watcher.RESOURCES_AVAILABLE_ARRAY,
            on_source_resource,
        )
        watcher.watch_resources([Resources.SOURCE])

        result: Optional[Exception] = done.result(1)
        if result:
            raise result
        client.disconnect()


def run_tests(script_path):
    DevtoolsTests.script_path = script_path
    verbosity = 1 if logging.getLogger().level >= logging.WARN else 2
    suite = unittest.TestLoader().loadTestsFromTestCase(DevtoolsTests)
    return unittest.TextTestRunner(verbosity=verbosity).run(suite).wasSuccessful()