#
# Copyright (C) 2026 Red Hat, Inc.
# SPDX-License-Identifier: GPL-3.0-or-later

import argparse
import asyncio
import ssl
from pathlib import Path
from typing import Any, AsyncGenerator

import pytest
import pytest_asyncio

from cockpit.bridge import Bridge

from .mocktransport import MockTransport

MOCK_DATA = Path(__file__).parent.parent / 'data'


@pytest_asyncio.fixture
async def no_init_transport() -> AsyncGenerator[MockTransport, None]:
    bridge = Bridge(argparse.Namespace(privileged=False, beipack=False))
    transport = MockTransport(bridge)
    try:
        yield transport
    finally:
        await transport.stop()


@pytest.fixture
def transport(no_init_transport: MockTransport) -> MockTransport:
    no_init_transport.init()
    return no_init_transport


async def assert_http_response(transport: MockTransport, ch: str, *, tls: bool = False) -> None:
    await transport.assert_msg('', command='response', channel=ch, status=200, reason='OK')

    channel, data = await transport.next_frame()
    assert channel == ch
    if tls:
        assert data == b'Hello TLS\r\n'
    else:
        assert data == b'Hello\r\n'

    await transport.assert_msg('', command='done', channel=ch)
    await transport.assert_msg('', command='close', channel=ch)


@pytest_asyncio.fixture(params=['tcp', 'unix'])
async def http_server(request: pytest.FixtureRequest, tmp_path: Path) -> AsyncGenerator[dict[str, Any], None]:
    async def serve(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
        while True:
            line = await reader.readline()
            if not line.strip():
                break
        writer.write(b'HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nHello\r\n')
        await writer.drain()
        writer.close()

    if request.param == 'tcp':
        server = await asyncio.start_server(serve, '127.0.0.1', 0)
        port = server.sockets[0].getsockname()[1]
        open_args = {'port': port}
    else:
        sock = str(tmp_path / 'http.sock')
        server = await asyncio.start_unix_server(serve, sock)
        open_args = {'unix': sock}

    yield open_args
    server.close()


@pytest_asyncio.fixture
async def tls_server() -> AsyncGenerator[int, None]:
    async def serve(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
        while True:
            line = await reader.readline()
            if not line.strip():
                break
        writer.write(b'HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nHello TLS\r\n')
        await writer.drain()
        writer.close()

    server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    server_ctx.load_cert_chain(MOCK_DATA / 'mock-server.crt', MOCK_DATA / 'mock-server.key')

    server = await asyncio.start_server(serve, '127.0.0.1', 0, ssl=server_ctx)
    yield server.sockets[0].getsockname()[1]
    server.close()


@pytest.mark.asyncio
async def test_open_error(transport: MockTransport) -> None:
    await transport.check_open('http-stream2', problem='protocol-error',
                               reply_keys={'message': "attribute 'method' required"})
    await transport.check_open('http-stream2', method='POST', path='/tmp/',
                               problem='protocol-error',
                               reply_keys={'message': 'no "port" or "unix" option for channel'})
    await transport.check_open('http-stream2', method='POST', path='/tmp/', port="foo",
                               problem='protocol-error',
                               reply_keys={'message': "attribute 'port': must have type int"})
    await transport.check_open('http-stream2', method='GET', path='/test', unix='/tmp/sock',
                               tls={}, problem='protocol-error',
                               reply_keys={'message': 'TLS on Unix socket is not supported'})


@pytest.mark.asyncio
async def test_http_request(transport: MockTransport, http_server: dict[str, Any]) -> None:
    ch = await transport.check_open('http-stream2', method='GET', path='/test', **http_server)
    transport.send_done(ch)
    await assert_http_response(transport, ch, tls=False)


@pytest.mark.asyncio
async def test_tls_request(transport: MockTransport, tls_server: int) -> None:
    ch = await transport.check_open('http-stream2', method='GET', path='/test', port=tls_server,
                                    tls={'authority': {'file': str(MOCK_DATA / 'mock-server.crt')}})
    transport.send_done(ch)
    await assert_http_response(transport, ch, tls=True)


@pytest.mark.asyncio
async def test_tls_request_no_validate(transport: MockTransport, tls_server: int) -> None:
    ch = await transport.check_open('http-stream2', method='GET', path='/test', port=tls_server,
                                    tls={'validate': False})
    transport.send_done(ch)
    await assert_http_response(transport, ch, tls=True)


@pytest.mark.asyncio
async def test_tls_no_validate(transport: MockTransport, tls_server: int) -> None:
    ch = await transport.check_open('http-stream2', method='GET', path='/test', port=tls_server,
                                    tls={'validate': False})
    transport.send_done(ch)
    await assert_http_response(transport, ch, tls=True)