dtf-system/tests/print_file_test.py

"""Print files and payment issues, against a running stack.

A paid order must get a print file generated from its approved layout, which
the operator can download and approve as the final file without re-uploading
anything. Artwork the generator cannot read goes to hand preparation with a
reason, and a paid notification that did not become an order stays on the
Kanban until someone records what was done.

Run inside the API container (it needs Pillow): python -m tests.print_file_test
"""
import io
import re
import time
from urllib.request import urlopen
from uuid import uuid4

from PIL import Image

from tests.payment_test import deliver
from tests.smoke_test import Client, upload_bytes

PT_PER_CM = 72 / 2.54
CUSTOMER = {'cnpj': '11222333000181', 'zap': '11999999999', 'mail': 'print-test@example.test'}


def artwork(kind='PNG'):
    image = Image.new('RGBA', (600, 300), (0, 0, 0, 0))
    for x in range(40, 560):
        for y in range(40, 260):
            image.putpixel((x, y), (220, 30, 60, 255))
    out = io.BytesIO()
    if kind == 'PDF':
        image.convert('RGB').save(out, 'PDF', resolution=72)
    else:
        image.save(out, kind)
    return out.getvalue()


def loose_item(uid, copies=2):
    """Two copies of a 20 x 10 cm artwork side by side on 57 cm film."""
    placements = [{'source_index': 0, 'copy_index': i, 'x_cm': 20 * i, 'y_cm': 0,
                   'width_cm': 20, 'length_cm': 10, 'rotation_degrees': 0, 'mirrored': False}
                  for i in range(copies)]
    return {'mode': 'avulsa', 'metres': '0.1', 'grade': 90, 'uploads': [uid],
            'production': {'version': 2, 'film_width_cm': 57, 'height_cm': 10,
                           'sources': [{'upload_id': uid, 'kind': 'artwork', 'width_cm': 20,
                                        'length_cm': 10, 'copies': copies, 'rotation_degrees': 0,
                                        'mirrored': False, 'measurement': 'file'}],
                           'placements': placements},
            'quality_status': 'ok', 'quality_acknowledged': False}


def approved_quote(client, item):
    quote = client.call('/quotes', {'request_key': str(uuid4()), 'customer': CUSTOMER,
                                    'items': [item], 'freight': {'service': 'pickup'}})
    approved = client.call('/operator/quotes/' + quote['id'] + '/approve', {'items': [item]}, operator=True)
    return quote['id'], approved['total_cents']


def paid_order(client, item):
    quote_id, _ = approved_quote(client, item)
    return client.call('/orders/dev-paid', {'quote_id': quote_id})


def wait_print(client, oid, wanted):
    deadline = time.monotonic() + 90
    while time.monotonic() < deadline:
        order = next(o for o in client.call('/operator/board', operator=True)['orders'] if o['id'] == oid)
        rows = order['print_files']
        if rows and rows[0]['status'] in ('ready', 'manual', 'failed'):
            assert rows[0]['status'] == wanted, rows
            return order, rows[0]
        time.sleep(1)
    raise AssertionError('Print file was not generated in time')


