import asyncio import tempfile import unittest from datetime import datetime, timedelta, timezone from pathlib import Path from unittest.mock import patch from fastapi.testclient import TestClient from server.db import decode, encode from server.main import create_app from server.notifications import Config, Transport from server.notification_config import TelegramConfigInput TOKEN_A = '123456789:' + 'A' * 35 TOKEN_B = '987654321:' + 'B' * 35 SMTP_A = {'host':'smtp.example.com','port':587,'mode':'starttls','user':'alice@example.com','password':' password $ with spaces ','from':'alice@example.com'} class NotificationConfigTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.path = Path(self.temp.name)/'test.db' self.app = create_app(self.path, start_worker=False, config=Config()) self.db = self.app.state.db self.notifier = self.app.state.notifier self.a, self.b = TestClient(self.app), TestClient(self.app) self.addCleanup(self.a.close) self.addCleanup(self.b.close) self.people, self.headers = [], [] for client, name in [(self.a, 'Alice'), (self.b, 'Bob')]: reply = client.post('/api/auth/register', json={'name':name,'email':f'{name.lower()}@example.com','password':'long-password-123'}) self.assertEqual(reply.status_code, 201, reply.text) self.people.append(reply.json()['user']['id']) self.headers.append({'X-CSRF-Token':reply.json()['csrfToken']}) def save(self, channel, payload, client=None, headers=None): reply = (client or self.a).put(f'/api/notifications/config/{channel}', headers=headers or self.headers[0], json=payload) self.assertEqual(reply.status_code, 200, reply.text) return reply def services(self, client=None): return (client or self.a).get('/api/workspace').json()['workspace']['services'] def test_saved_configuration_is_owned_persistent_and_never_returns_secrets(self): for channel, payload in [('telegram',{'token':TOKEN_A}), ('email',SMTP_A)]: reply = self.save(channel, payload) self.assertNotIn(TOKEN_A, reply.text) self.assertNotIn(SMTP_A['password'], reply.text) self.assertTrue(self.services()[channel]['configured']) self.assertFalse(self.services(self.b)[channel]['configured']) response = self.a.get('/api/notifications/config') self.assertEqual(response.status_code, 200, response.text) self.assertNotIn(TOKEN_A, response.text) self.assertNotIn(SMTP_A['password'], response.text) self.assertEqual(response.json()['services']['email']['host'], 'smtp.example.com') self.assertTrue(response.json()['services']['telegram']['tokenSaved']) restarted = create_app(self.path, start_worker=False, config=Config()) with TestClient(restarted) as client: client.cookies.update(self.a.cookies) self.assertTrue(self.services(client)['email']['configured']) self.assertTrue(self.services(client)['telegram']['configured']) def test_blank_secret_preserves_saved_credentials_and_password_whitespace(self): self.save('telegram', {'token':TOKEN_A}) self.save('telegram', {'token':''}) self.save('email', SMTP_A) self.save('email', {**SMTP_A, 'password':'', 'port':465, 'mode':'ssl'}) received = [] class SMTP: def __init__(self, host, port, **kwargs): received.append((host, port)) def __enter__(self): return self def __exit__(self, *args): pass def ehlo_or_helo_if_needed(self): pass def ehlo(self): return 250, b'ok' def login(self, user, password): received.append((user, password)) def send_message(self, message): received.append((message['From'], message['To'])) return {} with patch('server.notifications.smtplib.SMTP_SSL', SMTP): reply = self.a.post('/api/notifications/email/verify', headers=self.headers[0]) self.assertEqual(reply.status_code, 200, reply.text) self.assertEqual(received[:2], [('smtp.example.com',465), ('alice@example.com',' password $ with spaces ')]) self.assertEqual(received[2], ('alice@example.com','alice@example.com')) def test_save_requires_session_csrf_and_does_not_accept_another_owner(self): route = '/api/notifications/config/telegram' with TestClient(self.app) as visitor: self.assertEqual(visitor.put(route,json={'token':TOKEN_A}).status_code,401) self.assertEqual(self.a.put(route,json={'token':TOKEN_A}).status_code,403) reply = self.a.put(route,headers=self.headers[0],json={'token':TOKEN_A,'userId':self.people[1]}) self.assertEqual(reply.status_code,422) self.assertFalse(self.services(self.b)['telegram']['configured']) def test_invalid_configuration_never_echoes_credentials_and_keeps_previous_value(self): self.save('telegram', {'token':TOKEN_A}) for secret in ['https://secret.example/token','1234:secret/malicious','secret\npassword']: reply = self.a.put('/api/notifications/config/telegram',headers=self.headers[0],json={'token':secret}) self.assertEqual(reply.status_code,422,reply.text) self.assertNotIn(secret,reply.text) for change in [{'port':0},{'mode':'unsafe'},{'host':'https://smtp.example.com'},{'from':'alice@example.com\r\nBcc:secret@example.com'}]: reply = self.a.put('/api/notifications/config/email',headers=self.headers[0],json={**SMTP_A,**change}) self.assertEqual(reply.status_code,422,reply.text) self.assertNotIn(SMTP_A['password'],reply.text) self.assertTrue(self.services()['telegram']['configured']) self.assertFalse(self.services()['email']['configured']) def test_telegram_bot_cannot_be_claimed_by_two_accounts(self): self.save('telegram', {'token':TOKEN_A}) reply = self.b.put('/api/notifications/config/telegram',headers=self.headers[1],json={'token':TOKEN_A}) self.assertEqual(reply.status_code,409,reply.text) self.assertNotIn(TOKEN_A,reply.text) self.assertFalse(self.services(self.b)['telegram']['configured']) def test_changing_bot_invalidates_old_binding_and_requires_rebinding(self): self.save('telegram', {'token':TOKEN_A}) async def telegram(transport, method, payload): return {'id':123456789,'is_bot':True,'first_name':'Test','username':'AliceTestBot'} with patch.object(Transport,'telegram',telegram): code = self.a.post('/api/notifications/telegram/binding',headers=self.headers[0]).json()['code'] message = {'message':{'chat':{'id':12345,'type':'private'},'text':f'/start {code}'}} self.assertTrue(self.notifier.process_update(message,user_id=self.people[0])) self.save('telegram', {'token':TOKEN_B}) settings = self.a.get('/api/workspace').json()['workspace']['settings'] self.assertFalse(settings['telegramVerified']) self.assertFalse(settings['telegram']) self.assertEqual(settings['chatId'],'') self.assertFalse(self.notifier.process_update(message,user_id=self.people[0])) def test_personal_bot_cannot_consume_another_accounts_binding_code(self): self.save('telegram', {'token':TOKEN_A}) self.save('telegram', {'token':TOKEN_B},self.b,self.headers[1]) async def telegram(transport, method, payload): return {'id':123456789,'is_bot':True,'first_name':'Test','username':'AliceTestBot'} with patch.object(Transport,'telegram',telegram): code = self.a.post('/api/notifications/telegram/binding',headers=self.headers[0]).json()['code'] message = {'message':{'chat':{'id':12345,'type':'private'},'text':f'/start {code}'}} self.assertFalse(self.notifier.process_update(message,user_id=self.people[1])) self.assertFalse(self.notifier.process_update(message)) self.assertTrue(self.notifier.process_update(message,user_id=self.people[0])) def test_background_reminders_use_each_accounts_saved_token_immediately(self): messages = [] for client, headers, token, chat in [(self.a,self.headers[0],TOKEN_A,'100'),(self.b,self.headers[1],TOKEN_B,'200')]: self.save('telegram',{'token':token},client,headers) reply = client.post('/api/cards',headers=headers,json={'name':'Personal SIM','provider':'Example','type':'SIM','country':'US','activatedAt':'2026-01-01','rules':[{'action':'sms','interval':90,'unit':'days','anchor':'completion','dueDate':'2026-10-07'}]}) self.assertEqual(reply.status_code,201,reply.text) owner = client.get('/api/auth/me').json()['user']['id'] with self.db.connect(write=True) as conn: settings = decode(conn.execute('SELECT settings FROM users WHERE id=?',(owner,)).fetchone()[0]) settings.update(telegram=True,telegramVerified=True,chatId=chat) conn.execute('UPDATE users SET settings=? WHERE id=?',(encode(settings),owner)) async def telegram(transport, method, payload): messages.append((transport.config.telegram_token,method,payload['chat_id'])) return {'message_id':1} now = datetime(2026,10,4,1,0,tzinfo=timezone.utc) self.notifier.enqueue(now) with patch.object(Transport,'telegram',telegram): asyncio.run(self.notifier.deliver(now)) self.assertCountEqual(messages,[(TOKEN_A,'sendMessage','100'),(TOKEN_B,'sendMessage','200')]) self.assertEqual(self.a.get('/api/workspace').json()['workspace']['notifications'][0]['status'],'sent') def test_connection_check_never_sends_a_message_and_redacts_network_errors(self): self.save('telegram', {'token':TOKEN_A}) methods = [] async def telegram(transport, method, payload): methods.append(method) return {'id':123456789,'is_bot':True,'first_name':'Test','username':'AliceTestBot'} with patch.object(Transport,'telegram',telegram): reply = self.a.post('/api/notifications/config/telegram/check',headers=self.headers[0]) self.assertEqual(reply.status_code,200,reply.text) self.assertTrue(reply.json()['checked']) self.assertEqual(reply.json()['botUsername'],'AliceTestBot') self.assertEqual(methods,['getMe']) async def fail(transport, method, payload): raise RuntimeError(TOKEN_A) with patch.object(Transport,'telegram',fail): reply = self.a.post('/api/notifications/config/telegram/check',headers=self.headers[0]) self.assertEqual(reply.status_code,503,reply.text) self.assertNotIn(TOKEN_A,reply.text) self.assertTrue(self.services()['telegram']['configured']) def test_clearing_configuration_only_affects_its_owner(self): self.save('email',SMTP_A) self.save('email',{**SMTP_A,'user':'bob@example.com','from':'bob@example.com'},self.b,self.headers[1]) reply = self.a.delete('/api/notifications/config/email',headers=self.headers[0]) self.assertEqual(reply.status_code,200,reply.text) self.assertFalse(self.services()['email']['configured']) self.assertTrue(self.services(self.b)['email']['configured']) def test_bot_change_during_connection_cannot_create_a_stale_binding(self): self.save('telegram', {'token':TOKEN_A}) async def telegram(transport, method, payload): self.notifier.configs.save(self.people[0],'telegram',TelegramConfigInput(token=TOKEN_B),Config()) return {'id':123456789,'is_bot':True,'first_name':'Test','username':'AliceTestBot'} with patch.object(Transport,'telegram',telegram): reply = self.a.post('/api/notifications/telegram/binding',headers=self.headers[0]) self.assertEqual(reply.status_code,409,reply.text) with self.db.connect() as conn: self.assertEqual(conn.execute('SELECT count(*) FROM bindings WHERE user_id=?',(self.people[0],)).fetchone()[0],0) def test_personal_polling_keeps_separate_offsets_and_failed_bot_does_not_block_others(self): self.save('telegram', {'token':TOKEN_A}) self.save('telegram', {'token':TOKEN_B},self.b,self.headers[1]) async def get_me(transport, method, payload): return {'id':123456789,'is_bot':True,'first_name':'Test','username':'AliceTestBot'} with patch.object(Transport,'telegram',get_me): code = self.a.post('/api/notifications/telegram/binding',headers=self.headers[0]).json()['code'] calls = [] async def updates(transport, method, payload): calls.append((transport.config.telegram_token,payload['offset'])) if transport.config.telegram_token == TOKEN_B: raise RuntimeError(TOKEN_B) return [] if payload['offset'] else [{'update_id':21,'message':{'chat':{'id':12345,'type':'private'},'text':f'/start {code}'}}] with patch.object(Transport,'telegram',updates), self.assertLogs('simkeep.notifications',level='WARNING') as logs: asyncio.run(self.notifier.poll()) asyncio.run(self.notifier.poll()) self.assertCountEqual(calls,[(TOKEN_A,0),(TOKEN_B,0),(TOKEN_A,22),(TOKEN_B,0)]) self.assertTrue(self.a.get('/api/workspace').json()['workspace']['settings']['telegramVerified']) self.assertNotIn(TOKEN_B,' '.join(logs.output)) def test_smtp_connection_check_authenticates_without_sending_email(self): self.save('email',SMTP_A) commands = [] class SMTP: def __init__(self, host, port, **kwargs): commands.append(('connect',host,port)) def __enter__(self): return self def __exit__(self, *args): commands.append(('quit',)) def ehlo_or_helo_if_needed(self): commands.append(('hello',)) def starttls(self, **kwargs): commands.append(('starttls',)) def ehlo(self): commands.append(('hello-tls',)) def login(self, user, password): commands.append(('login',user,password)) def send_message(self, message): raise AssertionError('Connection checking must not send email') with patch('server.notifications.smtplib.SMTP',SMTP): reply = self.a.post('/api/notifications/config/email/check',headers=self.headers[0]) self.assertEqual(reply.status_code,200,reply.text) self.assertTrue(reply.json()['checked']) self.assertEqual(commands,[('connect','smtp.example.com',587),('hello',),('starttls',),('hello-tls',),('login','alice@example.com',' password $ with spaces '),('quit',)]) def test_fake_token_with_same_bot_prefix_cannot_reserve_another_persons_credentials(self): self.save('telegram',{'token':'123456789:'+'X'*35}) self.save('telegram',{'token':TOKEN_A},self.b,self.headers[1]) self.assertTrue(self.services(self.b)['telegram']['configured']) # The actual full credential remains exclusive, including canonical ID spelling. reply = self.a.put('/api/notifications/config/telegram',headers=self.headers[0],json={'token':'000'+TOKEN_A}) self.assertEqual(reply.status_code,409,reply.text) def test_repaired_smtp_retries_failed_reminder_without_resending_a_sent_job(self): self.save('email',SMTP_A) reply = self.a.post('/api/cards',headers=self.headers[0],json={'name':'Personal SIM','provider':'Example','type':'SIM','country':'US','activatedAt':'2026-01-01','rules':[{'action':'sms','interval':90,'unit':'days','anchor':'completion','dueDate':'2026-10-07'}]}) self.assertEqual(reply.status_code,201,reply.text) with self.db.connect(write=True) as conn: settings = decode(conn.execute('SELECT settings FROM users WHERE id=?',(self.people[0],)).fetchone()[0]) settings.update(email=True,emailVerified=True) conn.execute('UPDATE users SET settings=? WHERE id=?',(encode(settings),self.people[0])) now = datetime(2026,10,4,1,0,tzinfo=timezone.utc) self.notifier.enqueue(now) with patch.object(Transport,'smtp',side_effect=RuntimeError('old-password')): for attempt in range(5): asyncio.run(self.notifier.deliver(now+timedelta(minutes=10*attempt))) self.assertEqual(self.a.get('/api/workspace').json()['workspace']['notifications'][0]['status'],'failed') self.save('email',{**SMTP_A,'password':'corrected-password'}) delivered = [] def smtp(transport, destination, text): delivered.append(destination) with patch.object(Transport,'smtp',smtp): self.notifier.enqueue(now+timedelta(minutes=50)) asyncio.run(self.notifier.deliver(now+timedelta(minutes=50))) self.save('email',{**SMTP_A,'password':'another-password'}) self.notifier.enqueue(now+timedelta(minutes=60)) asyncio.run(self.notifier.deliver(now+timedelta(minutes=60))) self.assertEqual(delivered,['alice@example.com']) jobs = self.a.get('/api/workspace').json()['workspace']['notifications'] self.assertEqual(len(jobs),1) self.assertEqual(jobs[0]['status'],'sent') if __name__ == '__main__': unittest.main()