mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 19:01:39 -04:00
Repos keeps local mirrors of git repositories, cloned over https and kept up to date on a schedule. - A Repo is a Collection (kind='repo') plus a `repository` row. The new `git` downloader clones it, then updates it (weekly by default, or Never: the clone is kept as it is and only updated on request). - Repos are saved in `repos/<tag>/<name>` (the Repos Directory, ignored by default) or a directory the user chooses. An existing clone can be imported where it is. repos.yaml is the source of truth. - A GitHub-like page shows each repo's README, files, and history, with a ZIP download. READMEs are searchable under the global search's Other tab. - git is hardened against hostile repos and configs: https only, pinned addresses with no private addresses (for the repo and its submodules), sanitized configs, no hooks or links, and only clones WROLPi made or imported are updated. - Tagging a Repo moves its clone as one directory. Repo changes wait for repos.yaml to be imported at startup. - The file mover groups unindexed files with no primary file as one FileGroup, and never replaces a file at the destination. A failed tag move restores the Collection's tag. A Collection's directory cannot leave the media directory. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
91 lines
4.1 KiB
Python
91 lines
4.1 KiB
Python
import pytest
|
|
|
|
from modules.repos import lib
|
|
from modules.repos.models import Repository
|
|
from wrolpi.collections.models import Collection
|
|
from wrolpi.downloader import Download, DownloadFrequency
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dump_and_import(test_session, test_directory, async_client, test_wrolpi_config, await_switches):
|
|
"""repos.yaml is the source of truth; a lost database is rebuilt from it."""
|
|
lib.create_repository(test_session, 'https://github.com/kiwix/kiwix-tools', tag_name='software',
|
|
description='Tools', frequency=DownloadFrequency.daily)
|
|
lib.create_repository(test_session, 'https://github.com/a/b', mode='snapshot', branch='dev')
|
|
await await_switches()
|
|
|
|
config = lib.get_repos_config()
|
|
assert config.get_file().is_file()
|
|
# Each Repo's random clone token is kept, so a restored Repo still owns its clone.
|
|
dumped = [dict(i) for i in config.repos]
|
|
assert all(len(i.pop('clone_token')) == 32 for i in dumped)
|
|
assert dumped == [
|
|
dict(name='b', url='https://github.com/a/b', directory='repos/b', tag_name=None, description=None,
|
|
mode='snapshot', branch='dev', submodules=False, frequency=DownloadFrequency.weekly),
|
|
dict(name='kiwix-tools', url='https://github.com/kiwix/kiwix-tools', directory='repos/software/kiwix-tools',
|
|
tag_name='software', description='Tools', mode='full', branch=None, submodules=False,
|
|
frequency=DownloadFrequency.daily),
|
|
]
|
|
|
|
# Lose the database.
|
|
for repo in test_session.query(Repository).all():
|
|
lib._delete_repository(test_session, repo)
|
|
test_session.commit()
|
|
assert test_session.query(Collection).count() == 0
|
|
|
|
config.import_config()
|
|
test_session.expire_all()
|
|
repos = {i.name: i for i in test_session.query(Repository)}
|
|
assert repos['kiwix-tools'].directory == test_directory / 'repos/software/kiwix-tools'
|
|
assert repos['kiwix-tools'].tag_name == 'software'
|
|
assert repos['kiwix-tools'].collection.description == 'Tools'
|
|
assert repos['kiwix-tools'].download.frequency == DownloadFrequency.daily
|
|
assert (repos['b'].mode, repos['b'].branch) == ('snapshot', 'dev')
|
|
assert {i.downloader for i in test_session.query(Download)} == {'git'}
|
|
|
|
# A repo removed from the config is deleted.
|
|
data = config.read_config_file()
|
|
data['repos'] = data['repos'][1:]
|
|
data['version'] += 1
|
|
config.write_config_data(data, config.get_file())
|
|
config.import_config()
|
|
test_session.expire_all()
|
|
assert [i.name for i in test_session.query(Repository)] == ['kiwix-tools']
|
|
assert test_session.query(Download).count() == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_links_existing_download(test_session, async_client, test_wrolpi_config):
|
|
"""A Repo imported after download_manager.yaml re-uses its Download."""
|
|
lib.create_repository(test_session, 'https://github.com/a/b')
|
|
test_session.query(Repository).delete()
|
|
test_session.query(Collection).delete()
|
|
test_session.commit()
|
|
download = test_session.query(Download).one()
|
|
assert download.collection_id is None
|
|
|
|
config = lib.get_repos_config()
|
|
config.write_config_data(dict(version=1, repos=[dict(name='b', url='https://github.com/a/b', frequency=3600)]),
|
|
config.get_file())
|
|
config.import_config()
|
|
test_session.expire_all()
|
|
|
|
repo = test_session.query(Repository).one()
|
|
assert test_session.query(Download).one().id == download.id
|
|
assert download.collection_id == repo.collection_id
|
|
assert download.frequency == 3600
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_last_repo(test_session, async_client, test_wrolpi_config, await_switches):
|
|
"""Deleting the last Repo empties the config, so it does not return on the next import."""
|
|
repo = lib.create_repository(test_session, 'https://github.com/a/b')
|
|
await await_switches()
|
|
config = lib.get_repos_config()
|
|
assert len(config.read_config_file()['repos']) == 1
|
|
|
|
request, response = await async_client.delete(f'/api/repos/{repo.id}')
|
|
assert config.read_config_file()['repos'] == []
|
|
|
|
config.import_config()
|
|
assert test_session.query(Repository).count() == 0
|