railiance-platform/tests/test_telemetry_smtp_entry.py
codex 3d43a9b1a3
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
Prepare password-free telemetry SMTP entry for attended custody
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e6f1-443f-7783-9920-a16b2ffc467f
2026-09-28 11:11:34 +02:00

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('smtp_entry',Path(__file__).resolve().parents[1]/'scripts/telemetry_smtp_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/railiance-telemetry/smtp'),
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('SMTP_USERNAME=platform@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)