"""Exercise the real application with isolated data; never send notifications."""
import argparse
from datetime import date
import io
import os
from pathlib import Path
import re
import secrets
import sys
import tempfile
import threading
import urllib.request


def run(source, artifacts=None):
    import openpyxl
    from PIL import Image
    from waitress import create_server
    from werkzeug.security import generate_password_hash, check_password_hash

    password = 'Win7-Test-' + secrets.token_hex(10)
    hashed = generate_password_hash(password)
    assert check_password_hash(hashed, password), 'Password hashing failed'
    # Also test the scrypt parameters generated by newer Windows deployments.
    assert check_password_hash(generate_password_hash(password, method='scrypt:32768:8:1'), password)
    Image.new('RGB', (3, 3)).save(io.BytesIO(), format='PNG')

    with tempfile.TemporaryDirectory(prefix='po-win7-check-') as temporary:
        os.environ['PO_DATA_DIR'] = temporary
        os.environ.pop('PO_SECRET_KEY', None)
        os.environ.pop('PO_HTTPS', None)
        sys.path.insert(0, str(source.resolve()))
        import app as module
        from models import User
        module.app.config['TESTING'] = True
        assert (Path(temporary)/'instance/setup.token').is_file()
        with module.Session() as session:
            for name, role in [('maker', 'maker'), ('admin', 'admin')]:
                session.add(User(username=name, name=name, role=role, active=True,
                                 must_change_password=False, session_version=1, password_hash=hashed))
            session.commit()

        def post(client, path, data):
            client.get('/login')
            with client.session_transaction() as session:
                token = session['csrf_token']
            return client.post(path, data=dict(data, csrf_token=token))

        clients = {}
        for name in ('maker', 'admin'):
            client = module.app.test_client()
            assert post(client, '/login', {'username': name, 'password': password}).status_code == 302
            clients[name] = client

        def version(oid):
            with module.Session() as session:
                return session.get(module.Order, oid).row_version

        def action(actor, oid, endpoint, **fields):
            fields['row_version'] = str(version(oid))
            return post(clients[actor], '/orders/%d/%s' % (oid, endpoint), fields)

        try:
            for kind in ('purchase', 'outsourcing'):
                response = post(clients['maker'], '/orders/new',
                                {'doc_type': kind, 'request_key': secrets.token_hex(16)})
                assert response.status_code == 302
                oid = int(re.search(r'/orders/(\d+)', response.location).group(1))
                response = action('maker', oid, 'edit', supplier_name='Win7 test supplier',
                                  our_company='苏州博特维智能科技有限公司', payment_terms='月结',
                                  delivery_place='苏州', order_date=date.today().isoformat(),
                                  item_code='TEST-001', item_spec='测试物料', item_unit='件',
                                  item_qty='2', item_price='12.3456', item_date=date.today().isoformat(),
                                  item_remark='', intent='submit')
                assert response.status_code == 302, response.get_data(as_text=True)
                assert action('maker', oid, 'approve', stamped='1').status_code == 403
                assert action('admin', oid, 'approve', stamped='1').status_code == 302
                pdf = clients['maker'].get('/orders/%d/pdf' % oid).data
                assert pdf.startswith(b'%PDF-') and b'%%EOF' in pdf
                if artifacts:
                    artifacts.mkdir(parents=True, exist_ok=True)
                    (artifacts/(kind + '.pdf')).write_bytes(pdf)
                for payment, delivery in [('partial', 'partial'), ('paid', 'delivered')]:
                    assert action('maker', oid, 'follow-up', needs_prepayment='1',
                                  invoice_status='received', delivery_status=delivery,
                                  fulfillment_note='离线环境自检', reason='').status_code == 302
                    assert action('admin', oid, 'payment', payment_status=payment,
                                  payment_date=date.today().isoformat(), payment_note='', reason='').status_code == 302
                assert clients['maker'].get('/orders/%d/pdf' % oid).data == pdf
                print('PASS:', kind, 'login, submit, approval, PDF, partial/full follow-up, archive')
            workbook = openpyxl.load_workbook(io.BytesIO(clients['admin'].get('/export/ledger').data))
            assert workbook.active.max_row == 3
            workbook.close()
            assert clients['admin'].get('/orders?tracking=completed').status_code == 200
            assert clients['admin'].get('/settings/notifications').status_code == 200
            # Test a real HTTP socket using the exact Waitress compatibility wheel.
            server = create_server(module.app, host='127.0.0.1', port=0, threads=2)
            thread = threading.Thread(target=server.run, daemon=True)
            thread.start()
            try:
                http = urllib.request.build_opener(urllib.request.ProxyHandler({}))
                with http.open('http://127.0.0.1:%s/login' % server.effective_port, timeout=10) as response:
                    assert response.status == 200
                    response.read()
            finally:
                server.task_dispatcher.shutdown()
                server.close()
                thread.join(timeout=10)
            print('PASS: Excel, dashboard, notification settings, Waitress HTTP')
        finally:
            module.engine.dispose()
    print('SELFTEST PASSED. Only temporary data was used. No group messages sent.')


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('source', type=Path)
    parser.add_argument('--artifacts', type=Path)
    args = parser.parse_args()
    run(args.source, args.artifacts)