Haketilo

# SPDX-License-Identifier: CC0-1.0

"""

Haketilo unit tests - IndexedDB access

"""

# This file is part of Haketilo

#

# Copyright (C) 2021,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.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

from .utils import *

# Sample resource definitions. They'd normally contain more fields but here we

# use simplified versions.

def make_sample_resource():

    return {

        'source_copyright': [

            sample_file_ref('report.spdx'),

            sample_file_ref('LICENSES/somelicense.txt')

        ],

        'type': 'resource',

        'identifier': 'helloapple',

        'scripts': [sample_file_ref('hello.js'), sample_file_ref('bye.js')]

    }

def make_sample_mapping():

    return {

        'source_copyright': [

            sample_file_ref('report.spdx'),

            sample_file_ref('README.md')

        ],

        'type': 'mapping',

        'identifier': 'helloapple'

    }

def mock_broadcast(execute_in_page):

    execute_in_page(

        '''{

        const broadcast_mock = {};

        const nop = () => {};

        for (const key in broadcast)

            broadcast_mock[key] = nop;

        broadcast = broadcast_mock;

        }''')

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

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.

    """

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

    mock_broadcast(execute_in_page)

    # Start with no database.

    clear_indexeddb(execute_in_page)

    sample_item = make_sample_resource()

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

    execute_in_page(

        '''{

        const promise = start_items_transaction(["resource"], arguments[1])

            .then(ctx => save_item(arguments[0], ctx).then(() => ctx))

            .then(finalize_transaction);

        returnval(promise);

        }''',

        sample_item, sample_files_by_hash)

    database_contents = get_db_contents(execute_in_page)

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

    assert all([sample_files_by_hash[file['hash_key']] == file['contents']

                for file in database_contents['files']])

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

    assert set([uses['hash_key'] for uses in database_contents['file_uses']]) \

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

    assert database_contents['mapping'] == []

    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.

    sample_item['scripts'].append(sample_file_ref('combined.js'))

    incomplete_files = {**sample_files_by_hash}

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

    exception = execute_in_page(

        '''{

        const args = arguments;

        async function try_add_item()

        {

            const context =

                await start_items_transaction(["resource"], args[1]);

            try {

                await save_item(args[0], context);

                await finalize_transaction(context);

                return;

            } catch(e) {

                return e;

            }

        }

        returnval(try_add_item());

        }''',

        sample_item, incomplete_files)

    previous_database_contents = database_contents

    database_contents = get_db_contents(execute_in_page)

    assert 'file not present' in exception

    for key, val in database_contents.items():

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

        assert sorted(previous_database_contents[key], key=keyfun) \

            == sorted(val,                             key=keyfun)

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

    sample_item = make_sample_mapping()

    database_contents = execute_in_page(

        '''{

        const promise = start_items_transaction(["mapping"], arguments[1])

            .then(ctx => save_item(arguments[0], ctx).then(() => ctx))

            .then(finalize_transaction);

        returnval(promise);

        }''',

        sample_item, sample_files_by_hash)

    database_contents = get_db_contents(execute_in_page)

    names = ['README.md', 'report.spdx', 'LICENSES/somelicense.txt', 'hello.js',

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

    del database_contents['resource'][0]['source_copyright'][0]['extra_prop']

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

         execute_in_page(

            f'''{{

            const remover = remove_{item_type};

            const promise =

                start_items_transaction(["{item_type}"], {{}})

                .then(ctx => remover('helloapple', ctx).then(() => ctx))

                .then(finalize_transaction);

            returnval(promise);

            }}''')

         results[i] = get_db_contents(execute_in_page)

    names = ['README.md', 'report.spdx']

    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]['resource'] == []

    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['resource'] == [sample_resource]

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

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

def test_haketilodb_settings(driver, execute_in_page):

    """

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

    Verify assigning/retrieving values of simple "settings" item works properly.

    """

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

    mock_broadcast(execute_in_page)

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

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

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.

    """

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

    mock_broadcast(execute_in_page)

    # 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.

    """

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

    mock_broadcast(execute_in_page)

    # 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());')

test_page_html = '''

<!DOCTYPE html>

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

<body>

</body>

'''

@pytest.mark.ext_data({

    'background_script': broker_js,

    'test_page':         test_page_html,

    'extra_files': {

        'testpage.js':   lambda: load_script('common/indexeddb.js')

    }

})

@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.

    """

    # Let's open the same extension's test page in a second window. Window 1

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

    # changes.

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

    windows = [*driver.window_handles]

    assert len(windows) == 2

    # Create elements that will have tracked data inserted under them.

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

    }

    ''')

    # Mock initial_data.

    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

    }

    driver.switch_to.window(windows[1])

    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.

    driver.switch_to.window(windows[0])

    execute_in_page(

        '''

        function update_item(store_name, change)

        {

            const elem_id = `${store_name}_${change.key}`;

            let elem = document.getElementById(elem_id);

            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;

        async function start_reporting()

        {

            const props = new Map(stores.map(([sn, opt]) => [sn, opt.keyPath]));

            for (const store_name of trackable) {

                [tracking, items] =

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

                const prop = props.get(store_name);

                for (const item of items)

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

            }

        }

        returnval(start_reporting());

        ''')

    item_counts = execute_in_page(

        '''{

        const childcount = id => document.getElementById(id).childElementCount;

        returnval(trackable.map(childcount));

        }''')

    assert item_counts == [1 for _ in item_counts]

    for elem_id, json_value in [

            ('resource_helloapple', sample_resource),

            ('mapping_helloapple', sample_mapping),

            ('settings_option15', {'name': 'option15', 'value': '123'}),

            ('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(save_items(arguments[0]));', sample_data)

    execute_in_page('returnval(set_setting("option22", "abc"));')

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

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

    driver.switch_to.window(windows[0])

    driver.implicitly_wait(10)

    for elem_id, json_value in [

            ('resource_helloapple-copy', sample_resource2),

            ('mapping_helloapple-copy', sample_mapping2),

            ('settings_option22', {'name': 'option22', 'value': 'abc'}),

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

    driver.implicitly_wait(0)

    # See if item deletions/modifications get tracked properly.

    driver.switch_to.window(windows[1])

    execute_in_page(

        '''{

        async function change_remove_items()

        {

            const store_names = ["resource", "mapping"];

            const ctx = await start_items_transaction(store_names, {});

            await remove_resource("helloapple", ctx);

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

            await finalize_transaction(ctx);

            await set_setting("option22", null);

            await del_repo("https://hydril.la");

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

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

        }

        returnval(change_remove_items());

        }''')

    removed_ids = ['mapping_helloapple-copy', 'resource_helloapple',

                   'repos_https://hydril.la', 'blocking_file:///*']

    def condition_items_absent_and_changed(driver):

        for id in removed_ids:

            try:

                driver.find_element_by_id(id)

                return False

            except WebDriverException:

                pass

        option_text = driver.find_element_by_id('settings_option22').text

        blocking_text = driver.find_element_by_id('blocking_ftp://a.bc/').text

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

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

    driver.switch_to.window(windows[0])

    WebDriverWait(driver, 10).until(condition_items_absent_and_changed)