import importlib.util import json from pathlib import Path import tempfile import time import unittest from unittest.mock import patch import task_core as core import taskd class TaskTests(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory(prefix='scheduled-task-qa-') self.original=(core.BASE,core.LOGS,core.UNITS,taskd.ADMIN) folder=Path(self.temp.name) core.BASE=folder/'data';core.LOGS=folder/'logs';core.UNITS=folder/'units';core.UNITS.mkdir() taskd.ADMIN=folder/'admin.json';taskd.SESSIONS.clear();taskd.ATTEMPTS.clear();core.init_store() core.atom_write(taskd.ADMIN,json.dumps(taskd.password_record('only-a-test-password'))) def tearDown(self): core.BASE,core.LOGS,core.UNITS,taskd.ADMIN=self.original taskd.SESSIONS.clear();taskd.ATTEMPTS.clear();self.temp.cleanup() def definition(self,**values): return {'name':'测试任务','mode':'command','body':'printf hello','enabled':False, 'schedule':{'kind':'daily','time':'03:00'},**values} def test_daily_weekly_monthly_and_interval(self): self.assertEqual(core.schedule_definition({'kind':'daily','time':'03:15'})[0],['OnCalendar=*-*-* 03:15:00 Asia/Shanghai']) self.assertIn('Mon,Sun',core.schedule_definition({'kind':'weekly','weekdays':[0,6],'time':'04:00'})[0][0]) self.assertIn('跳过',core.schedule_definition({'kind':'monthly','day':31,'time':'02:00'})[1]) interval=core.schedule_definition({'kind':'minutes','every':10})[0] self.assertEqual(interval,['OnActiveSec=600s','OnUnitInactiveSec=600s']) self.assertFalse(any('OnBootSec' in x for x in interval)) def test_cron_lists_ranges_steps_and_day_or(self): output=core.cron_calendars('*/15 3 * * MON-FRI') self.assertEqual(len(output),1) self.assertIn('0,15,30,45:00',output[0]) self.assertIn('Mon,Tue,Wed,Thu,Fri',output[0]) output=core.cron_calendars('0 3 1 * 1') self.assertEqual(len(output),2) self.assertIn('*-*-1',output[0]);self.assertIn('Mon',output[1]) for bad in ['0 3 * *','*/0 * * * *','0 24 * * *','0 3 32 * *','0 3 * * 9','0 3 L * *']: with self.subTest(bad=bad),self.assertRaises(core.TaskError):core.cron_calendars(bad) def test_task_text_never_becomes_unit_instructions(self): definition=core.validate_definition(self.definition(body='echo test\n# ExecStart=/bad/example'),check_paths=False) service,timer=core.render_units('a'*32,definition) self.assertNotIn('echo test',service);self.assertNotIn('/bad/example',service) self.assertIn('task_exec.py '+'a'*32,service) self.assertIn('Persistent=false',timer) with self.assertRaises(core.TaskError):core.validate_definition(self.definition(name='name\nExecStart=bad'),check_paths=False) with self.assertRaises(core.TaskError):core.validate_id('../../etc/shadow') def test_validation_limits_and_past_dates(self): for changes in [{'body':''},{'timeout':1},{'timeout':90000},{'body':'x'*65537},{'enabled':'false'}, {'schedule':{'kind':'once','at':'2020-01-01T03:00'}}]: with self.subTest(changes=list(changes)),self.assertRaises(core.TaskError):core.validate_definition(self.definition(**changes),check_paths=False) self.assertTrue(core.validate_definition(self.definition(body='systemctl reboot'),check_paths=False)['dangerous']) def test_session_and_csrf_gate_privileged_operations(self): with patch.object(core,'save_task') as save: with self.assertRaises(core.TaskError) as error:taskd.dispatch({'action':'tasks.save','payload':self.definition()}) self.assertEqual(error.exception.status,401);save.assert_not_called() login=taskd.login({'username':'admin','password':'only-a-test-password'},'qa') with patch.object(core,'set_enabled') as toggle: with self.assertRaises(core.TaskError) as error:taskd.dispatch({'action':'tasks.enable','token':login['token'],'csrf':'wrong','payload':{'id':'a'*32,'enabled':True}}) self.assertEqual(error.exception.status,403);toggle.assert_not_called() self.assertTrue(taskd.dispatch({'action':'auth.status','token':login['token']})['logged_in']) taskd.dispatch({'action':'auth.logout','token':login['token'],'csrf':login['csrf']}) self.assertFalse(taskd.dispatch({'action':'auth.status','token':login['token']})['logged_in']) def test_password_change_revokes_all_sessions(self): one=taskd.login({'username':'admin','password':'only-a-test-password'},'qa') two=taskd.login({'username':'admin','password':'only-a-test-password'},'qa-2') taskd.dispatch({'action':'auth.password','token':one['token'],'csrf':one['csrf'], 'payload':{'old_password':'only-a-test-password','new_password':'new-test-password-123'}}) self.assertFalse(taskd.verify('only-a-test-password'));self.assertTrue(taskd.verify('new-test-password-123')) self.assertFalse(taskd.dispatch({'action':'auth.status','token':two['token']})['logged_in']) def test_wrong_old_password_keeps_current_login(self): login=taskd.login({'username':'admin','password':'only-a-test-password'},'qa') with self.assertRaises(core.TaskError) as error: taskd.dispatch({'action':'auth.password','token':login['token'],'csrf':login['csrf'], 'payload':{'old_password':'wrong-password','new_password':'new-test-password-123'}}) self.assertEqual(error.exception.status,400) self.assertTrue(taskd.dispatch({'action':'auth.status','token':login['token']})['logged_in']) def test_log_tail_is_bounded_and_missing_id_rejected(self): task_id='b'*32;run_id='c'*32 with core.db() as con: con.execute('INSERT INTO runs(id,task_id,task_name,started,state) VALUES(?,?,?,?,?)',(run_id,task_id,'test',time.time(),'success')) (core.LOGS/(run_id+'.log')).write_text('x'*200000) result=core.read_run(run_id) self.assertTrue(result['tail_only']);self.assertEqual(len(result['output']),131072) with self.assertRaises(core.TaskError):core.read_run('../password') if __name__=='__main__':unittest.main(verbosity=2)