Haketilo

# SPDX-License-Identifier: CC0-1.0

"""

Haketilo unit tests - repository querying

"""

# This file is part of Haketilo

#

# Copyright (C) 2022 Wojtek Kosior <koszko@koszko.org>

#

# This program is free software: you can redistribute it and/or modify

# it under the terms of the CC0 1.0 Universal License as published by

# the Creative Commons Corporation.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# CC0 1.0 Universal License for more details.

import pytest

import json

from selenium.webdriver.support.ui import WebDriverWait

from selenium.common.exceptions import ElementNotInteractableException

from ..extension_crafting import ExtraHTML

from ..script_loader import load_script

from .utils import *

def reload_with_target(driver, target):

    current_url = driver.execute_script('return location.href')

    driver.execute_script(

        '''

        window.location.href = arguments[0];

        window.location.reload();

        ''',

        f'{current_url}#{target}')

unprivileged_page_info = {

    'url': 'https://example_a.com/something',

    'allow': False

}

mapping_page_info = {

    **unprivileged_page_info,

    'mapping': 'm1',

    'payload': {'identifier': 'res1'}

}

mocked_page_infos = {

    'privileged': {

        'url': 'moz-extension://<some-id>/file.html',

        'privileged': True

    },

    'blocked_default': unprivileged_page_info,

    'allowed_default': {

        **unprivileged_page_info,

        'allow': True

    },

    'blocked_rule': {

        **unprivileged_page_info,

        'mapping': '~allow'

    },

    'allowed_rule': {

        **unprivileged_page_info,

        'allow': True,

        'mapping': '~allow'

    },

    'mapping': mapping_page_info,

    'error_deciding_policy': {

        **mapping_page_info,

        'error': {'haketilo_error_type': 'deciding_policy'}

    },

    'error_missing': {

        **mapping_page_info,

        'error': {'haketilo_error_type': 'missing', 'id': 'some-missing-res'}

    },

    'error_circular': {

        **mapping_page_info,

        'error': {'haketilo_error_type': 'circular', 'id': 'some-circular-res'}

    },

    'error_db': {

        **mapping_page_info,

        'error': {'haketilo_error_type': 'db'}

    },

    'error_other': {

        **mapping_page_info,

        'error': {'haketilo_error_type': 'other'}

    }

}

tab_mock_js = '''

;

const mocked_page_info = (%s)[/#mock_page_info-(.*)$/.exec(document.URL)[1]];

browser.tabs.sendMessage = async function(tab_id, msg) {

    const this_tab_id = (await browser.tabs.getCurrent()).id;

    if (tab_id !== this_tab_id)

        throw `not current tab id (${tab_id} instead of ${this_tab_id})`;

    if (msg[0] === "page_info") {

        return mocked_page_info;

    } else if (msg[0] === "repo_query") {

        const response = await fetch(msg[1]);

        if (!response)

            return {error: "Something happened :o"};

        const result = {ok: response.ok, status: response.status};

        try {

            result.json = await response.json();

        } catch(e) {

            result.error_json = "" + e;

        }

        return result;

    } else {

        throw `bad sendMessage message type: '${msg[0]}'`;

    }

}

const old_tabs_query = browser.tabs.query;

browser.tabs.query = async function(query) {

    const tabs = await old_tabs_query(query);

    tabs.forEach(t => t.url = mocked_page_info.url);

    return tabs;

}

''' % json.dumps(mocked_page_infos)

popup_ext_data = {

    'background_script': broker_js,

    'extra_html': ExtraHTML(

        'html/popup.html',

        {

            'common/browser.js':   tab_mock_js,

            'common/indexeddb.js': '; set_repo("https://hydril.la/");'

        },

        wrap_into_htmldoc=False

    ),

    'navigate_to': 'html/popup.html'

}

@pytest.mark.ext_data(popup_ext_data)

@pytest.mark.usefixtures('webextension')

@pytest.mark.parametrize('page_info_key', ['', *mocked_page_infos.keys()])

