33 lines
1.8 KiB
Python
33 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)
|