multischleuder/test/report.py

#!/usr/bin/env python3

import json
import mailbox
import os
import subprocess

import deepdiff
import pgpy


def subscribermap(subs):
    return {
        s: pgpy.PGPKey.from_file(f'/tmp/{s}.asc')[0].fingerprint.replace(' ', '')
        for s in subs
    }


def keymap(subs):
    return {
        pgpy.PGPKey.from_file(f'/tmp/{s}.asc')[0].fingerprint.replace(' ', ''): s
        for s in subs
    }

# Test subscribers

subsp = subprocess.Popen(['/usr/bin/schleuder-cli', 'subscriptions', 'list', 'test-global@schleuder.example.org'],
                         stdout=subprocess.PIPE, stderr=subprocess.PIPE)
subsout, _ = subsp.communicate()
keysp = subprocess.Popen(['/usr/bin/schleuder-cli', 'keys', 'list', 'test-global@schleuder.example.org'],
                         stdout=subprocess.PIPE, stderr=subprocess.PIPE)
keysout, _ = keysp.communicate()
fprp = subprocess.Popen(['/usr/bin/schleuder-cli', 'lists', 'show', 'test-global@schleuder.example.org', 'fingerprint'],
                         stdout=subprocess.PIPE, stderr=subprocess.PIPE)
fprout, _ = fprp.communicate()

expected_subscribers = subscribermap([
    'aaron.example@example.org',
    'admin@example.org',
    'alex.example@example.org',
    'amy.example@example.org',
    'andy.example@example.org',
    'anna.example@example.org',
    'anotherspammer@example.org'
])

actual_subscribers = {}
for s in subsout.decode().splitlines():
    if '\t' not in s:
        continue
    uid, fpr, *_ = s.split('\t', 2)
    actual_subscribers[uid] = fpr

subsdiff = deepdiff.DeepDiff(expected_subscribers, actual_subscribers)
if len(subsdiff) > 0:
    print(subsdiff)
    exit(1)

# Test keys

expected_keys = keymap([
    'aaron.example@example.org',
    'admin@example.org',
    'alex.example@example.org',
    'amy.example@example.org',
    'andy.example@example.org',
    'anna.example@example.org',
    'anotherspammer@example.org'
])
expected_keys[fprout.strip().decode()] = 'test-global@schleuder.example.org'

actual_keys = {}
for s in keysout.decode().splitlines():
    if ' ' not in s:
        continue
    fpr, uid = s.split(' ', 1)
    actual_keys[fpr] = uid

keysdiff = deepdiff.DeepDiff(expected_keys, actual_keys)
if len(keysdiff) > 0:
    print(keysdiff)
    exit(1)

# Test mbox

mbox = mailbox.mbox('/var/spool/mail/root')
if len(mbox) != 2:
    print(f'Expected 2 messages in mbox, got {len(mbox)}')
    exit(1)
_, msg1 = mbox.popitem()
_, msg2 = mbox.popitem()
mbox.close()
if 'X-MultiSchleuder-Digest' not in msg1:
    msg1, msg2 = msg2, msg1


if 'X-MultiSchleuder-Digest' not in msg1:
    print(f'Key conflict message should have a X-MultiSchleuder-Digest header, missing')
    exit(1)
digest = msg1['X-MultiSchleuder-Digest'].strip()
if msg1['From'] != 'test-global-owner@schleuder.example.org':
    print(f'Expected "From: test-global-owner@schleuder.example.org", got {msg1["From"]}')
    exit(1)
if msg1['To'] != 'andy.example@example.org':
    print(f'Expected "To: andy.example@example.org", got {msg1["To"]}')
    exit(1)
if msg1['Auto-Submitted'] != 'auto-generated':
    print(f'Expected "Auto-Submitted: auto-generated", got {msg1["Auto-Submitted"]}')
    exit(1)
if msg1['Precedence'] != 'list':
    print(f'Expected "Precedence: list", got {msg1["Precedence"]}')
    exit(1)

if 'X-MultiSchleuder-Digest' in msg2:
    print(f'Admin report message should not have a X-MultiSchleuder-Digest header, got {msg2["X-MultiSchleuder-Digest"]}')
    exit(1)
if msg2['From'] != 'test-global-owner@schleuder.example.org':
    print(f'Expected "From: test-global-owner@schleuder.example.org", got {msg2["From"]}')
    exit(1)
if msg2['To'] != 'admin@example.org':
    print(f'Expected "To: admin@example.org", got {msg2["To"]}')
    exit(1)
if msg2['Auto-Submitted'] != 'auto-generated':
    print(f'Expected "Auto-Submitted: auto-generated", got {msg2["Auto-Submitted"]}')
    exit(1)
if msg2['Precedence'] != 'list':
    print(f'Expected "Precedence: list", got {msg2["Precedence"]}')
    exit(1)

gpg1 = subprocess.Popen(['/usr/bin/gpg', '-d'],
                        stdin=subprocess.PIPE,
                        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
gpg1o, _ = gpg1.communicate(msg1.get_payload()[1].get_payload(decode=True))
print(f'Key conflict message (decrypted):\n{gpg1o.decode()}')
gpg2 = subprocess.Popen(['/usr/bin/gpg', '-d'],
                        stdin=subprocess.PIPE,
                        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
gpg2o, _ = gpg2.communicate(msg2.get_payload()[1].get_payload(decode=True))
print(f'Admin report message (decrypted):\n{gpg2o.decode()}')

# Test conflict statefile

with open('/tmp/conflict.json', 'r') as f:
    conflict = json.load(f)
if len(conflict) != 1:
    print('Expected 1 entry in conflict statefile, got:')
    print(json.dumps(conflict))
    exit(1)
if digest not in conflict:
    print(f'Expected key "{digest}" in conflict statefile, got:')
    print(json.dumps(conflict))
    exit(1)

print('All checks passed!')