mirror of
https://github.com/YunoHost/yunohost.git
synced 2024-09-03 20:06:10 +02:00
Rewrite appslist system
This commit is contained in:
parent
a6607eaf30
commit
65b81e8677
3 changed files with 396 additions and 316 deletions
|
@ -51,6 +51,11 @@
|
|||
"app_upgraded": "{app:s} has been upgraded",
|
||||
"apps_permission_not_found": "No permission found for the installed apps",
|
||||
"apps_permission_restoration_failed": "Permission '{permission:s}' for app {app:s} restoration has failed",
|
||||
"appslist_init_success": "Appslist system initialized!",
|
||||
"appslist_updating": "Updating application list...",
|
||||
"appslist_failed_to_download": "Unable to download the {applist} appslist : {error}",
|
||||
"appslist_obsolete_cache": "The applist cache is empty or obsolete.",
|
||||
"appslist_update_success": "The application list has been updated!",
|
||||
"ask_current_admin_password": "Current administration password",
|
||||
"ask_email": "Email address",
|
||||
"ask_firstname": "First name",
|
||||
|
|
|
@ -39,7 +39,8 @@ from datetime import datetime
|
|||
|
||||
from moulinette import msignals, m18n, msettings
|
||||
from moulinette.utils.log import getActionLogger
|
||||
from moulinette.utils.filesystem import read_json, read_toml
|
||||
from moulinette.utils.network import download_json
|
||||
from moulinette.utils.filesystem import read_json, read_toml, read_yaml, write_to_file, write_to_json, write_to_yaml, chmod, chown, mkdir
|
||||
|
||||
from yunohost.service import service_log, service_status, _run_service_command
|
||||
from yunohost.utils import packages
|
||||
|
@ -48,12 +49,16 @@ from yunohost.log import is_unit_operation, OperationLogger
|
|||
|
||||
logger = getActionLogger('yunohost.app')
|
||||
|
||||
REPO_PATH = '/var/cache/yunohost/repo'
|
||||
APPS_PATH = '/usr/share/yunohost/apps'
|
||||
APPS_SETTING_PATH = '/etc/yunohost/apps/'
|
||||
INSTALL_TMP = '/var/cache/yunohost'
|
||||
APP_TMP_FOLDER = INSTALL_TMP + '/from_file'
|
||||
APPSLISTS_JSON = '/etc/yunohost/appslists.json'
|
||||
|
||||
APPSLISTS_CACHE = '/var/cache/yunohost/repo'
|
||||
APPSLISTS_CONF = '/etc/yunohost/appslists.yml'
|
||||
APPSLISTS_CRON_PATH = "/etc/cron.daily/yunohost-fetch-appslists"
|
||||
APPSLISTS_API_VERSION = 1
|
||||
APPSLISTS_DEFAULT_URL = "https://app.yunohost.org/default"
|
||||
|
||||
re_github_repo = re.compile(
|
||||
r'^(http[s]?://|git@)github.com[/:]'
|
||||
|
@ -2542,6 +2547,155 @@ def _parse_app_instance_name(app_instance_name):
|
|||
return (appid, app_instance_nb)
|
||||
|
||||
|
||||
#
|
||||
# ############################### #
|
||||
# Applications list management #
|
||||
# ############################### #
|
||||
#
|
||||
|
||||
|
||||
def _initialize_appslists_system():
|
||||
"""
|
||||
This function is meant to intialize the appslist system with YunoHost's default applist.
|
||||
|
||||
It also creates the cron job that will update the list every day
|
||||
"""
|
||||
|
||||
default_appslist_list = [{"id": "default", "url": APPSLISTS_DEFAULT_URL}]
|
||||
|
||||
cron_job = []
|
||||
cron_job.append("#!/bin/bash")
|
||||
# We add a random delay between 0 and 60 min to avoid every instance fetching
|
||||
# the appslist at the same time every night
|
||||
cron_job.append("(sleep $((RANDOM%3600));")
|
||||
cron_job.append("yunohost tools update --apps > /dev/null) &")
|
||||
try:
|
||||
logger.debug("Initializing appslist system with YunoHost's default app list")
|
||||
write_to_yaml(APPSLISTS_CONF, default_appslist_list)
|
||||
|
||||
logger.debug("Installing appslist fetch daily cron job")
|
||||
write_to_file(APPSLISTS_CRON_PATH, '\n'.join(cron_job))
|
||||
chown(APPSLISTS_CRON_PATH, uid="root", gid="root")
|
||||
chmod(APPSLISTS_CRON_PATH, 0o755)
|
||||
except Exception as e:
|
||||
raise YunohostError("Could not initialize the appslist system... : %s" % str(e))
|
||||
|
||||
logger.success(m18n.n("appslist_init_success"))
|
||||
|
||||
|
||||
def _read_appslist_list():
|
||||
"""
|
||||
Read the json corresponding to the list of appslists
|
||||
"""
|
||||
|
||||
try:
|
||||
list_ = read_yaml(APPSLISTS_CONF)
|
||||
# Support the case where file exists but is empty
|
||||
# by returning [] if list_ is None
|
||||
return list_ if list_ else []
|
||||
except Exception as e:
|
||||
raise YunohostError("Could not read the appslist list ... : %s" % str(e))
|
||||
|
||||
|
||||
def _actual_appslist_api_url(base_url):
|
||||
|
||||
return "{base_url}/v{version}/apps.json".format(base_url=base_url, version=APPSLISTS_API_VERSION)
|
||||
|
||||
|
||||
def _update_appslist():
|
||||
"""
|
||||
Fetches the json for each appslist and update the cache
|
||||
|
||||
appslist_list is for example :
|
||||
[ {"id": "default", "url": "https://app.yunohost.org/default/"} ]
|
||||
|
||||
Then for each appslist, the actual json URL to be fetched is like :
|
||||
https://app.yunohost.org/default/vX/apps.json
|
||||
|
||||
And store it in :
|
||||
/var/cache/yunohost/repo/default.json
|
||||
"""
|
||||
|
||||
appslist_list = _read_appslist_list()
|
||||
|
||||
logger.info(m18n.n("appslist_updating"))
|
||||
|
||||
# Create cache folder if needed
|
||||
if not os.path.exists(APPSLISTS_CACHE):
|
||||
logger.debug("Initialize folder for appslist cache")
|
||||
mkdir(APPSLISTS_CACHE, mode=0o750, parents=True, uid='root')
|
||||
|
||||
for appslist in appslist_list:
|
||||
applist_id = appslist["id"]
|
||||
actual_api_url = _actual_appslist_api_url(appslist["url"])
|
||||
|
||||
# Fetch the json
|
||||
try:
|
||||
appslist_content = download_json(actual_api_url)
|
||||
except Exception as e:
|
||||
raise YunohostError("appslist_failed_to_download", applist=applist_id, error=str(e))
|
||||
|
||||
# Remember the appslist api version for later
|
||||
appslist_content["from_api_version"] = APPSLISTS_API_VERSION
|
||||
|
||||
# Save the appslist data in the cache
|
||||
cache_file = "{cache_folder}/{list}.json".format(cache_folder=APPSLISTS_CACHE, list=applist_id)
|
||||
try:
|
||||
write_to_json(cache_file, appslist_content)
|
||||
except Exception as e:
|
||||
raise YunohostError("Unable to write cache data for %s appslist : %s" % (applist_id, str(e)))
|
||||
|
||||
logger.success(m18n.n("appslist_update_success"))
|
||||
|
||||
|
||||
def _load_appslist():
|
||||
"""
|
||||
Read all the appslist cache file and build a single dict (app_dict)
|
||||
corresponding to all known apps in all indexes
|
||||
"""
|
||||
|
||||
app_dict = {}
|
||||
|
||||
for appslist_id in [L["id"] for L in _read_appslist_list()]:
|
||||
|
||||
# Let's load the json from cache for this appslist
|
||||
cache_file = "{cache_folder}/{list}.json".format(cache_folder=APPSLISTS_CACHE, list=appslist_id)
|
||||
|
||||
try:
|
||||
appslist_content = read_json(cache_file) if os.path.exists(cache_file) else None
|
||||
except Exception as e:
|
||||
raise ("Unable to read cache for appslist %s : %s" % (appslist_id, str(e)))
|
||||
|
||||
# Check that the version of the data matches version ....
|
||||
# ... otherwise it means we updated yunohost in the meantime
|
||||
# and need to update the cache for everything to be consistent
|
||||
if not appslist_content or appslist_content.get("from_api_version") != APPSLISTS_API_VERSION:
|
||||
logger.info(m18n.n("appslist_obsolete_cache"))
|
||||
_update_appslist()
|
||||
appslist_content = read_json(cache_file)
|
||||
|
||||
del appslist_content["from_api_version"]
|
||||
|
||||
# Add apps from this applist to the output
|
||||
for app, info in appslist_content.items():
|
||||
|
||||
# (N.B. : there's a small edge case where multiple appslist could be listing the same apps ...
|
||||
# in which case we keep only the first one found)
|
||||
if app in app_dict:
|
||||
logger.warning("Duplicate app %s found between appslist %s and %s" % (app, appslist_id, app_dict[app]['repository']))
|
||||
continue
|
||||
|
||||
info['repository'] = appslist_id
|
||||
app_dict[app] = info
|
||||
|
||||
return app_dict
|
||||
|
||||
#
|
||||
# ############################### #
|
||||
# Small utilities #
|
||||
# ############################### #
|
||||
#
|
||||
|
||||
def is_true(arg):
|
||||
"""
|
||||
Convert a string into a boolean
|
||||
|
|
|
@ -3,34 +3,49 @@ import pytest
|
|||
import requests
|
||||
import requests_mock
|
||||
import glob
|
||||
import time
|
||||
import shutil
|
||||
|
||||
from moulinette import m18n
|
||||
from moulinette.utils.filesystem import read_json, write_to_json, write_to_yaml
|
||||
|
||||
from yunohost.utils.error import YunohostError
|
||||
from yunohost.app import (_initialize_appslists_system,
|
||||
_read_appslist_list,
|
||||
_update_appslist,
|
||||
_actual_appslist_api_url,
|
||||
_load_appslist,
|
||||
logger,
|
||||
APPSLISTS_CACHE,
|
||||
APPSLISTS_CONF,
|
||||
APPSLISTS_CRON_PATH,
|
||||
APPSLISTS_API_VERSION,
|
||||
APPSLISTS_DEFAULT_URL)
|
||||
|
||||
from yunohost.app import app_fetchlist, app_removelist, app_listlists, _using_legacy_appslist_system, _migrate_appslist_system, _register_new_appslist
|
||||
APPSLISTS_DEFAULT_URL_FULL = _actual_appslist_api_url(APPSLISTS_DEFAULT_URL)
|
||||
CRON_FOLDER, CRON_NAME = APPSLISTS_CRON_PATH.rsplit("/", 1)
|
||||
|
||||
URL_OFFICIAL_APP_LIST = "https://app.yunohost.org/official.json"
|
||||
REPO_PATH = '/var/cache/yunohost/repo'
|
||||
APPSLISTS_JSON = '/etc/yunohost/appslists.json'
|
||||
DUMMY_APPLIST = """{
|
||||
"foo": {"id": "foo", "level": 4},
|
||||
"bar": {"id": "bar", "level": 7}
|
||||
}
|
||||
"""
|
||||
|
||||
class AnyStringWith(str):
|
||||
def __eq__(self, other):
|
||||
return self in other
|
||||
|
||||
def setup_function(function):
|
||||
|
||||
# Clear all appslist
|
||||
files = glob.glob(REPO_PATH + "/*")
|
||||
for f in files:
|
||||
os.remove(f)
|
||||
# Clear applist cache
|
||||
shutil.rmtree(APPSLISTS_CACHE, ignore_errors=True)
|
||||
|
||||
# Clear appslist crons
|
||||
files = glob.glob("/etc/cron.d/yunohost-applist-*")
|
||||
for f in files:
|
||||
os.remove(f)
|
||||
# Clear appslist cron
|
||||
if os.path.exists(APPSLISTS_CRON_PATH):
|
||||
os.remove(APPSLISTS_CRON_PATH)
|
||||
|
||||
if os.path.exists("/etc/cron.daily/yunohost-fetch-appslists"):
|
||||
os.remove("/etc/cron.daily/yunohost-fetch-appslists")
|
||||
|
||||
if os.path.exists(APPSLISTS_JSON):
|
||||
os.remove(APPSLISTS_JSON)
|
||||
# Clear appslist conf
|
||||
if os.path.exists(APPSLISTS_CONF):
|
||||
os.remove(APPSLISTS_CONF)
|
||||
|
||||
|
||||
def teardown_function(function):
|
||||
|
@ -38,352 +53,258 @@ def teardown_function(function):
|
|||
|
||||
|
||||
def cron_job_is_there():
|
||||
r = os.system("run-parts -v --test /etc/cron.daily/ | grep yunohost-fetch-appslists")
|
||||
r = os.system("run-parts -v --test %s | grep %s" % (CRON_FOLDER, CRON_NAME))
|
||||
return r == 0
|
||||
|
||||
|
||||
#
|
||||
# Test listing of appslists and registering of appslists #
|
||||
# ################################################
|
||||
#
|
||||
|
||||
|
||||
def test_appslist_list_empty():
|
||||
"""
|
||||
Calling app_listlists() with no registered list should return empty dict
|
||||
"""
|
||||
def test_appslist_init(mocker):
|
||||
|
||||
assert app_listlists() == {}
|
||||
# Cache is empty
|
||||
assert not glob.glob(APPSLISTS_CACHE + "/*")
|
||||
# Conf doesn't exist yet
|
||||
assert not os.path.exists(APPSLISTS_CONF)
|
||||
# Conf doesn't exist yet
|
||||
assert not os.path.exists(APPSLISTS_CRON_PATH)
|
||||
|
||||
# Initialize ...
|
||||
mocker.spy(m18n, "n")
|
||||
_initialize_appslists_system()
|
||||
m18n.n.assert_any_call('appslist_init_success')
|
||||
|
||||
def test_appslist_list_register():
|
||||
"""
|
||||
Register a new list
|
||||
"""
|
||||
|
||||
# Assume we're starting with an empty app list
|
||||
assert app_listlists() == {}
|
||||
|
||||
# Register a new dummy list
|
||||
_register_new_appslist("https://lol.com/appslist.json", "dummy")
|
||||
|
||||
appslist_dict = app_listlists()
|
||||
assert "dummy" in appslist_dict.keys()
|
||||
assert appslist_dict["dummy"]["url"] == "https://lol.com/appslist.json"
|
||||
|
||||
# Then there's a cron enabled
|
||||
assert cron_job_is_there()
|
||||
|
||||
# And a conf with at least one list
|
||||
assert os.path.exists(APPSLISTS_CONF)
|
||||
appslist_list = _read_appslist_list()
|
||||
assert len(appslist_list)
|
||||
|
||||
def test_appslist_list_register_conflict_name():
|
||||
"""
|
||||
Attempt to register a new list with conflicting name
|
||||
"""
|
||||
|
||||
_register_new_appslist("https://lol.com/appslist.json", "dummy")
|
||||
with pytest.raises(YunohostError):
|
||||
_register_new_appslist("https://lol.com/appslist2.json", "dummy")
|
||||
|
||||
appslist_dict = app_listlists()
|
||||
|
||||
assert "dummy" in appslist_dict.keys()
|
||||
assert "dummy2" not in appslist_dict.keys()
|
||||
# Cache is expected to still be empty though
|
||||
# (if we did update the appslist during init,
|
||||
# we couldn't differentiate easily exceptions
|
||||
# related to lack of network connectivity)
|
||||
assert not glob.glob(APPSLISTS_CACHE + "/*")
|
||||
|
||||
|
||||
def test_appslist_list_register_conflict_url():
|
||||
"""
|
||||
Attempt to register a new list with conflicting url
|
||||
"""
|
||||
def test_appslist_emptylist():
|
||||
|
||||
_register_new_appslist("https://lol.com/appslist.json", "dummy")
|
||||
with pytest.raises(YunohostError):
|
||||
_register_new_appslist("https://lol.com/appslist.json", "plopette")
|
||||
# Initialize ...
|
||||
_initialize_appslists_system()
|
||||
|
||||
appslist_dict = app_listlists()
|
||||
# Let's imagine somebody removed the default applist because uh idk they dont want to use our default applist
|
||||
os.system("rm %s" % APPSLISTS_CONF)
|
||||
os.system("touch %s" % APPSLISTS_CONF)
|
||||
|
||||
assert "dummy" in appslist_dict.keys()
|
||||
assert "plopette" not in appslist_dict.keys()
|
||||
appslist_list = _read_appslist_list()
|
||||
assert not len(appslist_list)
|
||||
|
||||
|
||||
#
|
||||
# Test fetching of appslists #
|
||||
#
|
||||
def test_appslist_update_success(mocker):
|
||||
|
||||
# Initialize ...
|
||||
_initialize_appslists_system()
|
||||
|
||||
# Cache is empty
|
||||
assert not glob.glob(APPSLISTS_CACHE + "/*")
|
||||
|
||||
# Update
|
||||
with requests_mock.Mocker() as m:
|
||||
|
||||
_actual_appslist_api_url,
|
||||
# Mock the server response with a dummy applist
|
||||
m.register_uri("GET", APPSLISTS_DEFAULT_URL_FULL, text=DUMMY_APPLIST)
|
||||
|
||||
mocker.spy(m18n, "n")
|
||||
_update_appslist()
|
||||
m18n.n.assert_any_call("appslist_updating")
|
||||
m18n.n.assert_any_call("appslist_update_success")
|
||||
|
||||
# Cache shouldn't be empty anymore empty
|
||||
assert glob.glob(APPSLISTS_CACHE + "/*")
|
||||
|
||||
app_dict = _load_appslist()
|
||||
assert "foo" in app_dict.keys()
|
||||
assert "bar" in app_dict.keys()
|
||||
|
||||
|
||||
def test_appslist_fetch():
|
||||
"""
|
||||
Do a fetchlist and test the .json got updated.
|
||||
"""
|
||||
assert app_listlists() == {}
|
||||
def test_appslist_update_404(mocker):
|
||||
|
||||
_register_new_appslist(URL_OFFICIAL_APP_LIST, "yunohost")
|
||||
# Initialize ...
|
||||
_initialize_appslists_system()
|
||||
|
||||
with requests_mock.Mocker() as m:
|
||||
|
||||
# Mock the server response with a valid (well, empty, yep) json
|
||||
m.register_uri("GET", URL_OFFICIAL_APP_LIST, text='{ }')
|
||||
# 404 error
|
||||
m.register_uri("GET", APPSLISTS_DEFAULT_URL_FULL,
|
||||
status_code=404)
|
||||
|
||||
official_lastUpdate = app_listlists()["yunohost"]["lastUpdate"]
|
||||
app_fetchlist()
|
||||
new_official_lastUpdate = app_listlists()["yunohost"]["lastUpdate"]
|
||||
with pytest.raises(YunohostError):
|
||||
mocker.spy(m18n, "n")
|
||||
_update_appslist()
|
||||
m18n.n.assert_any_call("appslist_failed_to_download")
|
||||
|
||||
assert new_official_lastUpdate > official_lastUpdate
|
||||
def test_appslist_update_timeout(mocker):
|
||||
|
||||
|
||||
def test_appslist_fetch_single_appslist():
|
||||
"""
|
||||
Register several lists but only fetch one. Check only one got updated.
|
||||
"""
|
||||
|
||||
assert app_listlists() == {}
|
||||
_register_new_appslist(URL_OFFICIAL_APP_LIST, "yunohost")
|
||||
_register_new_appslist("https://lol.com/appslist.json", "dummy")
|
||||
|
||||
time.sleep(1)
|
||||
# Initialize ...
|
||||
_initialize_appslists_system()
|
||||
|
||||
with requests_mock.Mocker() as m:
|
||||
|
||||
# Mock the server response with a valid (well, empty, yep) json
|
||||
m.register_uri("GET", URL_OFFICIAL_APP_LIST, text='{ }')
|
||||
|
||||
official_lastUpdate = app_listlists()["yunohost"]["lastUpdate"]
|
||||
dummy_lastUpdate = app_listlists()["dummy"]["lastUpdate"]
|
||||
app_fetchlist(name="yunohost")
|
||||
new_official_lastUpdate = app_listlists()["yunohost"]["lastUpdate"]
|
||||
new_dummy_lastUpdate = app_listlists()["dummy"]["lastUpdate"]
|
||||
|
||||
assert new_official_lastUpdate > official_lastUpdate
|
||||
assert new_dummy_lastUpdate == dummy_lastUpdate
|
||||
|
||||
|
||||
def test_appslist_fetch_unknownlist():
|
||||
"""
|
||||
Attempt to fetch an unknown list
|
||||
"""
|
||||
|
||||
assert app_listlists() == {}
|
||||
|
||||
with pytest.raises(YunohostError):
|
||||
app_fetchlist(name="swag")
|
||||
|
||||
|
||||
def test_appslist_fetch_url_but_no_name():
|
||||
"""
|
||||
Do a fetchlist with url given, but no name given
|
||||
"""
|
||||
|
||||
with pytest.raises(YunohostError):
|
||||
app_fetchlist(url=URL_OFFICIAL_APP_LIST)
|
||||
|
||||
|
||||
def test_appslist_fetch_badurl():
|
||||
"""
|
||||
Do a fetchlist with a bad url
|
||||
"""
|
||||
|
||||
app_fetchlist(url="https://not.a.valid.url/plop.json", name="plop")
|
||||
|
||||
|
||||
def test_appslist_fetch_badfile():
|
||||
"""
|
||||
Do a fetchlist and mock a response with a bad json
|
||||
"""
|
||||
assert app_listlists() == {}
|
||||
|
||||
_register_new_appslist(URL_OFFICIAL_APP_LIST, "yunohost")
|
||||
|
||||
with requests_mock.Mocker() as m:
|
||||
|
||||
m.register_uri("GET", URL_OFFICIAL_APP_LIST, text='{ not json lol }')
|
||||
|
||||
app_fetchlist()
|
||||
|
||||
|
||||
def test_appslist_fetch_404():
|
||||
"""
|
||||
Do a fetchlist and mock a 404 response
|
||||
"""
|
||||
assert app_listlists() == {}
|
||||
|
||||
_register_new_appslist(URL_OFFICIAL_APP_LIST, "yunohost")
|
||||
|
||||
with requests_mock.Mocker() as m:
|
||||
|
||||
m.register_uri("GET", URL_OFFICIAL_APP_LIST, status_code=404)
|
||||
|
||||
app_fetchlist()
|
||||
|
||||
|
||||
def test_appslist_fetch_sslerror():
|
||||
"""
|
||||
Do a fetchlist and mock an SSL error
|
||||
"""
|
||||
assert app_listlists() == {}
|
||||
|
||||
_register_new_appslist(URL_OFFICIAL_APP_LIST, "yunohost")
|
||||
|
||||
with requests_mock.Mocker() as m:
|
||||
|
||||
m.register_uri("GET", URL_OFFICIAL_APP_LIST,
|
||||
exc=requests.exceptions.SSLError)
|
||||
|
||||
app_fetchlist()
|
||||
|
||||
|
||||
def test_appslist_fetch_timeout():
|
||||
"""
|
||||
Do a fetchlist and mock a timeout
|
||||
"""
|
||||
assert app_listlists() == {}
|
||||
|
||||
_register_new_appslist(URL_OFFICIAL_APP_LIST, "yunohost")
|
||||
|
||||
with requests_mock.Mocker() as m:
|
||||
|
||||
m.register_uri("GET", URL_OFFICIAL_APP_LIST,
|
||||
# Timeout
|
||||
m.register_uri("GET", APPSLISTS_DEFAULT_URL_FULL,
|
||||
exc=requests.exceptions.ConnectTimeout)
|
||||
|
||||
app_fetchlist()
|
||||
with pytest.raises(YunohostError):
|
||||
mocker.spy(m18n, "n")
|
||||
_update_appslist()
|
||||
m18n.n.assert_any_call("appslist_failed_to_download")
|
||||
|
||||
|
||||
#
|
||||
# Test remove of appslist #
|
||||
#
|
||||
def test_appslist_update_sslerror(mocker):
|
||||
|
||||
# Initialize ...
|
||||
_initialize_appslists_system()
|
||||
|
||||
with requests_mock.Mocker() as m:
|
||||
|
||||
# SSL error
|
||||
m.register_uri("GET", APPSLISTS_DEFAULT_URL_FULL,
|
||||
exc=requests.exceptions.SSLError)
|
||||
|
||||
with pytest.raises(YunohostError):
|
||||
mocker.spy(m18n, "n")
|
||||
_update_appslist()
|
||||
m18n.n.assert_any_call("appslist_failed_to_download")
|
||||
|
||||
|
||||
def test_appslist_remove():
|
||||
"""
|
||||
Register a new appslist, then remove it
|
||||
"""
|
||||
def test_appslist_update_corrupted(mocker):
|
||||
|
||||
# Assume we're starting with an empty app list
|
||||
assert app_listlists() == {}
|
||||
# Initialize ...
|
||||
_initialize_appslists_system()
|
||||
|
||||
# Register a new dummy list
|
||||
_register_new_appslist("https://lol.com/appslist.json", "dummy")
|
||||
app_removelist("dummy")
|
||||
with requests_mock.Mocker() as m:
|
||||
|
||||
# Should end up with no list registered
|
||||
assert app_listlists() == {}
|
||||
# Corrupted json
|
||||
m.register_uri("GET", APPSLISTS_DEFAULT_URL_FULL,
|
||||
text=DUMMY_APPLIST[:-2])
|
||||
|
||||
with pytest.raises(YunohostError):
|
||||
mocker.spy(m18n, "n")
|
||||
_update_appslist()
|
||||
m18n.n.assert_any_call("appslist_failed_to_download")
|
||||
|
||||
|
||||
def test_appslist_remove_unknown():
|
||||
"""
|
||||
Attempt to remove an unknown list
|
||||
"""
|
||||
def test_appslist_load_with_empty_cache(mocker):
|
||||
|
||||
with pytest.raises(YunohostError):
|
||||
app_removelist("dummy")
|
||||
# Initialize ...
|
||||
_initialize_appslists_system()
|
||||
|
||||
# Cache is empty
|
||||
assert not glob.glob(APPSLISTS_CACHE + "/*")
|
||||
|
||||
# Update
|
||||
with requests_mock.Mocker() as m:
|
||||
|
||||
# Mock the server response with a dummy applist
|
||||
m.register_uri("GET", APPSLISTS_DEFAULT_URL_FULL, text=DUMMY_APPLIST)
|
||||
|
||||
# Try to load the applist
|
||||
# This should implicitly trigger an update in the background
|
||||
mocker.spy(m18n, "n")
|
||||
app_dict = _load_appslist()
|
||||
m18n.n.assert_any_call("appslist_obsolete_cache")
|
||||
m18n.n.assert_any_call("appslist_update_success")
|
||||
|
||||
|
||||
#
|
||||
# Test migration from legacy appslist system #
|
||||
#
|
||||
# Cache shouldn't be empty anymore empty
|
||||
assert glob.glob(APPSLISTS_CACHE + "/*")
|
||||
|
||||
assert "foo" in app_dict.keys()
|
||||
assert "bar" in app_dict.keys()
|
||||
|
||||
|
||||
def add_legacy_cron(name, url):
|
||||
with open("/etc/cron.d/yunohost-applist-%s" % name, "w") as f:
|
||||
f.write('00 00 * * * root yunohost app fetchlist -u %s -n %s > /dev/null 2>&1\n' % (url, name))
|
||||
def test_appslist_load_with_conflicts_between_lists(mocker):
|
||||
|
||||
# Initialize ...
|
||||
_initialize_appslists_system()
|
||||
|
||||
conf = [{"id": "default", "url": APPSLISTS_DEFAULT_URL},
|
||||
{"id": "default2", "url": APPSLISTS_DEFAULT_URL.replace("yunohost.org", "yolohost.org")}]
|
||||
|
||||
write_to_yaml(APPSLISTS_CONF, conf)
|
||||
|
||||
# Update
|
||||
with requests_mock.Mocker() as m:
|
||||
|
||||
# Mock the server response with a dummy applist
|
||||
# + the same applist for the second list
|
||||
m.register_uri("GET", APPSLISTS_DEFAULT_URL_FULL, text=DUMMY_APPLIST)
|
||||
m.register_uri("GET", APPSLISTS_DEFAULT_URL_FULL.replace("yunohost.org", "yolohost.org"), text=DUMMY_APPLIST)
|
||||
|
||||
# Try to load the applist
|
||||
# This should implicitly trigger an update in the background
|
||||
mocker.spy(logger, "warning")
|
||||
app_dict = _load_appslist()
|
||||
logger.warning.assert_any_call(AnyStringWith("Duplicate"))
|
||||
|
||||
# Cache shouldn't be empty anymore empty
|
||||
assert glob.glob(APPSLISTS_CACHE + "/*")
|
||||
|
||||
assert "foo" in app_dict.keys()
|
||||
assert "bar" in app_dict.keys()
|
||||
|
||||
|
||||
def test_appslist_check_using_legacy_system_testFalse():
|
||||
"""
|
||||
If no legacy cron job is there, the check should return False
|
||||
"""
|
||||
assert glob.glob("/etc/cron.d/yunohost-applist-*") == []
|
||||
assert _using_legacy_appslist_system() is False
|
||||
def test_appslist_load_with_oudated_api_version(mocker):
|
||||
|
||||
# Initialize ...
|
||||
_initialize_appslists_system()
|
||||
|
||||
# Update
|
||||
with requests_mock.Mocker() as m:
|
||||
|
||||
mocker.spy(m18n, "n")
|
||||
m.register_uri("GET", APPSLISTS_DEFAULT_URL_FULL, text=DUMMY_APPLIST)
|
||||
_update_appslist()
|
||||
|
||||
# Cache shouldn't be empty anymore empty
|
||||
assert glob.glob(APPSLISTS_CACHE + "/*")
|
||||
|
||||
# Tweak the cache to replace the from_api_version with a different one
|
||||
for cache_file in glob.glob(APPSLISTS_CACHE + "/*"):
|
||||
cache_json = read_json(cache_file)
|
||||
assert cache_json["from_api_version"] == APPSLISTS_API_VERSION
|
||||
cache_json["from_api_version"] = 0
|
||||
write_to_json(cache_file, cache_json)
|
||||
|
||||
# Update
|
||||
with requests_mock.Mocker() as m:
|
||||
|
||||
# Mock the server response with a dummy applist
|
||||
m.register_uri("GET", APPSLISTS_DEFAULT_URL_FULL, text=DUMMY_APPLIST)
|
||||
|
||||
mocker.spy(m18n, "n")
|
||||
app_dict = _load_appslist()
|
||||
m18n.n.assert_any_call("appslist_update_success")
|
||||
|
||||
assert "foo" in app_dict.keys()
|
||||
assert "bar" in app_dict.keys()
|
||||
|
||||
# Check that we indeed have the new api number in cache
|
||||
for cache_file in glob.glob(APPSLISTS_CACHE + "/*"):
|
||||
cache_json = read_json(cache_file)
|
||||
assert cache_json["from_api_version"] == APPSLISTS_API_VERSION
|
||||
|
||||
|
||||
def test_appslist_check_using_legacy_system_testTrue():
|
||||
"""
|
||||
If there's a legacy cron job, the check should return True
|
||||
"""
|
||||
assert glob.glob("/etc/cron.d/yunohost-applist-*") == []
|
||||
add_legacy_cron("yunohost", "https://app.yunohost.org/official.json")
|
||||
assert _using_legacy_appslist_system() is True
|
||||
|
||||
def test_appslist_migrate_legacy_explicitly():
|
||||
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
def test_appslist_system_migration():
|
||||
"""
|
||||
Test that legacy cron jobs get migrated correctly when calling app_listlists
|
||||
"""
|
||||
def test_appslist_migrate_legacy_implicitly():
|
||||
|
||||
# Start with no legacy cron, no appslist registered
|
||||
assert glob.glob("/etc/cron.d/yunohost-applist-*") == []
|
||||
assert app_listlists() == {}
|
||||
assert not os.path.exists("/etc/cron.daily/yunohost-fetch-appslists")
|
||||
|
||||
# Add a few legacy crons
|
||||
add_legacy_cron("yunohost", "https://app.yunohost.org/official.json")
|
||||
add_legacy_cron("dummy", "https://swiggitty.swaggy.lol/yolo.json")
|
||||
|
||||
# Migrate
|
||||
assert _using_legacy_appslist_system() is True
|
||||
_migrate_appslist_system()
|
||||
assert _using_legacy_appslist_system() is False
|
||||
|
||||
# No legacy cron job should remain
|
||||
assert glob.glob("/etc/cron.d/yunohost-applist-*") == []
|
||||
|
||||
# Check they are in app_listlists anyway
|
||||
appslist_dict = app_listlists()
|
||||
assert "yunohost" in appslist_dict.keys()
|
||||
assert appslist_dict["yunohost"]["url"] == "https://app.yunohost.org/official.json"
|
||||
assert "dummy" in appslist_dict.keys()
|
||||
assert appslist_dict["dummy"]["url"] == "https://swiggitty.swaggy.lol/yolo.json"
|
||||
|
||||
assert cron_job_is_there()
|
||||
|
||||
|
||||
def test_appslist_system_migration_badcron():
|
||||
"""
|
||||
Test the migration on a bad legacy cron (no url found inside cron job)
|
||||
"""
|
||||
|
||||
# Start with no legacy cron, no appslist registered
|
||||
assert glob.glob("/etc/cron.d/yunohost-applist-*") == []
|
||||
assert app_listlists() == {}
|
||||
assert not os.path.exists("/etc/cron.daily/yunohost-fetch-appslists")
|
||||
|
||||
# Add a "bad" legacy cron
|
||||
add_legacy_cron("wtflist", "ftp://the.fuck.is.this")
|
||||
|
||||
# Migrate
|
||||
assert _using_legacy_appslist_system() is True
|
||||
_migrate_appslist_system()
|
||||
assert _using_legacy_appslist_system() is False
|
||||
|
||||
# No legacy cron should remain, but it should be backuped in /etc/yunohost
|
||||
assert glob.glob("/etc/cron.d/yunohost-applist-*") == []
|
||||
assert os.path.exists("/etc/yunohost/wtflist.oldlist.bkp")
|
||||
|
||||
# Appslist should still be empty
|
||||
assert app_listlists() == {}
|
||||
|
||||
|
||||
def test_appslist_system_migration_conflict():
|
||||
"""
|
||||
Test migration of conflicting cron job (in terms of url)
|
||||
"""
|
||||
|
||||
# Start with no legacy cron, no appslist registered
|
||||
assert glob.glob("/etc/cron.d/yunohost-applist-*") == []
|
||||
assert app_listlists() == {}
|
||||
assert not os.path.exists("/etc/cron.daily/yunohost-fetch-appslists")
|
||||
|
||||
# Add a few legacy crons
|
||||
add_legacy_cron("yunohost", "https://app.yunohost.org/official.json")
|
||||
add_legacy_cron("dummy", "https://app.yunohost.org/official.json")
|
||||
|
||||
# Migrate
|
||||
assert _using_legacy_appslist_system() is True
|
||||
_migrate_appslist_system()
|
||||
assert _using_legacy_appslist_system() is False
|
||||
|
||||
# No legacy cron job should remain
|
||||
assert glob.glob("/etc/cron.d/yunohost-applist-*") == []
|
||||
|
||||
# Only one among "dummy" and "yunohost" should be listed
|
||||
appslist_dict = app_listlists()
|
||||
assert (len(appslist_dict.keys()) == 1)
|
||||
assert ("dummy" in appslist_dict.keys()) or ("yunohost" in appslist_dict.keys())
|
||||
|
||||
assert cron_job_is_there()
|
||||
raise NotImplementedError
|
||||
|
|
Loading…
Add table
Reference in a new issue