def test_popup_display(driver, execute_in_page, page_info_key):

    """

    Test popup viewing while on a page. Test parametrized with different

    possible values of page_info object passed in message from the content

    script.

    """

    reload_with_target(driver, f'mock_page_info-{page_info_key}')

    def get_nodes_by_id(driver):

        by_id = driver.execute_script(

            '''

            const nodes = [...document.querySelectorAll("[id]")];

            const reductor = (ob, node) => Object.assign(ob, {[node.id]: node});

            return nodes.reduce(reductor, {});

            ''');

        return by_id if by_id and 'repo_query_container' in by_id else None

    by_id = WebDriverWait(driver, 10).until(get_nodes_by_id)

    if page_info_key == '':

        error_msg = 'Page info not avaialable. Try reloading the page.'

        error_msg_shown = lambda d: by_id['loading_info'].text == error_msg

        WebDriverWait(driver, 10).until(error_msg_shown)

        return

    WebDriverWait(driver, 10).until(lambda d: by_id['info_form'].is_displayed())

    assert (page_info_key == 'privileged') == \

        by_id['privileged_page_info'].is_displayed()

    assert (page_info_key == 'privileged') ^ \

        by_id['unprivileged_page_info'].is_displayed()

    assert by_id['page_url'].text == mocked_page_infos[page_info_key]['url']

    assert not by_id['repo_query_container'].is_displayed()

    if 'allow' in page_info_key:

        assert by_id['scripts_blocked'].text.lower() == 'no'

    elif page_info_key != 'privileged':

        assert by_id['scripts_blocked'].text.lower() == 'yes'

    payload_text = by_id['injected_payload'].text

    if page_info_key == 'mapping':

        assert payload_text == 'res1'

    elif page_info_key == 'error_missing':

        assert payload_text == \

            "None (error: resource with id 'some-missing-res' missing from the database)"

    elif page_info_key == 'error_circular':

        assert payload_text == \

            "None (error: circular dependency of resource with id 'some-circular-res' on itself)"

    elif page_info_key == 'error_db':

        assert payload_text == \

            'None (error: failure reading Haketilo internal database)'

    elif page_info_key == 'error_other':

        assert payload_text == \

            'None (error: unknown failure occured)'

    elif page_info_key != 'privileged':

        assert payload_text == 'None'

    mapping_text = by_id['mapping_used'].text

    if page_info_key == 'error_deciding_policy':

        assert mapping_text == 'None (error occured when determining policy)'

    elif page_info_key == 'mapping' or page_info_key.startswith('error'):

        assert mapping_text == 'm1'

    if 'allowed' in page_info_key:

        assert 'None (scripts allowed by' in mapping_text

    elif 'blocked' in page_info_key:

        assert 'None (scripts blocked by' in mapping_text

    if 'rule' in page_info_key:

        assert 'by a rule)' in mapping_text

    elif 'default' in page_info_key:

        assert 'by default policy)' in mapping_text

@pytest.mark.ext_data(popup_ext_data)

@pytest.mark.usefixtures('webextension')

def test_popup_repo_query(driver, execute_in_page):

    """

    Test opening and closing the repo query view in popup.

    """

    reload_with_target(driver, f'mock_page_info-blocked_rule')

    driver.implicitly_wait(10)

    search_but = driver.find_element_by_id("search_resources_but")

    driver.implicitly_wait(0)

    # For unknown reasons waiting for search_but.is_displayed() to return True

    # does not guarantee the button will be interactable afterwards under newer

    # browsers. Hence, this workaround.

    def click_search_but(driver):

        try:

            search_but.click()

            return True

        except ElementNotInteractableException:

            pass

    WebDriverWait(driver, 10).until(click_search_but)

    containers = dict([(name, driver.find_element_by_id(f'{name}_container'))

                       for name in ('page_info', 'repo_query')])

    assert not containers['page_info'].is_displayed()

    assert containers['repo_query'].is_displayed()

    shown = lambda d: 'https://hydril.la/' in containers['repo_query'].text

    WebDriverWait(driver, 10).until(shown)

    # Click the "Show results" button.

    selector = '.repo_query_buttons > button:first-child'

    driver.find_element_by_css_selector(selector).click()

    shown = lambda d: 'MAPPING_A' in containers['repo_query'].text

    WebDriverWait(driver, 10).until(shown)

    # Click the "Cancel" button

    selector = '.repo_query_bottom_buttons > button'

    driver.find_element_by_css_selector(selector).click()

    assert containers['page_info'].is_displayed()

    assert not containers['repo_query'].is_displayed()

@pytest.mark.ext_data(popup_ext_data)

@pytest.mark.usefixtures('webextension')

def test_popup_settings_opening(driver, execute_in_page):

    """

    Test opening the settings page from popup through button click.

    """

    driver.find_element_by_id("settings_but").click()

    first_handle = driver.current_window_handle

    WebDriverWait(driver, 10).until(lambda d: len(d.window_handles) == 2)

    new_handle = [h for h in driver.window_handles if h != first_handle][0]

    driver.switch_to.window(new_handle)

    driver.implicitly_wait(10)

    assert "Extension's options page for testing" in \

        driver.find_element_by_tag_name("h1").text