56fcc4c921
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e6f1-443f-7783-9920-a16b2ffc467f
40 lines
2.3 KiB
Python
40 lines
2.3 KiB
Python
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)
|