finoex/tests/test_script_order_confirmation_mail_parser.py

import pytest

import dingguo.parser
import email
import os
import shutil
import subprocess
import yaml

project_root_path = os.path.realpath(os.path.join(__file__, '..', '..'))
script_path = os.path.join(project_root_path, 'scripts', 'order-confirmation-mail-parser')
test_data_path = os.path.join(project_root_path, 'tests', 'data')

def get_mail_path_single_order_platform_a():
    return os.path.join(test_data_path, 'amazon', 'mail_1.eml')

def get_mail_path_single_order_platform_b():
    return os.path.join(test_data_path, 'banggood', '1.eml')

def get_mail_path_two_orders():
    return os.path.join(test_data_path, 'amazon', 'mail_2.eml')

@pytest.mark.parametrize('mail_path', [
    get_mail_path_single_order_platform_a(),
    get_mail_path_two_orders(),
    ])
def test_stdin(mail_path):
    process = subprocess.Popen([script_path], stdin = subprocess.PIPE, stdout = subprocess.PIPE)
    with open(mail_path, 'r') as mail_file:
        script_stdout, script_stderr = process.communicate(input = mail_file.read())
    parsed_orders = yaml.load(script_stdout)
    with open(mail_path.replace('.eml', '.yml'), 'r') as expected_result_file:
        expected_orders = yaml.load(expected_result_file.read())
    assert expected_orders == parsed_orders

@pytest.mark.parametrize('mail_paths', [
    [
        get_mail_path_single_order_platform_a(),
        ],
    [
        get_mail_path_two_orders(),
        ],
    [
        get_mail_path_single_order_platform_a(),
        get_mail_path_two_orders(),
        ],
    ])
def test_path(mail_paths):
    parsed_orders = yaml.load(subprocess.check_output([script_path] + mail_paths))
    expected_orders = []
    for mail_path in mail_paths:
        with open(mail_path.replace('.eml', '.yml'), 'r') as expected_result_file:
            expected_orders += yaml.load(expected_result_file.read())
    assert expected_orders == parsed_orders

@pytest.mark.parametrize('mail_paths', [
    [
        get_mail_path_single_order_platform_a(),
        get_mail_path_single_order_platform_b(),
        get_mail_path_two_orders(),
        ],
    ])
def test_catalogue(tmpdir, mail_paths):
    os.chdir(tmpdir.strpath)
    for mail_index, mail_path in enumerate(mail_paths):
        shutil.copyfile(mail_path, '%d.eml' % mail_index)
    assert len(os.listdir('.')) == len(mail_paths)
    orders = yaml.load(subprocess.check_output(
        [script_path, '--catalogue'] + os.listdir('.'),
        ))
    assert set(os.listdir('.')) == set([o.platform for o in orders])
    for dir_name in os.listdir('.'):
        assert set(os.listdir(dir_name)) \
                == set([o.order_id for o in orders if o.platform == dir_name])

@pytest.mark.parametrize('mail_paths', [
    [
        get_mail_path_single_order_platform_a(),
        ],
    [
        get_mail_path_two_orders(),
        ],
    [
        get_mail_path_single_order_platform_a(),
        get_mail_path_two_orders(),
        ],
    ])
def test_register_create(tmpdir, mail_paths):
    os.chdir(tmpdir.strpath)
    subprocess.check_output(
        [script_path, '--register', 'registry.yml'] + mail_paths,
        )
    with open('registry.yml') as registry_file:
        registry = yaml.load(registry_file.read())
    orders = []
    for mail_path in mail_paths:
        with open(mail_path) as mail:
            orders += dingguo.parser.parse_order_confirmation_mail(
                    email.message_from_file(mail)
                    )
    for order in orders:
        assert order == registry.registry[order.platform][order.order_id]
    for platform in registry.registry:
        for order_id in registry.registry[platform]:
            assert registry.registry[platform][order_id] in orders

@pytest.mark.parametrize('pre_mail_paths,post_mail_paths', [
    (
        [get_mail_path_single_order_platform_a()], 
        [get_mail_path_single_order_platform_b()],
        ),
    (
        [get_mail_path_single_order_platform_a(), get_mail_path_single_order_platform_b()], 
        [get_mail_path_two_orders()],
        ),
    ])
def test_register_update(tmpdir, pre_mail_paths, post_mail_paths):
    os.chdir(tmpdir.strpath)
    subprocess.check_output(
        [script_path, '--register', 'registry.yml'] + pre_mail_paths,
        )
    subprocess.check_output(
        [script_path, '--register', 'registry.yml'] + post_mail_paths,
        )
    with open('registry.yml') as registry_file:
        registry = yaml.load(registry_file.read())
    orders = []
    for mail_path in (pre_mail_paths + post_mail_paths):
        with open(mail_path) as mail:
            orders += dingguo.parser.parse_order_confirmation_mail(
                    email.message_from_file(mail)
                    )
    for order in orders:
        assert order == registry.registry[order.platform][order.order_id]
    for platform in registry.registry:
        for order_id in registry.registry[platform]:
            assert registry.registry[platform][order_id] in orders