Add attended SMTP verification and exact reader admission

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e6f1-443f-7783-9920-a16b2ffc467f

tests/test_telemetry_smtp_delivery.py

@@ -0,0 +1,40 @@
+import json
+import sys
+from pathlib import Path
+import subprocess
+import unittest
+from unittest.mock import patch
+sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
+import telemetry_smtp_delivery as delivery
+
+class DeliveryTests(unittest.TestCase):
+    def test_valid_entry(self):
+        value=dict(delivery.FIELDS,SMTP_PASSWORD='fixture-password')
+        self.assertEqual(delivery.validate_entry({'data':value}),'fixture-password')
+    def test_password_missing_or_invalid(self):
+        for password in (None,'',' ','bad\npassword'):
+            with self.subTest(password=password),self.assertRaises(delivery.LaneError):
+                delivery.validate_entry({'data':dict(delivery.FIELDS,SMTP_PASSWORD=password)})
+    def test_wrong_identity_or_server_refused(self):
+        for field in ('SMTP_USERNAME','SMTP_HOST','SMTP_FROM','SMTP_STARTTLS'):
+            value=dict(delivery.FIELDS,SMTP_PASSWORD='fixture');value[field]='wrong'
+            with self.subTest(field=field),self.assertRaises(delivery.LaneError):
+                delivery.validate_entry({'data':value})
+    def test_permission_denied_does_not_create_policy(self):
+        with patch.object(delivery,'bao',return_value=subprocess.CompletedProcess([],2,b'',b'permission denied')) as bao:
+            with self.assertRaises(delivery.LaneError):delivery.ensure('policy',{},lambda _:None)
+            self.assertEqual(bao.call_count,1)
+    def test_different_existing_policy_never_overwritten(self):
+        result=subprocess.CompletedProcess([],0,b'{"data":{"policy":"different"}}',b'')
+        with patch.object(delivery,'bao',return_value=result) as bao:
+            with self.assertRaises(delivery.LaneError):
+                delivery.ensure('policy',{'policy':'wanted'},lambda a:delivery.require(a['policy']=='wanted','drift'))
+            self.assertEqual(bao.call_count,1)
+    def test_exact_reader_role(self):
+        role=delivery.role_payload(delivery.LANE)
+        self.assertEqual(role['bound_service_account_names'],['telemetry-smtp-eso'])
+        self.assertEqual(role['bound_service_account_namespaces'],['telemetry'])
+        self.assertEqual(role['audience'],'openbao')
+        self.assertEqual(role['token_explicit_max_ttl'],'15m')
+        self.assertTrue(role['token_no_default_policy'])
+        self.assertNotIn('*',delivery.POLICY)