import contextlib
import http.server
import io
import json
import threading
import unittest

from tests.fakes import FakeModel, config, pr_ref
from tests.test_flow import docs_gitea
from werkrevisor.cli import execute
from werkrevisor.redact import Redactor
from werkrevisor.statsclient import StatsClient, StatsError
from tests.fakes import deps as make_deps


class FakeStats:
    def __init__(self, error=None):
        self.records = []
        self.error = error

    def report(self, record):
        if self.error:
            raise self.error
        self.records.append(record)


def review(stats, base_repo={"private": False}, publish=True):
    gitea = docs_gitea()
    gitea.base_repo = base_repo
    d = make_deps(gitea, FakeModel(prompt_files=["doc/PR/1.md"]))
    d.stats = stats
    err = io.StringIO()
    with contextlib.redirect_stderr(err):
        outcome = execute(config(gate={"yellow_gate": "block"}), pr_ref(), d, Redactor(("s3cret-token",)), publish=publish)
    return outcome, err.getvalue()


class StatsClientTest(unittest.TestCase):
    def test_run_reports_its_usage_record(self):
        stats = FakeStats()
        outcome, _ = review(stats)
        [record] = stats.records
        self.assertEqual((record["repo"], record["pr"], record["verdict"]), ("team/app", 7, outcome.result.verdict.name))
        self.assertEqual(record["pr_url"], "https://git.example.org/team/app/pulls/7")
        self.assertIs(record["repo_private"], False)

    def test_a_dry_run_reports_its_usage_record_as_local(self):
        for publish, local in ((True, False), (False, True)):
            with self.subTest(publish=publish):
                stats = FakeStats()
                review(stats, publish=publish)
                self.assertIs(stats.records[0]["local"], local)

    def test_a_private_repository_is_reported_as_such(self):
        stats = FakeStats()
        review(stats, {"private": True})
        self.assertIs(stats.records[0]["repo_private"], True)

    def test_a_missing_or_odd_private_flag_is_reported_as_unknown(self):
        for base_repo in (None, {}, {"private": None}, {"private": "false"}, {"private": 0}):
            with self.subTest(base_repo=base_repo):
                stats = FakeStats()
                review(stats, base_repo)
                self.assertIsNone(stats.records[0]["repo_private"])

    def test_failed_report_never_changes_the_review(self):
        expected, _ = review(None)
        for error in (StatsError("POST /api/usage failed: connection refused"),
                      StatsError("POST /api/usage failed with HTTP 500"),
                      StatsError("POST /api/usage failed with HTTP 401 for s3cret-token")):
            with self.subTest(error=str(error)):
                outcome, log = review(FakeStats(error))
                self.assertEqual((outcome.result.verdict, outcome.result.exit_code),
                                 (expected.result.verdict, expected.result.exit_code))
                self.assertIn("werkrevisor: usage not reported to the statistics service: POST /api/usage failed", log)
                self.assertNotIn("s3cret-token", log)


class SlowStats:
    def report(self, record):
        import time
        time.sleep(30)


class DeadlineTest(unittest.TestCase):
    def test_a_hanging_report_ends_after_the_deadline(self):
        from unittest import mock
        from werkrevisor import cli
        with mock.patch.object(cli, "STATS_DEADLINE_SECONDS", 0.2):
            outcome, log = review(SlowStats())
        self.assertIn("usage not reported to the statistics service: no answer within 0.2 s", log)


class Handler(http.server.BaseHTTPRequestHandler):
    status = 201
    received = []

    def do_POST(self):
        body = self.rfile.read(int(self.headers["Content-Length"]))
        Handler.received.append((self.path, self.headers["Authorization"], json.loads(body)))
        self.send_response(Handler.status)
        self.end_headers()

    def log_message(self, *args):
        pass


class HttpClientTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.server = http.server.HTTPServer(("127.0.0.1", 0), Handler)
        threading.Thread(target=cls.server.serve_forever, daemon=True).start()
        cls.url = f"http://127.0.0.1:{cls.server.server_port}"

    @classmethod
    def tearDownClass(cls):
        cls.server.shutdown()
        cls.server.server_close()

    def test_posts_the_record_with_the_source_token(self):
        Handler.status, Handler.received = 201, []
        StatsClient(self.url, "source-token").report({"repo": "team/app", "pr": 7})
        self.assertEqual(Handler.received, [("/api/usage", "Bearer source-token", {"repo": "team/app", "pr": 7})])

    def test_refused_or_failed_report_raises(self):
        for status in (401, 500):
            with self.subTest(status=status):
                Handler.status = status
                with self.assertRaisesRegex(StatsError, f"HTTP {status}"):
                    StatsClient(self.url, "source-token").report({})
        with self.assertRaisesRegex(StatsError, "POST /api/usage failed"):
            StatsClient("http://127.0.0.1:9", "source-token", timeout=2).report({})

    def test_token_travels_only_over_https_or_localhost(self):
        with self.assertRaisesRegex(StatsError, "must use https"):
            StatsClient("http://stats.example.org", "source-token")
        StatsClient("https://stats.example.org", "source-token")


if __name__ == "__main__":
    unittest.main()