#if os(macOS)
import Foundation
import XCTest
@testable import RastrilloApple

private actor OpenEvents {
  var values: [String] = []
  func append(_ value: String) { values.append(value) }
}

/// A live socket must close when its consumer rejects or cancels the response;
/// returning a Swift error alone can leave URLSession downloading indefinitely.
final class HTTPStreamCancellationTests: XCTestCase {
  private struct Fixture {
    let process: Process
    let client: NativeHTTP
    let directory: URL
    var closed: URL { directory.appendingPathComponent("closed") }
    func stop() {
      if process.isRunning { process.terminate() }
      try? FileManager.default.removeItem(at: directory)
    }
  }

  private func fixture(contentType: String = "text/event-stream", prefix: String = ": ready\n\n") throws -> Fixture {
    let directory = FileManager.default.temporaryDirectory.appendingPathComponent(UUID().uuidString)
    try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true)
    let process = Process()
    process.executableURL = URL(fileURLWithPath: "/usr/bin/python3")
    process.arguments = ["-u", "-c", #"""
import http.server, pathlib, socket, sys
marker, content_type, prefix = sys.argv[1:]
class Handler(http.server.BaseHTTPRequestHandler):
    def log_message(self, *args): pass
    def do_GET(self):
        self.send_response(200)
        self.send_header('Content-Type', content_type)
        self.end_headers()
        self.wfile.write(prefix.encode('utf-8'))
        self.wfile.flush()
        # Keep the response open without more data. A read blocks in the client
        # until cancellation; the server sees EOF only when the socket closes.
        self.connection.settimeout(4)
        try:
            if self.connection.recv(1) == b'':
                pathlib.Path(marker).write_text('closed')
        except (ConnectionResetError, BrokenPipeError):
            pathlib.Path(marker).write_text('closed')
        except socket.timeout:
            pass
server = http.server.HTTPServer(('127.0.0.1', 0), Handler)
print(server.server_port, flush=True)
server.handle_request()
"""#, directory.appendingPathComponent("closed").path, contentType, prefix]
    let output = Pipe()
    process.standardOutput = output
    do { try process.run() } catch {
      try? FileManager.default.removeItem(at: directory)
      throw error
    }
    var port = Data()
    while port.count < 12 {
      let byte = output.fileHandleForReading.readData(ofLength: 1)
      if byte.isEmpty || byte == Data([10]) { break }
      port.append(byte)
    }
    guard let number = String(data: port, encoding: .utf8).flatMap(Int.init), number > 0,
          let origin = URL(string: "http://127.0.0.1:\(number)") else {
      if process.isRunning { process.terminate() }
      try? FileManager.default.removeItem(at: directory)
      throw NativeClientError.message("The isolated cancellation fixture could not start.")
    }
    return Fixture(process: process, client: NativeHTTP(origin: origin), directory: directory)
  }

  private func assertClosed(_ fixture: Fixture, file: StaticString = #filePath, line: UInt = #line) async throws {
    for _ in 0..<60 {
      if FileManager.default.fileExists(atPath: fixture.closed.path) { return }
      try await Task.sleep(nanoseconds: 25_000_000)
    }
    XCTFail("The rejected or cancelled response left its network socket open", file: file, line: line)
  }

  func testRejectedContentTypeClosesLiveSocket() async throws {
    // Clear Foundation's small text-response sniffing buffer before waiting.
    let server = try fixture(contentType: "text/plain", prefix: String(repeating: "x", count: 4096))
    defer { server.stop() }
    do {
      try await server.client.stream("/live", onOpen: { XCTFail("A refused response invoked onOpen") }) { _ in
        XCTFail("Rejected content reached the consumer")
      }
      XCTFail("Accepted a response with the wrong content type")
    } catch { XCTAssertTrue(error.localizedDescription.contains("refused")) }
    try await assertClosed(server)
  }

  func testResponseSizeFailureClosesLiveSocket() async throws {
    let server = try fixture(contentType: "application/octet-stream", prefix: "12345")
    defer { server.stop() }
    do {
      _ = try await server.client.responseDetails("/live", maxBytes: 4)
      XCTFail("Accepted an oversized response")
    } catch { XCTAssertTrue(error.localizedDescription.contains("too large")) }
    try await assertClosed(server)
  }

  func testThrowingLineConsumerClosesLiveSocket() async throws {
    enum ConsumerFailure: Error { case stopped }
    let server = try fixture()
    defer { server.stop() }
    do {
      try await server.client.stream("/live") { _ in throw ConsumerFailure.stopped }
      XCTFail("A consumer failure was swallowed")
    } catch ConsumerFailure.stopped {} catch { XCTFail("Unexpected error: \(error)") }
    try await assertClosed(server)
  }

  func testOpenCallbackCompletesBeforeFirstLine() async throws {
    enum Stop: Error { case finished }
    let server = try fixture()
    defer { server.stop() }
    let events = OpenEvents()
    do {
      try await server.client.stream("/live", onOpen: {
        // Suspending proves the transport awaits completion rather than merely
        // scheduling the snapshot callback concurrently with incoming lines.
        try await Task.sleep(nanoseconds: 25_000_000)
        await events.append("opened")
      }) { line in
        await events.append(line)
        throw Stop.finished
      }
      XCTFail("The line callback failure was swallowed")
    } catch Stop.finished {} catch { XCTFail("Unexpected error: \(error)") }
    let received = await events.values
    XCTAssertEqual(received, ["opened", ": ready"])
    try await assertClosed(server)
  }

  func testThrowingOpenCallbackClosesWithoutDeliveringLines() async throws {
    enum Stop: Error { case finished }
    let server = try fixture()
    defer { server.stop() }
    do {
      try await server.client.stream("/live", onOpen: { throw Stop.finished }) { _ in
        XCTFail("Lines arrived after snapshot initialization failed")
      }
      XCTFail("The open callback failure was swallowed")
    } catch Stop.finished {} catch { XCTFail("Unexpected error: \(error)") }
    try await assertClosed(server)
  }

  func testCancellationWhileAwaitingBytesClosesLiveSocket() async throws {
    let server = try fixture()
    defer { server.stop() }
    let ready = expectation(description: "received stream prefix")
    let finished = expectation(description: "cancelled stream returned")
    let task = Task {
      do {
        try await server.client.stream("/live") { line in
          if line == ": ready" { ready.fulfill() }
        }
        XCTFail("A cancelled stream completed successfully")
      } catch {
        XCTAssertTrue(Task.isCancelled)
        // Callers match cancellation by type; URLSession's -999 must not leak.
        XCTAssertTrue(error is CancellationError, "Expected CancellationError, got \(error)")
      }
      finished.fulfill()
    }
    await fulfillment(of: [ready], timeout: 2)
    // Give the iterator an opportunity to suspend waiting for the next byte.
    try await Task.sleep(nanoseconds: 50_000_000)
    task.cancel()
    await fulfillment(of: [finished], timeout: 2)
    try await assertClosed(server)
  }
}
#endif