# Copyright 2019 The Cobalt Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Connect to the web debugger and run some commands."""

# pylint: disable=g-doc-args,g-doc-return-or-yield,g-doc-exception

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import _env  # pylint: disable=unused-import,g-bad-import-order

import json
import logging
import os
import sys
import urlparse

from cobalt.black_box_tests import black_box_tests
from cobalt.black_box_tests.threaded_web_server import ThreadedWebServer

sys.path.append(
    os.path.join(
        os.path.dirname(__file__), '..', '..', '..',
        'third_party', 'websocket-client'))
import websocket  # pylint: disable=g-bad-import-order,g-import-not-at-top

# Set to True to add additional logging to debug the test.
_DEBUG = False


class DebuggerCommandError(Exception):
  """Exception when an error response is received for a command."""

  def __init__(self, error):
    code = '[{}] '.format(error['code']) if 'code' in error else ''
    super(DebuggerCommandError, self).__init__(code + error['message'])


class DebuggerConnection(object):
  """Connection to debugger over a WebSocket.

  Runs in the owner's thread by receiving and storing messages as needed when
  waiting for a command response or event.
  """

  def __init__(self, ws_url, timeout=3):
    self.ws_url = ws_url
    self.timeout = timeout
    self.last_id = 0
    self.commands = dict()
    self.responses = dict()
    self.events = list()

  def __enter__(self):
    websocket.enableTrace(_DEBUG)
    logging.debug('Debugger WebSocket: %s', self.ws_url)
    self.ws = websocket.create_connection(self.ws_url, timeout=self.timeout)
    return self

  def __exit__(self, exc_type, exc_value, exc_traceback):
    self.ws.close()

  def run_command(self, method, params=None):
    """Runs a debugger command and waits for the response.

    Fails the test with an exception if the debugger backend returns an error
    response, or the command times out without receiving a response.
    """
    response = self.wait_response(self.send_command(method, params))
    if 'error' in response:
      raise DebuggerCommandError(response['error'])
    return response

  def send_command(self, method, params=None):
    """Sends an async debugger command and returns the command id."""
    self.last_id += 1
    msg = {
        'id': self.last_id,
        'method': method,
    }
    if params:
      msg['params'] = params
    if _DEBUG:
      logging.debug('send >>>>>>>>\n%s\n>>>>>>>>',
                    json.dumps(msg, sort_keys=True, indent=4))
    logging.debug('send command: %s', method)
    self.ws.send(json.dumps(msg))
    self.commands[self.last_id] = msg
    return self.last_id

  def wait_response(self, command_id):
    """Wait for the response for a command to arrive.

    Fails the test with an exception if the WebSocket times out without
    receiving the response.
    """
    assert command_id in self.commands
    # Receive messages until the response we want arrives.
    # It may have already arrived before we even enter the loop.
    while command_id not in self.responses:
      try:
        self._receive_message()
      except websocket.WebSocketTimeoutException as err:
        # Show which command response we were waiting for that timed out.
        err.args = ('Command', self.commands[command_id])
        raise
    self.commands.pop(command_id)
    return self.responses.pop(command_id)

  def wait_event(self, method):
    """Wait for an event to arrive.

    Fails the test with an exception if the WebSocket times out without
    receiving the event.
    """
    # Check if we already received the event.
    for event in self.events:
      if event['method'] == method:
        self.events.remove(event)
        return event
    # Receive messages until the event we want arrives.
    while not self.events or self.events[-1]['method'] != method:
      try:
        self._receive_message()
      except websocket.WebSocketTimeoutException as err:
        # Show which event we were waiting for that timed out.
        err.args = ('Event', method)
        raise
    return self.events.pop()

  def _receive_message(self):
    """Receives one message and stores it in either responses or events."""
    msg = json.loads(self.ws.recv())
    if _DEBUG:
      logging.debug('recv <<<<<<<<\n%s\n<<<<<<<<',
                    json.dumps(msg, sort_keys=True, indent=4))
    if 'id' in msg:
      self.responses[msg['id']] = msg
    elif 'method' in msg:
      logging.debug('receive event: %s', msg['method'])
      self.events.append(msg)
    else:
      raise ValueError('Unexpected debugger message', msg)

  def enable_runtime(self):
    """Helper to enable the 'Runtime' domain and save the context_id."""
    self.run_command('Runtime.enable')
    context_event = self.wait_event('Runtime.executionContextCreated')
    self.context_id = context_event['params']['context']['id']
    assert self.context_id > 0

  def evaluate_js(self, expression):
    """Helper for the 'Runtime.evaluate' command to run some JavaScript."""
    return self.run_command('Runtime.evaluate', {
        'contextId': self.context_id,
        'expression': expression,
    })


