Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a10e2d-a5ef-7e92-b03d-b1694e915680
50 lines
2.4 KiB
Python
50 lines
2.4 KiB
Python
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
spec = importlib.util.spec_from_file_location("authelia_delivery", ROOT / "scripts/authelia_smtp_delivery.py")
|
|
delivery = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(delivery)
|
|
|
|
|
|
class AutheliaAdmissionTests(unittest.TestCase):
|
|
def test_role_cannot_be_used_by_telemetry_or_another_namespace(self):
|
|
role = delivery.lane.role_payload(delivery.lane.LANE)
|
|
self.assertEqual(role["bound_service_account_names"], ["authelia-smtp-eso"])
|
|
self.assertEqual(role["bound_service_account_namespaces"], ["sso"])
|
|
self.assertEqual(role["audience"], "openbao")
|
|
self.assertEqual(role["token_explicit_max_ttl"], "15m")
|
|
self.assertTrue(role["token_no_default_policy"])
|
|
self.assertNotIn("*", delivery.lane.POLICY)
|
|
self.assertNotIn('["update"]', delivery.lane.POLICY.split('auth/token')[0])
|
|
|
|
def test_different_role_or_policy_is_not_overwritten(self):
|
|
response = subprocess.CompletedProcess([], 0, b'{"data":{"policy":"different"}}', b'')
|
|
with patch.object(delivery.lane, "bao", return_value=response) as bao:
|
|
with self.assertRaises(delivery.lane.LaneError):
|
|
delivery.lane.ensure("policy", {"policy": delivery.lane.POLICY},
|
|
lambda actual: delivery.lane.require(actual["policy"] == delivery.lane.POLICY, "drift"))
|
|
self.assertEqual(bao.call_count, 1)
|
|
|
|
def test_existing_mailbox_identity_and_password_are_verified(self):
|
|
wanted = dict(delivery.lane.FIELDS, SMTP_PASSWORD="synthetic-fixture")
|
|
self.assertEqual(delivery.lane.validate_entry({"data": wanted}), "synthetic-fixture")
|
|
for field in ["SMTP_USERNAME", "SMTP_HOST", "SMTP_FROM", "SMTP_PASSWORD"]:
|
|
bad = dict(wanted, **{field: ""})
|
|
with self.subTest(field=field), self.assertRaises(delivery.lane.LaneError):
|
|
delivery.lane.validate_entry({"data": bad})
|
|
|
|
def test_reusing_lane_does_not_change_telemetry_admission(self):
|
|
import telemetry_smtp_delivery
|
|
self.assertEqual(telemetry_smtp_delivery.LANE["namespace"], "telemetry")
|
|
self.assertEqual(telemetry_smtp_delivery.LANE["service_account"], "telemetry-smtp-eso")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|