125 lines
5.2 KiB
Python
125 lines
5.2 KiB
Python
import asyncio
|
|
import io
|
|
import os
|
|
import uuid
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app import common, store, pipeline, download
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(store, 'DATA', tmp_path)
|
|
monkeypatch.setattr(store, 'DB_PATH', tmp_path / 'queue.sqlite3')
|
|
monkeypatch.setattr(pipeline, 'RAW', tmp_path / 'raw')
|
|
monkeypatch.setattr(pipeline, 'READY', tmp_path / 'ready')
|
|
monkeypatch.setattr(pipeline, 'WORK', tmp_path / 'work')
|
|
monkeypatch.setattr(pipeline, 'LOGS', tmp_path / 'logs')
|
|
store.init_db()
|
|
return tmp_path
|
|
|
|
|
|
def test_uris():
|
|
assert common.validate_url('https://archive.org/download/demo/game.iso')[0] == 'url'
|
|
assert common.validate_url('https://vimm.net/vault/12345')[0] == 'vimm'
|
|
assert common.validate_url('magnet:?xt=urn:btih:c98fbaa2a36ee48e3d9423be02cdbefd9281bc7d')[0] == 'magnet'
|
|
for url in ('file:///etc/passwd', 'http://127.0.0.1/admin',
|
|
'https://user:password@example.org', 'https://localhost/file',
|
|
'ftp://example.org/file'):
|
|
with pytest.raises(ValueError):
|
|
common.validate_url(url)
|
|
|
|
|
|
def test_archive_id_and_escape():
|
|
assert common.parse_archive_id('https://archive.org/details/some-item') == 'some-item'
|
|
assert common.archive_download_url('some-item', 'Game (1).7z').endswith('Game%20%281%29.7z')
|
|
for name in ('../target.iso', '/etc/password', 'foo/../../abc', 'a\\b', ''):
|
|
with pytest.raises(ValueError):
|
|
common.archive_download_url('item', name)
|
|
for name in ('../../secret', '/absolute/file', 'C:\\Windows\\System32\\x', 'folder/../x'):
|
|
with pytest.raises(pipeline.ProcessingError):
|
|
pipeline.safe_archive_path(name)
|
|
|
|
|
|
def test_database_queue(isolated):
|
|
for t in ('Test1', 'Test2'):
|
|
store.create({'id': t, 'title': t, 'source_type': 'url',
|
|
'source': 'https://example.org/test.zip', 'format': 'xex'})
|
|
assert len(store.list_jobs()) == 2
|
|
assert store.next_job()['id'] == 'Test1'
|
|
store.update('Test1', state='done', progress=100)
|
|
assert store.next_job()['id'] == 'Test2'
|
|
|
|
|
|
def test_process_xex_zip(isolated):
|
|
jid = str(uuid.uuid4())
|
|
store.create({'id': jid, 'title': 'Demo Game', 'source_type': 'import',
|
|
'source': 'game.zip', 'format': 'xex'})
|
|
raw = isolated / 'raw' / jid
|
|
raw.mkdir(parents=True)
|
|
with zipfile.ZipFile(raw / 'game.zip', 'w') as archive:
|
|
archive.writestr('folder/default.xex', b'This is a fake XEX for queue testing')
|
|
archive.writestr('folder/data/readme.txt', b'hello')
|
|
asyncio.run(pipeline.process_job(store.get(jid)))
|
|
job = store.get(jid)
|
|
assert job['state'] == 'done'
|
|
assert (Path(job['output']) / 'Gra' / 'default.xex').exists()
|
|
assert (Path(job['output']) / 'Gra' / 'data' / 'readme.txt').read_bytes() == b'hello'
|
|
|
|
|
|
def test_zip_traversal_rejected(isolated):
|
|
jid = 'traversal'
|
|
store.create({'id': jid, 'title': 'Bad', 'source_type': 'import',
|
|
'source': 'bad.zip', 'format': 'xex'})
|
|
raw = isolated / 'raw' / jid
|
|
raw.mkdir(parents=True)
|
|
with zipfile.ZipFile(raw / 'bad.zip', 'w') as archive:
|
|
archive.writestr('../../pwned', b'nope')
|
|
with pytest.raises(pipeline.ProcessingError):
|
|
asyncio.run(pipeline.process_job(store.get(jid)))
|
|
assert not (isolated / 'pwned').exists()
|
|
|
|
|
|
def test_raw_copy(isolated):
|
|
jid = 'simple'
|
|
store.create({'id': jid, 'title': 'Keep File', 'source_type': 'import',
|
|
'source': 'anything.bin', 'format': 'raw'})
|
|
raw = isolated / 'raw' / jid
|
|
raw.mkdir(parents=True)
|
|
(raw / 'anything.bin').write_bytes(b'abc')
|
|
asyncio.run(pipeline.process_job(store.get(jid)))
|
|
assert (Path(store.get(jid)['output']) / 'files' / 'anything.bin').read_bytes() == b'abc'
|
|
|
|
|
|
def test_multipart_urls():
|
|
job={'source_type':'archive_many', 'source':'["https://host/A.part1.rar", "https://host/A.part2.rar"]'}
|
|
assert len(download.source_urls(job)) == 2
|
|
assert download.target_filename('https://example.com/Gra%20XYZ.iso') == 'Gra XYZ.iso'
|
|
|
|
|
|
def test_api_auth_and_add(isolated, monkeypatch):
|
|
from fastapi.testclient import TestClient
|
|
from app import main
|
|
monkeypatch.setenv('WEB_USER', 'admin')
|
|
monkeypatch.setenv('WEB_PASSWORD', 'secretsecret1234')
|
|
async def idle_worker():
|
|
await asyncio.sleep(10000)
|
|
monkeypatch.setattr(main, 'worker', idle_worker)
|
|
with TestClient(main.app) as client:
|
|
assert client.get('/').status_code == 401
|
|
assert client.get('/api/jobs').status_code == 401
|
|
auth = ('admin', 'secretsecret1234')
|
|
assert client.get('/', auth=auth).status_code == 200
|
|
r = client.post('/api/jobs', auth=auth,
|
|
json={'url': 'https://example.com/DEMO.iso','format': 'xex'})
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()['title'] == 'DEMO.iso'
|
|
assert client.get('/api/jobs', auth=auth).json()[0]['id'] == r.json()['id']
|
|
assert client.post('/api/jobs', auth=auth,
|
|
json={'url':'file:///etc/passwd','format':'xex'}).status_code == 400
|
|
assert client.post('/api/jobs', auth=auth,
|
|
json={'url':'https://example.org/foo.iso','format':'oops'}).status_code == 400
|