#!/usr/bin/env python3
"""API-client checks with mocked HTTPS responses. No network requests are made."""
import contextlib
import io
import json
import os
import ssl
import tempfile
from http.client import IncompleteRead
from pathlib import Path
from unittest.mock import Mock, patch
from urllib.error import HTTPError, URLError
from urllib.parse import parse_qs, urlsplit
from urllib.request import HTTPSHandler

import fetch_events

TOKEN = "LOCAL-TEST-TOKEN"
ORIGIN = "https://box.example.invalid"


class Response(io.BytesIO):
    status = 200


def check_fetch():
    with tempfile.TemporaryDirectory() as folder:
        output = Path(folder) / "one-page.json"
        arguments = ["--page", "2", "--page-size", "20", "--sorting=-timestamp",
                     "--timeout", "2", "--output", str(output)]
        environment = {"BOX_ORIGIN": ORIGIN, "BOX_ACCESS_TOKEN": TOKEN}

        def run(body=b'{"events": []}', error=None, env=None, args=None):
            opener = Mock()
            opener.open.return_value = Response(body)
            opener.open.side_effect = error
            stdout, stderr = io.StringIO(), io.StringIO()
            with patch.dict(os.environ, environment if env is None else env, clear=True), \
                    patch.object(fetch_events, "build_opener", return_value=opener) as factory, \
                    contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
                code = fetch_events.main(arguments if args is None else args)
            assert TOKEN not in stdout.getvalue() + stderr.getvalue(), "credential in diagnostic"
            return code, stdout.getvalue(), stderr.getvalue(), opener, factory

        body = Path(__file__).with_name("sample-page-1.json").read_bytes()
        code, stdout, stderr, opener, factory = run(body)
        assert code == 0 and stderr == ""
        assert json.loads(stdout) == {"status": "saved", "events": 3,
                                      "page_requested": 2, "page_size_requested": 20}
        assert json.loads(output.read_text(encoding="utf-8")) == json.loads(body)
        request = opener.open.call_args.args[0]
        assert opener.open.call_count == 1 and request.get_method() == "GET"
        assert request.get_header("Authorization") == "Bearer " + TOKEN
        assert urlsplit(request.full_url).path == fetch_events.ENDPOINT
        assert parse_qs(urlsplit(request.full_url).query) == {
            "page": ["2"], "page_size": ["20"], "sorting": ["-timestamp"]}
        assert opener.open.call_args.kwargs["timeout"] == 2
        tls = next(handler for handler in factory.call_args.args if isinstance(handler, HTTPSHandler))
        assert tls._context.check_hostname and tls._context.verify_mode == ssl.CERT_REQUIRED

        original = output.read_bytes()
        code, _, _, opener, _ = run()
        assert code == 1 and output.read_bytes() == original and not opener.open.called
        output.unlink()
        target = Path(folder) / "target.json"
        target.write_bytes(original)
        output.symlink_to(target)
        code, _, _, opener, _ = run()
        assert code == 1 and target.read_bytes() == original and not opener.open.called
        output.unlink()

        assert run()[0] == 0  # An empty events array is a valid page.
        assert json.loads(output.read_text()) == {"events": []}
        output.unlink()
        for invalid in [b'not JSON', b'[]', b'{}', b'{"events": {}}']:
            assert run(invalid)[0] == 1 and not output.exists()
        with patch.object(fetch_events, "MAX_BYTES", 32):
            assert run(b"x" * 33)[0] == 1 and not output.exists()

        for status in [301, 302, 303, 307, 308, 401, 403, 422, 500, 503]:
            error = HTTPError(ORIGIN + "/" + TOKEN, status, TOKEN, {}, io.BytesIO(TOKEN.encode()))
            code, _, stderr, opener, _ = run(error=error)
            assert code == 1 and str(status) in stderr and not output.exists()
            assert opener.open.call_count == 1
        for error in [TimeoutError(TOKEN), URLError(TOKEN), IncompleteRead(TOKEN.encode()),
                      ssl.SSLCertVerificationError(TOKEN)]:
            assert run(error=error)[0] == 1 and not output.exists()

        for origin in ["", "http://box.example.invalid", "https://user:pass@box.example.invalid",
                       ORIGIN + "/api", ORIGIN + "?secret=" + TOKEN, ORIGIN + "#fragment",
                       ORIGIN + "\r\nX-Test: value"]:
            code, _, _, opener, _ = run(env={**environment, "BOX_ORIGIN": origin})
            assert code == 1 and not opener.open.called
        for token in ["", "value\r\nX-Test: value", "Bearer " + TOKEN]:
            code, _, _, opener, _ = run(env={**environment, "BOX_ACCESS_TOKEN": token})
            assert code == 1 and not opener.open.called
        for timeout in ["0", "-1", "nan", "inf"]:
            code, _, _, opener, _ = run(args=[*arguments, "--timeout", timeout])
            assert code == 1 and not opener.open.called
        for flag in ["--page", "--page-size"]:
            for number in ["0", "-1"]:
                try:
                    run(args=[*arguments, flag, number])
                except SystemExit as error:
                    assert error.code == 2
                else:
                    raise AssertionError("nonpositive page accepted")

        redirect = fetch_events.NoRedirect()
        for status in [301, 302, 303, 307, 308]:
            assert redirect.redirect_request(request, None, status, "", {},
                                             "http://other.example.invalid") is None
    print("PASS: API request/TLS, mock responses, empty page, HTTP/JSON errors, timeout, redirects, no overwrite, no credential diagnostics")


if __name__ == "__main__":
    check_fetch()