#!/usr/bin/env python3
"""Local regression checks. All network operations are mocked."""
import contextlib
import copy
import hashlib
import http.client
import io
import json
from pathlib import Path
import runpy
import sys
import tempfile
import urllib.error
from unittest import mock
import source_guard as g

TEXT = 'Synthetic source with Unicode: café\n\\n'
PAGE = {'ok': True, 'generation': 'b' * 32,
        'data': {'root_id': g.EXAMPLE},
        'messages': [{'id': g.EXAMPLE, 'type': 'message',
                      'visibility': 'public', 'hidden': False,
                      'author': 'a' * 64, 'text': TEXT,
                      'sha256': hashlib.sha256(TEXT.encode()).hexdigest()}]}
BASE, _ = g.inspect_root(PAGE, g.EXAMPLE)
results = []

def verify(name, action):
    action()
    results.append(name)

def blocked(action):
    try:
        action()
    except g.Blocked:
        return
    raise AssertionError('Unexpectedly accepted')

def change_message(patch):
    p = copy.deepcopy(PAGE)
    p['messages'][0].update(patch)
    return p

def assert_equal(actual, expected):
    assert actual == expected, (actual, expected)

for key, value in [('hidden', True), ('hidden', 0),
                   ('type', 'tombstone'), ('visibility', 'private'),
                   ('superseded_by', 'c' * 32), ('superseded_by', False),
                   ('superseded_by', None), ('superseded_by', []),
                   ('superseded_by', {}), ('author', 'anonymous'),
                   ('sha256', '0' * 64), ('text', 'Changed text'),
                   ('text', None)]:
    p = change_message({key: value})
    verify('message ' + key + '=' + repr(value),
           lambda p=p: blocked(lambda: g.inspect_root(p, g.EXAMPLE)))

for key in ('generation', 'messages', 'data', 'ok'):
    p = copy.deepcopy(PAGE)
    del p[key]
    verify('missing ' + key,
           lambda p=p: blocked(lambda: g.inspect_root(p, g.EXAMPLE)))

p = copy.deepcopy(PAGE)
p['messages'].append(copy.deepcopy(p['messages'][0]))
verify('duplicate root', lambda: blocked(lambda: g.inspect_root(p, g.EXAMPLE)))
p = copy.deepcopy(PAGE)
p['messages'] = []
verify('absent root', lambda: blocked(lambda: g.inspect_root(p, g.EXAMPLE)))

for field, value in [('author', 'c' * 64), ('generation', 'c' * 32),
                     ('sha256', 'c' * 64), ('id', 'c' * 32)]:
    current = dict(BASE, **{field: value})
    verify('changed ' + field,
           lambda current=current: blocked(lambda: g.compare_pin(current, BASE)))

p = change_message({'text': 'Correctly rehashed new text',
                    'sha256': hashlib.sha256(b'Correctly rehashed new text').hexdigest()})
pin, _ = g.inspect_root(p, g.EXAMPLE)
verify('body and digest changed together',
       lambda: blocked(lambda: g.compare_pin(pin, BASE)))
verify('boolean schema rejected',
       lambda: blocked(lambda: g.validate_pin(dict(BASE, schema=True))))

class FakeResponse:
    def __init__(self, raw, status=200):
        self.raw, self.status = raw, status
    def __enter__(self):
        return self
    def __exit__(self, *args):
        return False
    def read(self, n):
        return self.raw[:n]

def cli(args, expected_code, expected_reason=None, raw=None, error=None):
    stdout, stderr = io.StringIO(), io.StringIO()
    response = FakeResponse(json.dumps(PAGE).encode() if raw is None else raw)
    with mock.patch.object(sys, 'argv', ['source_guard.py'] + args), \
         mock.patch('urllib.request.OpenerDirector.open', return_value=response,
                    side_effect=error) as opened, \
         contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
        try:
            runpy.run_path(str(Path(g.__file__)), run_name='__main__')
        except SystemExit as e:
            assert e.code == expected_code, (e.code, stdout.getvalue(), stderr.getvalue())
        else:
            raise AssertionError('CLI did not exit')
    result = json.loads(stdout.getvalue())
    if expected_reason:
        assert result == {'status': 'BLOCKED', 'reason': expected_reason}, result
    if expected_code:
        assert 'untrusted_source_text' not in result, result
    return result, opened.call_args_list

with tempfile.TemporaryDirectory() as temp:
    baseline = Path(temp) / 'baseline.json'
    baseline.write_text(json.dumps(BASE))
    args = ['check', str(baseline)]
    for error in [urllib.error.URLError('offline'), TimeoutError('timeout'),
                  http.client.IncompleteRead(b'partial'),
                  http.client.BadStatusLine('invalid'),
                  RecursionError('JSON nested too deeply')]:
        verify('CLI ' + type(error).__name__,
               lambda error=error: cli(args, 1, type(error).__name__, error=error))
    verify('CLI malformed JSON', lambda: cli(args, 1, 'JSONDecodeError', raw=b'{'))
    verify('CLI invalid UTF-8', lambda: cli(args, 1, 'UnicodeDecodeError', raw=b'\xff'))
    verify('CLI oversize response',
           lambda: cli(args, 2, 'Response exceeded the local size limit', raw=b' ' * (g.MAX_BYTES + 1)))
    verify('CLI baseline refuses overwrite',
           lambda: cli(['capture', str(baseline)], 1, 'FileExistsError'))
    assert json.loads(baseline.read_text()) == BASE
    verify('CLI missing baseline',
           lambda: cli(['check', str(Path(temp) / 'missing.json')], 1, 'FileNotFoundError'))
    bad = Path(temp) / 'changed.json'
    bad.write_text(json.dumps(dict(BASE, sha256='0' * 64)))
    verify('CLI modified baseline hash',
           lambda: cli(['check', str(bad)], 2, 'Review required; changed: sha256'))
    result, calls = cli(args, 0)
    assert result['status'] == 'MATCH' and result['untrusted_source_text'] == TEXT
    request = calls[0].args[0]
    assert request.full_url == g.ORIGIN + '/api/thread/' + g.EXAMPLE + '?limit=1'
    assert request.get_method() == 'GET' and request.data is None
    results.append('CLI MATCH uses fixed HTTPS read-only endpoint')

verify('redirect request refused',
       lambda: assert_equal(g.NoRedirect().redirect_request(None, None, 302, '', {}, 'https://example.invalid/'), None))
print(json.dumps({'status': 'REGRESSION_TESTS_COMPLETE', 'checks': len(results),
                  'network_requests': 0, 'passed': results}, indent=2))