dtf-system/tests/kanban_test.py

"""Kanban artwork pictures, quote filters and the search over orders and quotes.

Run against the local stack: python3 -m tests.kanban_test
"""
import base64
import secrets
from urllib.error import HTTPError
from urllib.parse import urlencode
from urllib.request import Request
from uuid import uuid4

from psycopg.types.json import Jsonb

from app.core import db
from tests.smoke_test import BASE, Client, with_host

# A 1 x 1 PNG: the endpoint reads the type from the bytes.
PNG = base64.b64decode('iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==')


def raw(client, method, path, data=None, content_type=None):
    """A request whose body is not JSON; returns (status, headers, body)."""
    headers = with_host({'Content-Type': content_type} if content_type else {})
    request = Request(BASE + '/api' + path, data=data, headers=headers, method=method)
    try:
        with client.opener.open(request, timeout=30) as response:
            return response.status, response.headers, response.read()
    except HTTPError as exc:
        return exc.code, exc.headers, exc.read()


def session_upload(client, name):
    """A completed, clean upload owned by this client's session. The bytes are
    not the subject here, so none are stored."""
    sid = next(c.value for c in client.jar if c.name == 'dtf_session')
    uid = uuid4()
    with db.connect() as c:
        owner = c.execute('SELECT owner FROM dtf_local.sessions WHERE id=%s', (sid,)).fetchone()['owner']
        c.execute('''INSERT INTO dtf_local.uploads(id,owner,name,size,object_key,multipart_id,complete,expires_at,scan_state)
            VALUES(%s,%s,%s,1,%s,'none',true,now()+interval '1 day','clean')''', (uid, owner, name, f'originals/{uid}'))
    return str(uid)


def run():
    tag = secrets.token_hex(4)
    mail = f'kanban-{tag}@example.test'
    cnpj = '11.222.333/0001-81'
    customer = {'mail': mail, 'cnpj': cnpj, 'zap': '(16) 98765-' + str(4000 + int(tag[:3], 16) % 1000)}
    client = Client()
    client.call('/session')
    uid = session_upload(client, f'kanban-{tag}.cdr')

    # The customer's browser sends the picture of its own upload; nothing else.
    assert raw(client, 'PUT', f'/uploads/{uid}/thumbnail', PNG, 'image/png')[0] == 200
    assert raw(client, 'PUT', f'/uploads/{uid}/thumbnail', b'<svg onload=alert(1)>', 'image/png')[0] == 415
    assert raw(client, 'PUT', f'/uploads/{uid}/thumbnail', PNG + b'\0' * 300_001, 'image/png')[0] == 413
    stranger = Client()
    stranger.call('/session')
    assert raw(stranger, 'PUT', f'/uploads/{uid}/thumbnail', PNG, 'image/png')[0] == 404

    # Only an operator reads it, as an image the browser may keep for a while.
    assert raw(client, 'GET', f'/operator/uploads/{uid}/thumbnail')[0] == 401
    client.call('/operator/board', operator=True)
    status, headers, body = raw(client.operator_client, 'GET', f'/operator/uploads/{uid}/thumbnail')
    assert status == 200 and body == PNG, status
    assert headers['Content-Type'] == 'image/png' and 'private' in headers['Cache-Control']
    assert headers['X-Content-Type-Options'] == 'nosniff'

    owner = uuid4()
    pending, approved, paid, order_id = uuid4(), uuid4(), uuid4(), uuid4()
    item = lambda mode, warning=False: {'mode': mode, 'uploads': [uid], 'metres': '1.00', 'billed_metres': '1.00',
                                        'grade': 100, 'quality_status': 'warning' if warning else 'ok'}
    draft = lambda *items: {'customer': customer, 'items': list(items), 'freight': {'service': 'pickup'}}
    try:
        with db.connect() as c:
            for qid, content, accepted in [(pending, draft(item('avulsa', warning=True)), None),
                                           (approved, draft(item('uvfile')), {'items': [], 'total_cents': 6990}),
                                           (paid, draft(item('file')), {'items': [], 'total_cents': 1490})]:
                c.execute('''INSERT INTO dtf_local.quotes(id,owner,request_key,request_hash,draft,approved)
                    VALUES(%s,%s,%s,%s,%s,%s)''', (qid, owner, uuid4(), 'kanban-fixture', Jsonb(content),
                                                   Jsonb(accepted) if accepted else None))
            number = c.execute('''INSERT INTO dtf_local.orders(id,quote_id,owner,snapshot,payment)
                VALUES(%s,%s,%s,%s,%s) RETURNING number''',
                (order_id, paid, owner, Jsonb({**draft(item('file')), 'total_cents': 1490}),
                 Jsonb({'provider': 'teste', 'id': f'kanban-{tag}'}))).fetchone()['number']

        def quotes(**params):
            page = client.call('/operator/quotes?' + urlencode({'q': mail, **params}), operator=True)
            return page, {q['id'] for q in page['quotes']}

        page, ids = quotes(kind='pending')
        assert ids == {str(pending)} and page['total'] == 1, ids
        assert str(uid) in page['thumbnails']
        assert quotes(kind='pending', layout='avulsa', flag='ressalva')[1] == {str(pending)}
        assert quotes(kind='pending', product='uv')[1] == set()
        assert quotes(kind='approved', product='uv', layout='folha')[1] == {str(approved)}
        assert quotes(kind='approved', layout='avulsa')[1] == set()
        # CNPJ and WhatsApp match by digits, whatever the punctuation typed.
        assert client.call('/operator/quotes?' + urlencode({'kind': 'approved', 'q': '11222333'}),
                           operator=True)['total'] >= 1
        # LIKE's own wildcards are literal: "%%" is not "everything".
        assert str(pending) not in {q['id'] for q in client.call(
            '/operator/quotes?' + urlencode({'kind': 'pending', 'q': '%%'}), operator=True)['quotes']}

        # One search: the paid order (by customer and by number) and the two unpaid quotes.
        found = client.call('/operator/search?' + urlencode({'q': mail}), operator=True)
        assert {o['id'] for o in found['orders']} == {str(order_id)}
        assert {q['id'] for q in found['quotes']} == {str(pending), str(approved)}
        assert str(uid) in found['thumbnails']
        by_number = client.call('/operator/search?' + urlencode({'q': f'#{number}'}), operator=True)
        assert str(order_id) in {o['id'] for o in by_number['orders']}
        assert client.call('/operator/search?q=a', operator=True, expected=422)

        board = client.call('/operator/board', operator=True)
        assert str(uid) in board['thumbnails']
        print('PASS: artwork pictures (owner only, images only), quote filters and the order/quote search')
    finally:
        with db.connect() as c:
            c.execute('DELETE FROM dtf_local.orders WHERE id=%s', (order_id,))
            c.execute('DELETE FROM dtf_local.quotes WHERE id=ANY(%s)', ([pending, approved, paid],))
            c.execute('DELETE FROM dtf_local.upload_thumbnails WHERE upload_id=%s', (uid,))
            c.execute('DELETE FROM dtf_local.uploads WHERE id=%s', (uid,))


if __name__ == '__main__':
    run()