File size: 1,628 Bytes
e5e9d87 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 | from pathlib import Path
import asyncio
from services.ocean_batch import OceanBatchManager
from services.download_jobs import DownloadJobStore
def test_ocean_jobs_persist_and_are_user_scoped(tmp_path):
m=OceanBatchManager(tmp_path/'ocean','http://example.invalid')
spec={'start_date':'2002-08-01','end_date':'2002-08-02','variables':['sst'],'bbox':{'lon_min':120,'lon_max':140,'lat_min':20,'lat_max':40},'format':'csv','region_name':'test','region_source':'explicit_coordinates','days':2,'subtask_count':2}
async def fake_start(jid): return None
m.start=fake_start
job=asyncio.run(m.create('u1','test',spec,'thread-123'))
assert job['thread_id']=='thread-123'
restored=OceanBatchManager(tmp_path/'ocean','http://example.invalid')
items=restored.list_user('u1')
assert len(items)==1 and items[0]['job_id']==job['job_id']
assert restored.list_user('u2')==[]
def test_download_jobs_are_listed_after_store_reopen(tmp_path):
root=tmp_path/'downloads'
s=DownloadJobStore(root)
j=s.create('u1','https://example.com/a','a.nc',100,True,'application/octet-stream')
s.progress(j['job_id'],25)
s2=DownloadJobStore(root)
items=s2.list_user('u1')
assert len(items)==1
assert items[0]['transferred_bytes']==25
assert items[0]['resume_supported'] is True
def test_frontend_has_persistent_task_center_and_single_retry():
js=Path('static/js/app.js').read_text(encoding='utf-8')
assert '持久化任务' in js
assert '/subtasks/'+"'"+'+encodeURIComponent(btn.dataset.sub)+'+"'"+'/retry' in js or '/subtasks/' in js
assert '回到关联对话' in js
|