class WebDebuggerTest(black_box_tests.BlackBoxTestCase):
  """Test interaction with the web debugger over a WebSocket."""

  def setUp(self):
    platform_vars = self.platform_config.GetVariables(self.device_params.config)
    if platform_vars['javascript_engine'] != 'v8':
      self.skipTest('DevTools requires V8')

    cobalt_vars = self.cobalt_config.GetVariables(self.device_params.config)
    if not cobalt_vars['enable_debugger']:
      self.skipTest('DevTools is disabled on this platform')

  def create_debugger_connection(self, runner):
    devtools_url = runner.GetCval('Cobalt.Server.DevTools')
    parts = list(urlparse.urlsplit(devtools_url))
    parts[0] = 'ws'  # scheme
    parts[2] = '/devtools/page/cobalt'  # path
    ws_url = urlparse.urlunsplit(parts)
    return DebuggerConnection(ws_url)

  def test_runtime(self):
    with ThreadedWebServer(binding_address=self.GetBindingAddress()) as server:
      url = server.GetURL(file_name='testdata/web_debugger.html')
      with self.CreateCobaltRunner(url=url) as runner:
        with self.create_debugger_connection(runner) as debugger:
          runner.WaitForJSTestsSetup()
          debugger.enable_runtime()

          # Evaluate a simple expression.
          eval_response = debugger.evaluate_js('6 * 7')
          self.assertEqual(42, eval_response['result']['result']['value'])

          # Set an attribute and read it back w/ WebDriver.
          debugger.evaluate_js(
              'document.body.setAttribute("web_debugger", "tested")')
          self.assertEqual(
              'tested',
              runner.UniqueFind('body').get_attribute('web_debugger'))

          # Log to console, and check we get the console event.
          debugger.evaluate_js('console.log("hello")')
          console_event = debugger.wait_event('Runtime.consoleAPICalled')
          self.assertEqual('hello', console_event['params']['args'][0]['value'])

          # End the test.
          debugger.evaluate_js('onEndTest()')
          self.assertTrue(runner.JSTestsSucceeded())

  def test_dom(self):
    with ThreadedWebServer(binding_address=self.GetBindingAddress()) as server:
      url = server.GetURL(file_name='testdata/web_debugger.html')
      with self.CreateCobaltRunner(url=url) as runner:
        with self.create_debugger_connection(runner) as debugger:
          runner.WaitForJSTestsSetup()
          debugger.enable_runtime()
          debugger.run_command('DOM.enable')

          doc_response = debugger.run_command('DOM.getDocument')
          doc_root = doc_response['result']['root']
          self.assertEqual('#document', doc_root['nodeName'])

          doc_url = doc_root['documentURL']
          # remove query params (cert_scope, etc.)
          doc_url = doc_url.split('?')[0]
          self.assertEqual(url, doc_url)

          # document: <html><head></head><body></body></html>
          html_node = doc_root['children'][0]
          body_node = html_node['children'][1]
          self.assertEqual('BODY', body_node['nodeName'])

          # body:
          #   <h1><span>Web debugger</span></h1>
          #   <div#test>
          #     <div#A><div#A1/><div#A2/></div#A>
          #     <div#B/>
          #   </div#test>
          debugger.run_command('DOM.requestChildNodes', {
              'nodeId': body_node['nodeId'],
              'depth': -1,  # entire subtree
          })
          child_nodes_event = debugger.wait_event('DOM.setChildNodes')

          h1 = child_nodes_event['params']['nodes'][0]
          span = h1['children'][0]
          text = span['children'][0]
          self.assertEqual('H1', h1['nodeName'])
          self.assertEqual('SPAN', span['nodeName'])
          self.assertEqual('#text', text['nodeName'])
          self.assertEqual('Web debugger', text['nodeValue'])

          test_div = child_nodes_event['params']['nodes'][1]
          child_a = test_div['children'][0]
          child_a1 = child_a['children'][0]
          child_a2 = child_a['children'][1]
          child_b = test_div['children'][1]
          self.assertEqual(2, test_div['childNodeCount'])
          self.assertEqual(2, child_a['childNodeCount'])
          self.assertEqual(0, child_b['childNodeCount'])
          self.assertEqual(['id', 'test'], test_div['attributes'])
          self.assertEqual(['id', 'A'], child_a['attributes'])
          self.assertEqual(['id', 'A1'], child_a1['attributes'])
          self.assertEqual(['id', 'A2'], child_a2['attributes'])
          self.assertEqual(['id', 'B'], child_b['attributes'])
          self.assertEqual([], child_b['children'])

          # Repeat, but only to depth 2 - not reporting children of A & B.
          debugger.run_command('DOM.requestChildNodes', {
              'nodeId': body_node['nodeId'],
              'depth': 2,
          })
          child_nodes_event = debugger.wait_event('DOM.setChildNodes')

          test_div = child_nodes_event['params']['nodes'][1]
          child_a = test_div['children'][0]
          child_b = test_div['children'][1]
          self.assertFalse('children' in child_a)
          self.assertFalse('children' in child_b)
          self.assertEqual(2, test_div['childNodeCount'])
          self.assertEqual(2, child_a['childNodeCount'])
          self.assertEqual(0, child_b['childNodeCount'])
          self.assertEqual(['id', 'test'], test_div['attributes'])
          self.assertEqual(['id', 'A'], child_a['attributes'])
          self.assertEqual(['id', 'B'], child_b['attributes'])

          # Repeat, to default depth of 1 - not reporting children of "#test".
          debugger.run_command('DOM.requestChildNodes', {
              'nodeId': body_node['nodeId'],
          })
          child_nodes_event = debugger.wait_event('DOM.setChildNodes')

          test_div = child_nodes_event['params']['nodes'][1]
          self.assertFalse('children' in test_div)
          self.assertEqual(2, test_div['childNodeCount'])
          self.assertEqual(['id', 'test'], test_div['attributes'])

          # Get the test div as a remote object, and request it as a node.
          # This sends a 'DOM.setChildNodes' event for each node up to the root.
          eval_result = debugger.evaluate_js('document.getElementById("test")')
          node_response = debugger.run_command('DOM.requestNode', {
              'objectId': eval_result['result']['result']['objectId'],
          })
          self.assertEqual(test_div['nodeId'],
                           node_response['result']['nodeId'])

          # Event reporting the requested <div#test>
          node_event = debugger.wait_event('DOM.setChildNodes')
          self.assertEqual(test_div['nodeId'],
                           node_event['params']['nodes'][0]['nodeId'])
          self.assertEqual(body_node['nodeId'],
                           node_event['params']['parentId'])

          # Event reporting the parent <body>
          node_event = debugger.wait_event('DOM.setChildNodes')
          self.assertEqual(body_node['nodeId'],
                           node_event['params']['nodes'][0]['nodeId'])
          self.assertEqual(html_node['nodeId'],
                           node_event['params']['parentId'])

          # Event reporting the parent <html>
          node_event = debugger.wait_event('DOM.setChildNodes')
          self.assertEqual(html_node['nodeId'],
                           node_event['params']['nodes'][0]['nodeId'])
          self.assertEqual(doc_root['nodeId'], node_event['params']['parentId'])

          # Round trip resolving test div to an object, then back to a node.
          resolve_response = debugger.run_command('DOM.resolveNode', {
              'nodeId': test_div['nodeId'],
          })
          node_response = debugger.run_command('DOM.requestNode', {
              'objectId': resolve_response['result']['object']['objectId'],
          })
          self.assertEqual(test_div['nodeId'],
                           node_response['result']['nodeId'])

          # Event reporting the requested <div#test>
          node_event = debugger.wait_event('DOM.setChildNodes')
          self.assertEqual(test_div['nodeId'],
                           node_event['params']['nodes'][0]['nodeId'])
          self.assertEqual(body_node['nodeId'],
                           node_event['params']['parentId'])
          # Ignore the other two events reporting the parents.
          node_event = debugger.wait_event('DOM.setChildNodes')
          node_event = debugger.wait_event('DOM.setChildNodes')

          # End the test.
          debugger.evaluate_js('onEndTest()')
          self.assertTrue(runner.JSTestsSucceeded())