"""Executed behavior checks for the downloadable example, using loopback only."""

import asyncio
import hashlib
import json
import os
from pathlib import Path
import sys
import unittest

import httpx
from async_proxy_check import check_proxy, make_client
from pool_timeout_demo import BODY, LoopbackFixture, bounded_diagnostic


class ProxyExampleTests(unittest.IsolatedAsyncioTestCase):
    async def test_pool_timeout_before_proxy_and_recovery(self):
        result = await bounded_diagnostic()
        self.assertEqual(result["outcome"], "pass")
        self.assertFalse(result["blocked_request_reached_proxy"])
        self.assertFalse(result["blocked_request_reached_origin"])
        self.assertEqual(result["requests"], {
            "proxy": ["/hold/1", "/hold/2", "/ok"],
            "origin": ["/hold/1", "/hold/2", "/ok"],
        })
        self.assertEqual(result["recovery"]["body"], json.loads(BODY))
        self.assertEqual(result["cleanup"], {"handlers": 0, "writers": 0, "errors": [],
                                            "forced_cancellations": 0, "client_closed": True})

    async def test_exact_cli_uses_explicit_proxy_despite_environment(self):
        fixture = LoopbackFixture()
        async with fixture:
            env = os.environ.copy()
            env.update({"HTTP_PROXY": "http://127.0.0.1:1", "HTTPS_PROXY": "http://127.0.0.1:1",
                        "ALL_PROXY": "http://127.0.0.1:1", "NO_PROXY": "*",
                        "http_proxy": "http://127.0.0.1:1", "https_proxy": "http://127.0.0.1:1",
                        "all_proxy": "http://127.0.0.1:1", "no_proxy": "*",
                        "SSL_CERT_FILE": "/nonexistent-httpx-fixture-ca.pem",
                        "SSL_CERT_DIR": "/nonexistent-httpx-fixture-ca-dir",
                        "PROXY_URL": fixture.proxy_url, "TARGET_URL": fixture.origin_url + "/ok"})
            process = await asyncio.create_subprocess_exec(
                sys.executable, str(Path(__file__).with_name("async_proxy_check.py")),
                env=env, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
            try:
                stdout, stderr = await asyncio.wait_for(process.communicate(), 12.0)
            finally:
                if process.returncode is None:
                    process.kill()
                    await process.wait()
            self.assertEqual(process.returncode, 0, stderr.decode())
            self.assertEqual(stderr, b"")
            results = json.loads(stdout)
            self.assertEqual(len(results), 3)
            for result in results:
                self.assertEqual(result["status"], 200)
                self.assertEqual(result["body_bytes"], len(BODY))
                self.assertEqual(result["body_sha256"], hashlib.sha256(BODY).hexdigest())
            self.assertNotIn(fixture.origin_url.encode(), stdout)
            self.assertNotIn(BODY.strip(), stdout)
            self.assertEqual(fixture.requests, {"proxy": ["/ok"] * 3, "origin": ["/ok"] * 3})
        fixture.assert_clean()

    async def test_status_failure_is_reported_without_body_or_url(self):
        fixture = LoopbackFixture()
        async with fixture:
            results = await check_proxy(fixture.proxy_url, fixture.origin_url + "/status")
            self.assertEqual(results, [{"request": i, "error": "HTTPStatusError"} for i in range(1, 4)])
            self.assertEqual(len(fixture.requests["origin"]), 3)
        fixture.assert_clean()

    async def test_body_limit_closes_responses(self):
        fixture = LoopbackFixture()
        async with fixture:
            results = await check_proxy(fixture.proxy_url, fixture.origin_url + "/large")
            self.assertEqual(results, [{"request": i, "error": "BodyTooLarge"} for i in range(1, 4)])
            self.assertEqual(len(fixture.requests["origin"]), 3)
        fixture.assert_clean()

    async def test_cancellation_closes_streams_and_pending_work(self):
        fixture = LoopbackFixture()
        async with fixture:
            task = asyncio.create_task(check_proxy(fixture.proxy_url, fixture.origin_url + "/hold/1"))
            try:
                await asyncio.wait_for(fixture.ready["/hold/1"].wait(), 2.0)
            finally:
                task.cancel()
                with self.assertRaises(asyncio.CancelledError):
                    await asyncio.wait_for(task, 2.0)
        fixture.assert_clean()

    async def test_fixture_refuses_external_destination(self):
        fixture = LoopbackFixture()
        async with fixture:
            async with make_client(fixture.proxy_url) as client:
                with self.assertRaises(httpx.RemoteProtocolError):
                    await client.get("http://example.invalid/never-requested")
        self.assertEqual(fixture.requests, {"proxy": [], "origin": []})
        self.assertEqual(fixture.errors, [{"side": "proxy", "error": "AssertionError"}])
        self.assertFalse(fixture.tasks or fixture.writers)
        self.assertFalse(any(server.is_serving() for server in fixture.servers))


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