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.webdriver.support import expected_conditions as EC

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_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['files']])

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

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

        == set([file['hash_key'] for file in database_contents['files']])

    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']['hash_key'])

        async function try_add_item()

        {

                await finalize_transaction(context);

                return;

        returnval(try_add_item());

        sample_item, incomplete_files)

    database_contents = get_db_contents(execute_in_page)

    assert 'file not present' in exception

        keyfun = lambda item: item.get('hash_key') or item['identifier']

            == sorted(val,                             key=keyfun)

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

        '''{

        }''',

        sample_item, sample_files_by_hash)

             'bye.js']

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

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

    uses = dict([(uses['hash_key'], uses['uses'])

                 for uses in database_contents['file_uses']])

    assert uses  == dict([(file['hash_key'], nr)

                          for file, nr in zip(sample_files_list, uses_list)])

    files = dict([(file['hash_key'], file['contents'])

                  for file in database_contents['files']])

    assert files == dict([(file['hash_key'], file['contents'])

                          for file in sample_files_list])

    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]

    uses = dict([(uses['hash_key'], uses['uses'])

                 for uses in results[0]['file_uses']])

    assert uses  == dict([(file['hash_key'], 1) for file in sample_files_list])

    files = dict([(file['hash_key'], file['contents'])

                  for file in results[0]['files']])

    assert files == dict([(file['hash_key'], file['contents'])

                          for file in sample_files_list])

    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 = {

        'resources': {

            'helloapple': {

                '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',

            }

        },

        'mappings': {

            'helloapple': {

                '0.1.1': sample_mapping

            }

        },

        'files': sample_files_by_hash

    }

    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 get_db_contents(execute_in_page)['settings'] == []

    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)

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

    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])

    execute_in_page('''

    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 = {

        'resources': {

            'helloapple': {

                '1.0': sample_resource

            }

        },

        'mappings': {

            'helloapple': {

                '0.1.1': sample_mapping

            }

        },

        'files': sample_files_by_hash

    }

    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),

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

            ('blocking_file:///*', {'pattern': 'file:///*', 'allow': False})

    ]:

        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 = {

        'resources': {

            'helloapple-copy': {

                '1.0': sample_resource2

            }

        },

        'mappings': {

            'helloapple-copy': {

                '0.1.1': sample_mapping2

            }

        },

        'files': sample_files_by_hash

    }

    execute_in_page('returnval(set_allowed("ftp://a.bc/"));')

    driver.switch_to.window(windows[0])

    driver.implicitly_wait(10)

            ('mapping_helloapple-copy', sample_mapping2),

            ('repos_https://hydril2.la', {'url': 'https://hydril2.la'}),

            ('blocking_ftp://a.bc/', {'pattern': 'ftp://a.bc/', 'allow': True})

    ]:

        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/");

    def condition_items_absent_and_changed(driver):

            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])