RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
75 lines
4.5 KiB
Python
75 lines
4.5 KiB
Python
import os,sys,tempfile,json,unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch,MagicMock
|
|
|
|
BASE=Path(__file__).resolve().parents[1]
|
|
sys.path[:0]=[str(BASE/'bot-utility'),str(BASE)]
|
|
import mail
|
|
from tg_channel import TGChannel
|
|
|
|
|
|
class MigrationTests(unittest.TestCase):
|
|
def test_mail_session_requires_tls(self):
|
|
with patch.dict(os.environ,{'MAIL_COM_URL':'http://remote.example/admin/com'}):
|
|
with self.assertRaises(RuntimeError):
|
|
with mail._ComSession():pass
|
|
|
|
def test_mail_client_request_and_close(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
p=Path(d)/'token';p.write_text('t'*128)
|
|
with patch.dict(os.environ,{'MAIL_COM_TOKEN_FILE':str(p)}),patch('requests.Session') as factory:
|
|
factory.return_value.post.return_value.json.return_value={'items':[{'id':42}]}
|
|
with mail._ComSession() as session:
|
|
result=mail._list_inbox_messages(session,'a@rrrosmontis.icu',limit=5)
|
|
self.assertEqual(result,[{'id':42}])
|
|
self.assertFalse(factory.return_value.trust_env)
|
|
self.assertEqual(factory.return_value.post.call_args.kwargs['json']['limit'],5)
|
|
factory.return_value.close.assert_called_once()
|
|
|
|
def test_body_attachment_metadata_does_not_open_local_eml(self):
|
|
app=MagicMock()
|
|
app.post.return_value={'message':{'ID':8,'Subject':'s','From':'a@b.c','Date':'d','Body':'text','HTMLBody':'<p>body</p><script>bad()</script>','image_count':2,'attachment_names':['a.pdf']}}
|
|
m=mail._find_message_by_id(app,['a@rrrosmontis.icu'],8)
|
|
self.assertEqual(mail._analyze_message(m),(2,['a.pdf']))
|
|
self.assertEqual(mail._extract_body(m),'body')
|
|
|
|
def test_initial_poll_keeps_com_id_and_does_not_push_history(self):
|
|
with tempfile.TemporaryDirectory() as d,patch.object(mail,'MAIL_DIR',d),patch.object(mail,'_ComSession'),patch.object(mail,'_mail_changes',return_value={'max_id':2386,'items':[]}):
|
|
(Path(d)/'9000000001.json').write_text(json.dumps({'addresses':['a@rrrosmontis.icu'],'push':{'enabled':True,'last_seen':{}}}))
|
|
mail._poll_next.clear()
|
|
with patch.object(mail.notification_queue,'put') as put:
|
|
mail._poll_once();put.assert_not_called()
|
|
self.assertEqual(mail._load_user(9000000001)['push']['last_seen']['a@rrrosmontis.icu'],2386)
|
|
|
|
def test_incremental_poll_queues_without_advancing_watermark(self):
|
|
addr='a@rrrosmontis.icu'
|
|
with tempfile.TemporaryDirectory() as d,patch.object(mail,'MAIL_DIR',d),patch.object(mail,'_ComSession'),patch.object(mail.identity,'push_targets',return_value=[('tg',1)]):
|
|
(Path(d)/'9000000001.json').write_text(json.dumps({'addresses':[addr],'push':{'enabled':True,'last_seen':{addr:10}}}))
|
|
mail._poll_next.clear();mail._inflight.clear()
|
|
changes={'max_id':99,'items':[{'id':12,'subject':'s','from':'a','from_address':'a@b.c','ts':1},{'id':14,'subject':'s','from':'a','from_address':'a@b.c','ts':1}]}
|
|
with patch.object(mail,'_mail_changes',return_value=changes),patch.object(mail.notification_queue,'put') as put:
|
|
mail._poll_once()
|
|
self.assertEqual([c.args[0]['mid'] for c in put.call_args_list],[12,14])
|
|
self.assertEqual(mail._load_user(9000000001)['push']['last_seen'][addr],10)
|
|
mail._inflight.clear()
|
|
|
|
def test_oss_uris_and_traversal(self):
|
|
self.assertEqual(TGChannel._translate_oss_path('file:///C:/oss/pixiv/a%20b.jpg'),'/mnt/oss/pixiv/a b.jpg')
|
|
self.assertEqual(TGChannel._translate_oss_path('file:///mnt/A/pixiv/a.jpg'),'/mnt/oss/pixiv/a.jpg')
|
|
with self.assertRaises(ValueError):TGChannel._translate_oss_path('file:///mnt/A/../secret')
|
|
|
|
def test_no_windows_admin_secret_or_windows_paths(self):
|
|
self.assertIsNone(mail.HMAIL_ADMIN_PASSWORD)
|
|
for name in ['jm','pixiv','setu']:
|
|
module=__import__(name)
|
|
# Runtime proxy config is private and intentionally absent in a clone.
|
|
configured = os.environ.get('BOT_PROXY_CONFIG', str(BASE/'bot-utility'/'proxies.json'))
|
|
self.assertEqual(Path(module.PROXY_CONFIG_PATH), Path(configured))
|
|
self.assertNotIn('C:',module.PROXY_CONFIG_PATH)
|
|
import pixiv,setu,for_admin_komari_agent
|
|
self.assertEqual(pixiv.OSS_SAVE_DIR,setu.OSS_SAVE_DIR)
|
|
self.assertEqual(for_admin_komari_agent.KOMARI_API_BASE,'https://status.rrrosmontis.icu')
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|