01e0576132
Sibling of the telemetry SMTP entry for library@coulomb.social (inbound-library ILIB-WP-0002-T07). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 1170398@bnt-lap001 Assistant-Session: 622cd705-cb31-4322-ba08-dd42e44fa583
32 lines
1.8 KiB
Python
32 lines
1.8 KiB
Python
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import unittest
|
|
|
|
spec=importlib.util.spec_from_file_location('imap_entry',Path(__file__).resolve().parents[1]/'scripts/inbound_library_imap_entry.py')
|
|
entry=importlib.util.module_from_spec(spec);spec.loader.exec_module(entry)
|
|
|
|
class EntryTests(unittest.TestCase):
|
|
def test_create_without_password_and_verify(self):
|
|
calls=[]
|
|
results=[subprocess.CompletedProcess([],2,b'',b'No value found at platform/metadata/workloads/inbound-library/imap'),
|
|
subprocess.CompletedProcess([],0,b'{"data":{"version":1}}',b''),
|
|
subprocess.CompletedProcess([],0,b'{"data":{"current_version":1}}',b'')]
|
|
def invoke(args): calls.append(args);return results.pop(0)
|
|
self.assertEqual(entry.run(invoke),0)
|
|
self.assertIn('-cas=0',calls[1]);self.assertFalse(any('PASSWORD' in a for a in calls[1]))
|
|
self.assertIn('IMAP_USERNAME=library@coulomb.social',calls[1])
|
|
def test_existing_password_version_never_read_or_overwritten(self):
|
|
calls=[]
|
|
def invoke(args):
|
|
calls.append(args);return subprocess.CompletedProcess([],0,b'{"data":{"current_version":2}}',b'')
|
|
self.assertEqual(entry.run(invoke),0);self.assertEqual(len(calls),1)
|
|
self.assertEqual(calls[0][:3],['kv','metadata','get'])
|
|
def test_permission_error_never_triggers_write(self):
|
|
calls=[]
|
|
def invoke(args): calls.append(args);return subprocess.CompletedProcess([],2,b'',b'permission denied')
|
|
self.assertEqual(entry.run(invoke),21);self.assertEqual(len(calls),1)
|
|
def test_cas_conflict_is_not_success(self):
|
|
results=[subprocess.CompletedProcess([],2,b'',b'No value found at x'),subprocess.CompletedProcess([],2,b'',b'CAS mismatch')]
|
|
self.assertEqual(entry.run(lambda _:results.pop(0)),22)
|