Attachment 10917 Details for Bug 54452 – proposed pytest

proposed pytest

test_appcenter_cache_invalidate.py (text/x-python), 3.77 KB, created by Nikola Radovanovic on 2022-02-16 17:15:04 CET

(hide)

Description:

Filename:

MIME Type:

Creator: Nikola Radovanovic

Created: 2022-02-16 17:15:04 CET

Size: 3.77 KB

patch

obsolete

>import subprocess
>import threading
>from queue import Queue
>from string import Template
>from tempfile import NamedTemporaryFile
>
>import pytest
>from univention.appcenter.actions import get_action
>from univention.appcenter.app_cache import Apps
>
>
>def _get_all_apps():
>    return sorted(set(str(app) for app in Apps().get_every_single_app()))
>
>
>@pytest.fixture()
>def invalidate_cache(request):
>    # read UCS version this test is for
>    ucs_version = request.param
>
>    # perform update and store (new) applications in new_apps variable
>    get_action('update').call()
>    new_apps = _get_all_apps()
>
>    # use string template to create script that will invalidate app cache
>    script_content = Template(
>        '#!/bin/bash\n'
>        'set -e\n'
>        '\n'
>        'gzip -dk /usr/share/univention-appcenter/archives/appcenter.software-univention.de/$ucs_version/all.tar.gz\n'
>        'mv /usr/share/univention-appcenter/archives/appcenter.software-univention.de/$ucs_version/all.tar /var/cache/univention-appcenter/appcenter.software-univention.de/$ucs_version/.all.tar\n'
>        'touch /var/cache/univention-appcenter/appcenter.software-univention.de/$ucs_version/.all.tar\n'
>        'rm -rf /var/cache/univention-appcenter/appcenter.software-univention.de/$ucs_version/*\n'
>        'tar -C /var/cache/univention-appcenter/appcenter.software-univention.de/$ucs_version/ -xf /var/cache/univention-appcenter/appcenter.software-univention.de/$ucs_version/.all.tar\n'
>        'rm -rf /var/cache/univention-appcenter/appcenter.software-univention.de/$ucs_version/.etags\n'
>    )
>
>    with NamedTemporaryFile(prefix='test_appcenter_invalid_cache', delete=True) as f:
>        script_path = f.name
>
>        # populate invalidate app cache script and flush (save) it on disk
>        f.write(script_content.substitute(ucs_version=ucs_version).encode())
>        f.flush()
>
>        # run and wait invalidate cache script to finish. if it fails for any reason - raise an exception
>        process = subprocess.Popen(['sh', script_path], stderr=subprocess.PIPE, stdout=subprocess.PIPE)
>        process.wait()
>        out, err = process.communicate()
>        exit_code = process.returncode
>
>        if 0 != exit_code:
>            raise RuntimeError('Invalidating cache failed: %s' % err.decode())
>
>    # return list of new applications to the caller
>    yield new_apps
>
>
>class TestAppCacheReload(object):
>
>    @classmethod
>    def _read_apps(cls, result_queue):
>        result_queue.put(_get_all_apps())
>
>    @pytest.mark.parametrize('invalidate_cache', ['4.4'], indirect=['invalidate_cache'])
>    @pytest.mark.parametrize('readers_count', [2, 10, 20])
>    def test_single_writer_multiple_readers(self, invalidate_cache, readers_count):
>        # load new apps and invalidate app cache
>        new_apps = invalidate_cache
>
>        # create array of threads, but don't start them yet - in order to save some time and start faster later on
>        results_queue = Queue(readers_count)
>        readers = [threading.Thread(target=self._read_apps, args=(results_queue,)) for _ in range(readers_count)]
>
>        # create and start writer thread - it will rebuild the cache (since it's invalidated) using the update call
>        writer = threading.Thread(target=(lambda: get_action('update').call()))
>        writer.start()
>
>        # run readers
>        for reader in readers:
>            reader.start()
>
>        # wait for writer to finish and get all available apps
>        writer.join()
>        apps = _get_all_apps()
>
>        # wait for all readers to finish and collect errors if any
>        results = []
>        errors = 0
>        for reader in readers:
>            reader.join()
>            res = sorted(set(results_queue.get()))
>            results.extend(res)
>            if not (res == apps or res == new_apps):
>                errors += 1
>
>        assert 0 == errors

Actions: View

Attachments on bug 54452: 10917