def run():
    client = Client()
    client.call('/session')

    uid = upload_bytes(client, artwork(), name='LOCAL-PRINT-TEST.png')
    order = paid_order(client, loose_item(uid))
    order, row = wait_print(client, order['id'], 'ready')
    assert row['detail']['placements'] == 2 and row['detail']['min_dpi'] == round(600 / (20 / 2.54))
    link = client.call('/operator/uploads/' + str(row['upload_id']) + '/download', operator=True)
    with urlopen(link['url'], timeout=30) as response:
        pdf = response.read()
    assert pdf.startswith(b'%PDF-') and pdf.rstrip().endswith(b'%%EOF')
    width, height = map(float, re.search(rb'/MediaBox \[0 0 ([\d.]+) ([\d.]+)\]', pdf).groups())
    assert abs(width - 57 * PT_PER_CM) < 0.01 and abs(height - 10 * PT_PER_CM) < 0.01, (width, height)
    print('PASS: paid order generated a 57 x 10 cm print file with both copies')

    # The operator approves the generated file as the final one, with no upload,
    # and the order can then enter the print queue.
    oid = order['id']
    result = client.call('/operator/orders/' + oid + '/final-files',
                         {'version': order['version'], 'files': [{'item_index': 0, 'upload_id': str(row['upload_id'])}],
                          'note': 'Generated print file checked'}, operator=True)
    version = result['version']
    for state in ('tra', 'fil'):
        client.call('/operator/orders/' + oid + '/move', {'state': state, 'version': version}, operator=True)
        version += 1
    # Once queued for printing, the final set can no longer change.
    client.call('/operator/orders/' + oid + '/final-files',
                {'version': version, 'files': [{'item_index': 0, 'upload_id': str(row['upload_id'])}],
                 'note': 'again'}, operator=True, expected=409)
    print('PASS: generated file approved as final without re-uploading; order queued for printing')

    # A single-page PDF is placed as a vector form, not rasterised.
    pdf_upload = upload_bytes(client, artwork('PDF'), name='LOCAL-PRINT-TEST.pdf')
    pdf_order = paid_order(client, loose_item(pdf_upload))
    _, pdf_row = wait_print(client, pdf_order['id'], 'ready')
    assert pdf_row['detail']['vector_sources'] == 1 and pdf_row['detail']['min_dpi'] is None, pdf_row
    link = client.call('/operator/uploads/' + str(pdf_row['upload_id']) + '/download', operator=True)
    with urlopen(link['url'], timeout=30) as response:
        generated = response.read()
    assert b'/Subtype /Form' in generated or b'/Subtype/Form' in generated
    print('PASS: PDF artwork generated as a vector print file')

    # A customer cannot see or reuse another order's generated file.
    other = Client()
    other.call('/session')
    other.call('/uploads/' + str(row['upload_id']), expected=404)

    # Artwork the generator cannot read goes to hand preparation, with a reason.
    manual = upload_bytes(client, b'LOCAL PRINT TEST - NOT AN IMAGE', name='LOCAL-PRINT-TEST.cdr')
    order = paid_order(client, loose_item(manual, copies=1))
    order, row = wait_print(client, order['id'], 'manual')
    assert 'não é uma imagem' in row['detail']['reason'], row
    rows = client.call('/operator/orders/' + order['id'] + '/print-files', {}, operator=True)
    assert rows[0]['status'] == 'pending'
    wait_print(client, order['id'], 'manual')
    print('PASS: unreadable artwork is routed to hand preparation and can be retried')

    # A signed, paid notification for the wrong amount does not become an order;
    # it waits on the Kanban until an operator records the resolution.
    # Same client, so the same operator session: operator logins share a
    # 10-per-15-minutes account limit with every other suite in the run.
    quote_id, total = approved_quote(client, loose_item(uid, copies=1))
    event_id = 'print-test-underpaid-' + uuid4().hex
    outcome = deliver({'event_id': event_id, 'reference': quote_id,
                       'status': 'approved', 'amount_cents': total - 1})
    assert outcome['outcome'].startswith('refused'), outcome
    issues = client.call('/operator/board', operator=True)['payment_issues']
    issue = next(i for i in issues if i['event_id'] == event_id)
    client.call('/operator/payment-events/' + issue['id'] + '/resolve', {'note': 'no'},
                operator=True, expected=422)
    client.call('/operator/payment-events/' + issue['id'] + '/resolve',
                {'note': 'Local test: refunded the underpayment'}, operator=True)
    client.call('/operator/payment-events/' + issue['id'] + '/resolve',
                {'note': 'Local test: second resolution'}, operator=True, expected=404)
    issues = client.call('/operator/board', operator=True)['payment_issues']
    assert not any(i['event_id'] == event_id for i in issues)
    print('PASS: refused paid notification is listed until an operator resolves it')


if __name__ == '__main__':
    run()