Haketilo

# SPDX-License-Identifier: CC0-1.0

"""

Haketilo unit tests - IndexedDB access

"""

# This file is part of Haketilo

#

# 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

from selenium.webdriver.common.by import By

from selenium.webdriver.support.ui import WebDriverWait

from selenium.common.exceptions import WebDriverException

from ..script_loader import load_script

# use simplified versions.

def make_sample_resource():

    return {

        'source_copyright': [

            sample_file_ref('LICENSES/somelicense.txt')

        'type': 'resource',

        'identifier': 'helloapple',

def make_sample_mapping():

    return {

        'source_copyright': [

            sample_file_ref('README.md')

        'type': 'mapping',

        'identifier': 'helloapple'

    }

def test_haketilodb_update_1_0_1(driver, execute_in_page):

    """

    indexeddb.js modifies data when updating to database version 1.0.1. Verify

    the update procedure works properly.

    """

    execute_in_page(load_script('common/indexeddb.js'))

    execute_in_page('db_version_nr = version_nr([1, 0, 0]);')

    mock_broadcast(execute_in_page)

    # Start with no database.

    clear_indexeddb(execute_in_page)

    v1_url = 'https://hydrilla.koszko.org/api_v1/'

    v2_url = 'https://hydrilla.koszko.org/api_v2/'

    urls_v1 = {'https://sample.url/abc/', v1_url}

    urls_v2 = {'https://sample.url/abc/', v2_url}

    for url in urls_v1:

        execute_in_page('returnval(set_repo(arguments[0]));', url)

    assert urls_v1 == set(execute_in_page('returnval(get_repos());'))

    # Verify that url gets updated to v2 upon database update.

    driver.get(driver.execute_script('return window.location.href;'))

    execute_in_page(load_script('common/indexeddb.js'))

    mock_broadcast(execute_in_page)

    database_contents = get_db_contents(execute_in_page)

    assert set(execute_in_page('returnval(get_repos());')) == urls_v2

    # Verify that url does not get updated when there is no database update.

    execute_in_page('returnval(del_repo(arguments[0]));', v2_url)

    execute_in_page('returnval(set_repo(arguments[0]));', v1_url)

    driver.get(driver.execute_script('return window.location.href;'))

    execute_in_page(load_script('common/indexeddb.js'))

    mock_broadcast(execute_in_page)

    database_contents = get_db_contents(execute_in_page)

    assert set(execute_in_page('returnval(get_repos());')) == urls_v1

def test_haketilodb_item_modifications(driver, execute_in_page):

    """

    indexeddb.js facilitates operating on Haketilo's internal database.

    Verify database operations on mappings/resources work properly.

    """

    # Start with no database.

    clear_indexeddb(execute_in_page)

    sample_item['source_copyright'][0]['extra_prop'] = True

    database_contents = get_db_contents(execute_in_page)

    assert all([len(file) == 2 for file in database_contents['file']])

    assert len(database_contents['file_uses']) == 4

    assert all([uses['uses'] == 1 for uses in database_contents['file_uses']])

    assert database_contents['resource'] == [sample_item]

    # See if trying to add an item without providing all its files ends in an

    # exception and aborts the transaction as it should.

    incomplete_files.pop(sample_files['combined.js']['sha256'])

        async function try_add_item()

        {

                await finalize_transaction(context);

                return;

        returnval(try_add_item());

    database_contents = get_db_contents(execute_in_page)

    assert 'file not present' in exception

            == sorted(val,                             key=keyfun)

    # See if adding another item that partially uses first's files works OK.

        '''{

        }''',

             'bye.js']

    sample_files_list = [sample_files[name] for name in names]

    uses_list = [1, 2, 1, 1, 1]

    assert database_contents['resource'] == [make_sample_resource()]

    assert database_contents['mapping']  == [sample_item]

    # Try removing the items to get an empty database again.

    results = [None, None]

    for i, item_type in enumerate(['resource', 'mapping']):

            }}''')

    sample_files_list = [sample_files[name] for name in names]

    uses_list = [1, 1]

    assert results[0]['mapping'] == [sample_item]

    assert results[1] == dict([(key, []) for key in  results[0].keys()])

    # Try initializing an empty database with sample initial data object.

    sample_resource = make_sample_resource()

    sample_mapping = make_sample_mapping()

    initial_data = {

                '1.12':   sample_resource,

                '0.9':    'something_that_should_get_ignored',

                '1':      'something_that_should_get_ignored',

                '1.1':    'something_that_should_get_ignored',

                '1.11.1': 'something_that_should_get_ignored',

            }

        },

                '0.1.1': sample_mapping

            }

        },

            'sha256': sample_files_by_sha256

        }

    clear_indexeddb(execute_in_page)

    execute_in_page('initial_data = arguments[0];', initial_data)

    database_contents = get_db_contents(execute_in_page)

    assert database_contents['mapping']  == [sample_mapping]

def test_haketilodb_settings(driver, execute_in_page):

    """

    indexeddb.js facilitates operating on Haketilo's internal database.

    # Start with no database.

    clear_indexeddb(execute_in_page)

    assert execute_in_page('returnval(get_setting("option15"));') == None

    execute_in_page('returnval(set_setting("option15", "disable"));')

    assert execute_in_page('returnval(get_setting("option15"));') == 'disable'

    execute_in_page('returnval(set_setting("option15", "enable"));')

    assert execute_in_page('returnval(get_setting("option15"));') == 'enable'

def test_haketilodb_allowing(driver, execute_in_page):

    """

    indexeddb.js facilitates operating on Haketilo's internal database.

    Verify changing the "blocking" configuration for a URL works properly.

    """

    # Start with no database.

    clear_indexeddb(execute_in_page)

    assert get_db_contents(execute_in_page)['blocking'] == []

    def run_with_sample_url(expr):

        return execute_in_page(f'returnval({expr});', 'https://example.com/**')

    assert None == run_with_sample_url('get_allowing(arguments[0])')

    run_with_sample_url('set_disallowed(arguments[0])')

    assert False == run_with_sample_url('get_allowing(arguments[0])')

    run_with_sample_url('set_allowed(arguments[0])')

    assert True == run_with_sample_url('get_allowing(arguments[0])')

    run_with_sample_url('set_default_allowing(arguments[0])')

    assert None == run_with_sample_url('get_allowing(arguments[0])')

@pytest.mark.get_page('https://gotmyowndoma.in')

def test_haketilodb_repos(driver, execute_in_page):

    """

    indexeddb.js facilitates operating on Haketilo's internal database.

    Verify operations on repositories list work properly.

    """

    # Start with no database.

    clear_indexeddb(execute_in_page)

    sample_urls = ['https://hdrlla.example.com/', 'https://hdrlla.example.org']

    assert [] == execute_in_page('returnval(get_repos());')

    execute_in_page('returnval(set_repo(arguments[0]));', sample_urls[0])

    assert [sample_urls[0]] == execute_in_page('returnval(get_repos());')

    execute_in_page('returnval(set_repo(arguments[0]));', sample_urls[1])

    assert set(sample_urls) == set(execute_in_page('returnval(get_repos());'))

    execute_in_page('returnval(del_repo(arguments[0]));', sample_urls[0])

    assert [sample_urls[1]] == execute_in_page('returnval(get_repos());')

<!DOCTYPE html>

<script src="/testpage.js"></script>

</body>

@pytest.mark.ext_data({

    'background_script': broker_js,

    'test_page':         test_page_html,

    'extra_files': {

})

@pytest.mark.usefixtures('webextension')

def test_haketilodb_track(driver, execute_in_page, wait_elem_text):

    """

    Verify IndexedDB object change notifications are correctly broadcasted

    through extension's background script and allow for object store contents

    to be tracked in any execution context.

    """

    # will be used to make changes to IndexedDB and window 0 to "track" those

    driver.execute_script('window.open(window.location.href, "_blank");')

    driver.switch_to.window(windows[0])

        '''

        for (const store_name of trackable) {

            const h2 = document.createElement("h2");

            h2.innerText = store_name;

            document.body.append(h2);

            const ul = document.createElement("ul");

            ul.id = store_name;

            document.body.append(ul);

        }

        ''')

    sample_resource = make_sample_resource()

    sample_mapping = make_sample_mapping()

    initial_data = {

                '1.0': sample_resource

            }

        },

                '0.1.1': sample_mapping

            }

        },

            'sha256': sample_files_by_sha256

        }

    execute_in_page('initial_data = arguments[0];', initial_data)

    execute_in_page('returnval(set_setting("option15", "123"));')

    execute_in_page('returnval(set_repo("https://hydril.la"));')

    execute_in_page('returnval(set_disallowed("file:///*"));')

    # See if track.*() functions properly return the already-existing items.

        '''

        function update_item(store_name, change)

        {

            elem = elem || document.createElement("li");

            elem.id = elem_id;

            elem.innerText = JSON.stringify(change.new_val);

            document.getElementById(store_name).append(elem);

            if (change.new_val === undefined)

                elem.remove();

        }

        let resource_tracking, resource_items, mapping_tracking, mapping_items;

            for (const store_name of trackable) {

                    await track[store_name](ch => update_item(store_name, ch));

                    update_item(store_name, {key: item[prop], new_val: item});

            }

        '''{

        }''')

    assert item_counts == [1 for _ in item_counts]

    for elem_id, json_value in [

            ('mapping_helloapple', sample_mapping),

            ('repo_https://hydril.la', {'url': 'https://hydril.la'}),

    ]:

        assert json.loads(driver.find_element_by_id(elem_id).text) == json_value

    # See if item additions get tracked properly.

    driver.switch_to.window(windows[1])

    sample_resource2 = make_sample_resource()

    sample_resource2['identifier'] = 'helloapple-copy'

    sample_mapping2 = make_sample_mapping()

    sample_mapping2['identifier'] = 'helloapple-copy'

    sample_data = {

                '1.0': sample_resource2

            }

        },

                '0.1.1': sample_mapping2

            }

        },

            'sha256': sample_files_by_sha256

        'repo': [

            'https://hydril2.la/'

        ]

    driver.switch_to.window(windows[0])

    driver.implicitly_wait(10)

            ('mapping_helloapple-copy', sample_mapping2),

            ('repo_https://hydril3.la/', {'url': 'https://hydril3.la/'}),

    ]:

        assert json.loads(driver.find_element_by_id(elem_id).text) == json_value

    execute_in_page(

        '''{

            await remove_resource("helloapple", ctx);

            await remove_mapping("helloapple-copy", ctx);

            await set_setting("option22", null);

            await set_default_allowing("file:///*");

            await set_disallowed("ftp://a.bc/");

            try:

                driver.find_element_by_id(id)

                return False

            except WebDriverException:

                pass

        return (json.loads(option_text)['value'] == None and

                json.loads(blocking_text)['allow'] == False)

    driver.switch_to.window(windows[0])