import json
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta
from urllib.error import HTTPError
from urllib.parse import parse_qs, urlsplit

import pytest
from sqlalchemy import text

from conftest import PASSWORD, post
from test_workflow import action, approve, form, new, order, pending, save

HOOK = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=TEST-ONLY-NOT-A-REAL-KEY'
BASE = 'http://192.168.1.10:8080'


def configure(clients, **overrides):
    data = dict(enabled='1', base_url=BASE,
                approval_webhook=HOOK + '-approval', result_webhook=HOOK + '-result')
    data.update(overrides)
    return post(clients['admin'], '/settings/notifications', data)


def notices(app):
    from models import Notification
    with app.Session() as db:
        return [dict(id=n.id, status=n.status, channel=n.channel, attempts=n.attempts, error=n.last_error)
                for n in db.query(Notification).order_by(Notification.id)]


def test_workflow_queues_only_committed_transitions(app, clients, monkeypatch):
    import notifications
    from models import Notification, NotificationSettings
    monkeypatch.setattr(notifications, 'send_wecom', lambda *args: pytest.fail('HTTP request must not send'))
    oid = pending(app, clients)  # Disabled by default; no historical replay.
    assert not notices(app)
    assert configure(clients).status_code == 302
    assert action(app, clients['maker'], oid, 'withdraw', comment='fix').status_code == 302
    assert action(app, clients['maker'], oid, 'submit').status_code == 302
    stale = order(app, oid)['version']
    assert action(app, clients['approver'], oid, 'approve', stamped='1').status_code == 302
    assert post(clients['reviewer'], f'/orders/{oid}/approve', {'row_version': str(stale), 'stamped': '1'}).status_code == 409
    assert [n['channel'] for n in notices(app)] == ['result', 'approval', 'result']
    fresh = new(app, clients['maker'])
    assert save(app, clients['maker'], fresh, intent='submit', qty='-1').status_code == 400
    assert len(notices(app)) == 3
    # Even a failure after enqueue rolls back both order and notification.
    original = notifications.enqueue_transition
    def fail_after_enqueue(*args):
        original(*args)
        raise ValueError('simulated persistence failure')
    monkeypatch.setattr(notifications, 'enqueue_transition', fail_after_enqueue)
    assert save(app, clients['maker'], fresh, intent='submit').status_code == 400
    assert order(app, fresh)['status'] == 'draft'
    assert len(notices(app)) == 3


def test_worker_cancels_old_tasks_and_sends_minimal_lan_links(app, clients):
    from notifications import dispatch_once
    configure(clients, result_webhook='')
    oid = pending(app, clients)
    assert action(app, clients['maker'], oid, 'withdraw', comment='修改').status_code == 302
    assert save(app, clients['maker'], oid, intent='submit').status_code == 302
    sent = []
    def sender(hook, payload):
        # Sending must not hold SQLite's write lock.
        with app.Session() as db:
            db.execute(text('BEGIN IMMEDIATE'))
        sent.append((hook, payload))
    now = datetime.utcnow() + timedelta(seconds=1)
    assert dispatch_once(app.Session, sender, now)
    assert notices(app)[0]['status'] == 'cancelled'
    assert not sent
    assert dispatch_once(app.Session, sender, now)
    assert notices(app)[1]['status'] == 'sent'
    assert sent[0][0] == HOOK + '-approval'
    content = sent[0][1]['markdown']['content']
    assert f'{BASE}/orders/{oid}/view' in content
    assert 'Audit supplier' not in content and 'key=' not in content
    assert sent[0][1]['msgtype'] == 'markdown'
    configure(clients)
    assert action(app, clients['approver'], oid, 'approve', stamped='1').status_code == 302
    assert not dispatch_once(app.Session, sender, now)  # Shared rate limit.
    assert dispatch_once(app.Session, sender, now + timedelta(seconds=5))
    assert sent[-1][0] == HOOK + '-result'
    assert clients['maker'].get(f'/orders/{oid}/pdf').data.startswith(b'%PDF')


def test_retry_recovery_pause_and_no_credentials_in_errors(app, clients):
    from models import Notification, NotificationSettings
    from notifications import dispatch_once
    configure(clients)
    oid = pending(app, clients)
    now = datetime.utcnow() + timedelta(seconds=1)
    def fail(*args):
        raise RuntimeError(HOOK)
    for attempt in range(5):
        assert dispatch_once(app.Session, fail, now + timedelta(hours=attempt))
    row = notices(app)[0]
    assert row['status'] == 'failed' and row['attempts'] == 5
    assert HOOK not in row['error']
    assert post(clients['admin'], f"/settings/notifications/{row['id']}/retry").status_code == 302
    configure(clients, enabled='0')
    assert not dispatch_once(app.Session, fail, now + timedelta(hours=6))
    configure(clients)
    # Recover a crashed sender after its lease expires.
    with app.Session() as db:
        notice = db.get(Notification, row['id'])
        notice.status = 'sending'
        notice.lease_token = 'expired'
        notice.lease_until = now
        db.commit()
    sent = []
    assert dispatch_once(app.Session, lambda *a: sent.append(a), now + timedelta(hours=6))
    assert len(sent) == 1 and notices(app)[0]['status'] == 'sent'
    assert not dispatch_once(app.Session, fail, now + timedelta(hours=7))
    assert post(clients['admin'], f"/settings/notifications/{row['id']}/retry").status_code == 409
    html = clients['admin'].get('/settings/notifications').get_data(as_text=True)
    assert HOOK not in html and '已发送' in html


def test_parallel_workers_claim_once(app, clients):
    from notifications import dispatch_once
    configure(clients)
    pending(app, clients)
    sent = []
    now = datetime.utcnow() + timedelta(seconds=1)
    with ThreadPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(lambda _: dispatch_once(app.Session, lambda *a: sent.append(a), now), range(4)))
    assert results.count(True) == 1 and len(sent) == 1


