73 lines
3.7 KiB
Python
Executable File
73 lines
3.7 KiB
Python
Executable File
import unittest
|
|
from datetime import datetime, timedelta
|
|
import copy
|
|
from weekly import due, new_week, record, ZONE
|
|
from release import ssh_command, REMOTE_COMMAND
|
|
|
|
|
|
class WeeklyTests(unittest.TestCase):
|
|
def state(self):
|
|
return {'starts_on': '2026-11-02', 'weeks': {'2026-11-02': {'slots': [
|
|
{'slot_id': str(i), 'status': 'planned', 'planned_at': f'2026-11-0{day}T15:00:00+08:00',
|
|
'release_id': None, 'published_at': None} for i, day in enumerate([2, 4, 6])
|
|
]}}}
|
|
|
|
def test_random_plans_have_three_distinct_days_at_fifteen(self):
|
|
for _ in range(200):
|
|
slots = new_week('2026-11-02')['slots']
|
|
times = [datetime.fromisoformat(s['planned_at']) for s in slots]
|
|
self.assertEqual(len({t.date() for t in times}), 3)
|
|
self.assertTrue(all(t.hour == 15 and t.minute == t.second == 0 for t in times))
|
|
self.assertTrue(all(b-a >= timedelta(hours=24) for a, b in zip(times, times[1:])))
|
|
|
|
def test_missed_slot_catches_up_without_two_same_day(self):
|
|
state = self.state()
|
|
now = datetime(2026, 11, 4, 15, tzinfo=ZONE)
|
|
self.assertEqual(due(state, now)['slot']['slot_id'], '0')
|
|
slot = state['weeks']['2026-11-02']['slots'][0]
|
|
slot.update(status='published', published_at=now.isoformat())
|
|
self.assertEqual(due(state, now)['reason'], 'already_published_today')
|
|
self.assertEqual(due(state, now+timedelta(days=1))['slot']['slot_id'], '1')
|
|
|
|
def test_retry_pending_is_not_a_new_article(self):
|
|
state = self.state()
|
|
slot = state['weeks']['2026-11-02']['slots'][0]
|
|
slot.update(release_id='original', status='publishing')
|
|
self.assertEqual(due(state, datetime(2026, 11, 9, 15, tzinfo=ZONE))['action'], 'resume')
|
|
response = {'ok': True, 'data': {'release_id': 'original', 'phase': 'published_pending',
|
|
'published_at': '2026-11-08T16:11:10+08:00',
|
|
'receipt': {'status': 'published', 'publication_id': 'receipt', 'articles': [{'url': '/a/example.html'}]}}}
|
|
record(state, '0', response, datetime(2026, 11, 9, 15, tzinfo=ZONE))
|
|
self.assertEqual(slot['published_at'], '2026-11-08T16:11:10+08:00')
|
|
self.assertEqual(due(state, datetime(2026, 11, 9, 15, tzinfo=ZONE))['action'], 'synchronize')
|
|
response['ok'] = False
|
|
before = copy.deepcopy(state)
|
|
with self.assertRaises(ValueError):
|
|
record(state, '0', response, datetime(2026, 11, 9, 15, tzinfo=ZONE))
|
|
self.assertEqual(before, state)
|
|
|
|
def test_next_week_does_not_publish_unstarted_backlog(self):
|
|
state = self.state()
|
|
now = datetime(2026, 11, 9, 15, tzinfo=ZONE)
|
|
result = due(state, now)
|
|
if result['action'] == 'write':
|
|
self.assertTrue(result['slot']['slot_id'].startswith('2026-11-09'))
|
|
for i, slot in enumerate(state['weeks']['2026-11-02']['slots']):
|
|
slot.update(status='published', published_at=f'2026-11-0{2+i}T16:00:00+08:00')
|
|
self.assertEqual(due(state, datetime(2026, 11, 6, 15, tzinfo=ZONE))['reason'], 'weekly_quota_met')
|
|
|
|
def test_transport_always_forces_white_host_and_fixed_remote_command(self):
|
|
command = ssh_command()
|
|
proxy = next(a for a in command if a.startswith('ProxyCommand='))
|
|
self.assertIn('root@104.168.30.40', proxy)
|
|
self.assertIn('-W 45.153.247.44:22', proxy)
|
|
self.assertIn('StrictHostKeyChecking=yes', proxy)
|
|
self.assertIn('ControlPath=none', proxy)
|
|
self.assertEqual(command[-2], 'root@45.153.247.44')
|
|
self.assertEqual(command[-1], REMOTE_COMMAND)
|
|
self.assertEqual(command[1:3], ['-F', '/dev/null'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|