def test_settings_permissions_secret_preservation_and_testing(app, clients):
    from models import NotificationSettings
    for name in ('maker', 'reviewer', 'approver'):
        assert clients[name].get('/settings/notifications').status_code == 403
        for path in ('/settings/notifications', '/settings/notifications/test', '/settings/notifications/1/retry'):
            assert post(clients[name], path).status_code == 403
    assert clients['admin'].post('/settings/notifications', data={'enabled': '1'}).status_code == 400
    assert configure(clients).status_code == 302
    assert configure(clients, approval_webhook='', result_webhook='').status_code == 302
    with app.Session() as db:
        assert db.get(NotificationSettings, 1).approval_webhook == HOOK + '-approval'
    html = clients['admin'].get('/settings/notifications')
    assert html.headers['Cache-Control'] == 'no-store'
    assert HOOK not in html.get_data(as_text=True)
    for _ in range(2):
        assert post(clients['admin'], '/settings/notifications/test', {'channel': 'approval'}).status_code == 302
    assert len(notices(app)) == 1
    assert configure(clients, clear_approval='1').status_code == 400
    with app.Session() as db:
        assert db.get(NotificationSettings, 1).approval_webhook == HOOK + '-approval'
    assert configure(clients, clear_result='1').status_code == 302
    assert post(clients['admin'], '/settings/notifications/test', {'channel': 'result'}).status_code == 400


@pytest.mark.parametrize('url', ['http://127.0.0.1:8080', 'http://localhost:8080', 'http://0.0.0.0',
    'https://example.org/orders/1/view', 'https://user:pass@192.168.1.2', 'http://192.168.1.2/?x=1',
    'http://192.168.1.2/#x', 'javascript:alert(1)', 'http://8.8.8.8', 'http://192.168.1.2:99999'])
def test_invalid_lan_addresses(app, clients, url):
    assert configure(clients, base_url=url).status_code == 400


@pytest.mark.parametrize('url', ['http://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=x',
    'https://qyapi.weixin.qq.com.evil.invalid/cgi-bin/webhook/send?key=x',
    HOOK + '&redirect=evil', HOOK + '#secret', 'http://192.168.1.1'])
def test_webhook_rejects_untrusted_destinations(app, clients, url):
    assert configure(clients, approval_webhook=url).status_code == 400
    assert 'value="' + url + '"' not in clients['admin'].get('/settings/notifications').get_data(as_text=True)


@pytest.mark.parametrize('reply,expected', [(b'{"errcode":0,"errmsg":"ok"}', None),
    (b'{"errcode":45009,"errmsg":"secret url"}', '45009'), (b'{}', '格式'),
    (b'{"errcode":false}', '格式'), (b'not json', '格式'), (b'x'*16385, '异常')])
def test_wecom_protocol(app, monkeypatch, reply, expected):
    import notifications
    class Response:
        status = 200
        def __enter__(self): return self
        def __exit__(self, *args): pass
        def read(self, count): return reply[:count]
    class Opener:
        def open(self, req, timeout):
            assert req.full_url == HOOK and req.method == 'POST' and timeout == 5
            assert json.loads(req.data)['msgtype'] == 'markdown'
            return Response()
    monkeypatch.setattr(notifications, 'build_opener', lambda *args: Opener())
    if expected:
        with pytest.raises(notifications.DeliveryError, match=expected) as error:
            notifications.send_wecom(HOOK, {'msgtype': 'markdown'})
        assert 'secret url' not in str(error.value)
    else:
        notifications.send_wecom(HOOK, {'msgtype': 'markdown'})


def test_wecom_redirect_is_not_followed(app, monkeypatch):
    import notifications
    class Opener:
        def open(self, req, timeout):
            raise HTTPError(HOOK, 302, 'redirect', {}, None)
    def opener(handler):
        assert handler.redirect_request(None, None, 302, '', {}, 'https://evil.invalid') is None
        return Opener()
    monkeypatch.setattr(notifications, 'build_opener', opener)
    with pytest.raises(notifications.DeliveryError, match='302'):
        notifications.send_wecom(HOOK, {})


def test_notice_login_password_return_and_scope(app, clients):
    oid = pending(app, clients)
    destination = f'/orders/{oid}/view'
    with app.Session() as db:
        db.query(app.User).filter_by(username='reviewer').one().must_change_password = True
        db.commit()
    c = app.app.test_client()
    response = c.get(destination)
    assert parse_qs(urlsplit(response.location).query)['next'] == [destination]
    c.get(response.location)
    response = post(c, '/login', {'username': 'reviewer', 'password': PASSWORD, 'next': destination})
    assert urlsplit(response.location).path == '/account/password'
    assert parse_qs(urlsplit(response.location).query)['next'] == [destination]
    c.get(response.location)
    new_password = PASSWORD + '-personal'
    response = post(c, '/account/password', {'old_password': PASSWORD, 'password': new_password,
                'confirm_password': new_password, 'next': destination})
    assert response.location == destination
    assert c.get(destination).status_code == 200
    assert order(app, oid)['status'] == 'pending_approval'  # Viewing cannot approve.
    assert clients['other'].get(destination).status_code == 404
    assert c.get(f'/orders/{oid}/review').status_code == 405


@pytest.mark.parametrize('destination', ['https://evil.invalid', '//evil.invalid', '/\\evil.invalid',
    '/orders/1/approve', '/settings/notifications'])
def test_login_next_cannot_escape_or_execute(app, destination):
    c = app.app.test_client()
    c.get('/login')
    assert post(c, '/login', {'username': 'maker', 'password': PASSWORD, 'next': destination}).location == '/'