# Copyright (C) 2013, Eric Wong <normalperson@yhbt.net> and all contributors
# License: GPLv3 or later (https://www.gnu.org/licenses/gpl-3.0.txt)
require_relative 'server_helper'
require 'digest/md5'
require 'rack/file'

class TestOutputBuffering < Testcase
  parallelize_me!
  include ServerHelper
  alias setup server_helper_setup
  alias teardown server_helper_teardown

  GPLv3 = File.read("COPYING")
  RAND = IO.binread("/dev/urandom", 666) * 119
  dig = Digest::MD5.new
  NR = 1337
  MD5 = Thread.new do
    NR.times { dig << RAND }
    dig.hexdigest
  end

  class BigBody
    def each
      NR.times { yield RAND }
    end
  end

  def test_output_buffer_false_curl
    output_buffer(false, :curl)
  end

  def test_output_buffer_false_http09
    output_buffer(false, :http09)
  end

  def test_output_buffer_true_curl
    output_buffer(true, :curl)
  end

  def test_output_buffer_true_http09
    output_buffer(true, :http09)
  end

  def output_buffer(btype, check_type, delay = 4)
    err = @err
    cfg = Yahns::Config.new
    host, port = @srv.addr[3], @srv.addr[1]
    len = (RAND.size * NR).to_s
    cfg.instance_eval do
      ru = lambda do |e|
        [ 200, {'Content-Length'=>len}, BigBody.new ]
      end
      GTL.synchronize do
        app(:rack, ru) do
          listen "#{host}:#{port}"
          output_buffering btype
        end
      end
      logger(Logger.new(err.path))
    end
    srv = Yahns::Server.new(cfg)
    pid = fork do
      ENV["YAHNS_FD"] = @srv.fileno.to_s
      srv.start.join
    end

    case check_type
    when :curl
      # curl is faster for piping gigantic wads of data than Net::HTTP
      sh_sleep = delay ? "sleep #{delay} && " : ""
      md5 = `curl -sSf http://#{host}:#{port}/ | (#{sh_sleep} md5sum)`
      assert $?.success?, $?.inspect
      (md5 =~ /([a-f0-9]{32})/i) or raise "bad md5: #{md5.inspect}"
      md5 = $1
      assert_equal MD5.value, md5
    when :http09
      # HTTP/0.9
      c = TCPSocket.new(host, port)
      c.write("GET /\r\n\r\n")
      md5in = IO.pipe
      md5out = IO.pipe
      sleep(delay) if delay
      md5pid = Process.spawn("md5sum", :in => md5in[0], :out => md5out[1])
      md5in[0].close
      md5out[1].close
      assert_equal(NR * RAND.size, IO.copy_stream(c, md5in[1]))
      c.shutdown
      c.close
      md5in[1].close
      _, status = Timeout.timeout(10) { Process.waitpid2(md5pid) }
      assert status.success?, status.inspect
      md5 = md5out[0].read
      (md5 =~ /([a-f0-9]{32})/i) or raise "bad md5: #{md5.inspect}"
      md5 = $1
      assert_equal MD5.value, md5
      md5out[0].close
    else
      raise "TESTBUG"
    end
  rescue => e
    Yahns::Log.exception(Logger.new($stderr), "test", e)
    raise
  ensure
    quit_wait(pid)
  end

  class BigHeader
    A = "A" * 65536
    def initialize(h)
      @h = h
    end
    def each
      NR.times do |n|
        yield("X-#{n}", A)
      end
      @h.each { |k,v| yield(k,v) }
    end
  end

  def test_big_header
    err = @err
    cfg = Yahns::Config.new
    host, port = @srv.addr[3], @srv.addr[1]
    cfg.instance_eval do
      ru = lambda do |e|
        case e["PATH_INFO"]
        when "/COPYING"
          Rack::File.new(Dir.pwd).call(e)
          gplv3 = File.open("COPYING")
          def gplv3.each
            raise "SHOULD NOT BE CALLED"
          end
          size = gplv3.stat.size
          len = size.to_s
          ranges = Rack::Utils.byte_ranges(e, size)
          status = 200
          h = { "Content-Type" => "text/plain", "Content-Length" => len }
          if ranges && ranges.size == 1
            status = 206
            range = ranges[0]
            h["Content-Range"] = "bytes #{range.begin}-#{range.end}/#{size}"
            size = range.end - range.begin + 1
            len.replace(size.to_s)
          end
          [ status , BigHeader.new(h), gplv3 ]
        when "/"
          h = { "Content-Type" => "text/plain", "Content-Length" => "4" }
          [ 200, BigHeader.new(h), ["BIG\n"] ]
        else
          raise "WTF"
        end
      end
      GTL.synchronize do
        app(:rack, ru) do
          listen "#{host}:#{port}"
        end
      end
      logger(Logger.new(err.path))
    end
    srv = Yahns::Server.new(cfg)
    pid = fork do
      ENV["YAHNS_FD"] = @srv.fileno.to_s
      srv.start.join
    end
    threads = []

    # start with just a big header
    threads << Thread.new do
      c = TCPSocket.new(host, port)
      c.write "GET / HTTP/1.0\r\n\r\n"
      begin
        sleep 1
      end while c.nread == 0
      nr = 0
      last = nil
      c.each_line do |line|
        case line
        when %r{\AX-} then nr += 1
        else
          last = line
        end
      end
      assert_equal NR, nr
      assert_equal "BIG\n", last
      c.close
    end

    threads << Thread.new do
      c = TCPSocket.new(host, port)
      c.write "GET /COPYING HTTP/1.0\r\n\r\n"
      begin
        sleep 1
      end while c.nread == 0
      nr = 0
      c.each_line do |line|
        case line
        when %r{\AX-} then nr += 1
        else
          break if line == "\r\n"
        end
      end
      assert_equal NR, nr
      assert_equal GPLv3, c.read
      c.close
    end

    threads << Thread.new do
      c = TCPSocket.new(host, port)
      c.write "GET /COPYING HTTP/1.0\r\nRange: bytes=5-46\r\n\r\n"
      begin
        sleep 1
      end while c.nread == 0
      nr = 0
      c.each_line do |line|
        case line
        when %r{\AX-} then nr += 1
        else
          break if line == "\r\n"
        end
      end
      assert_equal NR, nr
      assert_equal GPLv3[5..46], c.read
      c.close
    end
    threads.each do |t|
      assert_equal t, t.join(30)
      assert_nil t.value
    end
  ensure
    quit_wait(pid)
  end

  def test_client_timeout
    err = @err
    apperr = tmpfile(%w(app .err))
    cfg = Yahns::Config.new
    size = RAND.size * NR
    host, port = @srv.addr[3], @srv.addr[1]
    cfg.instance_eval do
      ru = lambda do |e|
        if e["PATH_INFO"] == "/bh"
          h = { "Content-Type" => "text/plain", "Content-Length" => "4" }
          [ 200, BigHeader.new(h), ["BIG\n"] ]
        else
          [ 200, {'Content-Length' => size.to_s }, BigBody.new ]
        end
      end
      GTL.synchronize do
        app(:rack, ru) do
          listen "#{host}:#{port}"
          output_buffering false
          client_timeout 3
          logger(Logger.new(apperr.path))
        end
      end
      logger(Logger.new(err.path))
    end
    srv = Yahns::Server.new(cfg)
    pid = fork do
      ENV["YAHNS_FD"] = @srv.fileno.to_s
      srv.start.join
    end
    threads = []
    threads << Thread.new do
      c = TCPSocket.new(host, port)
      c.write("GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")
      sleep(5) # wait for timeout
      assert_operator c.nread, :>, 0
      c
    end

    threads << Thread.new do
      c = TCPSocket.new(host, port)
      c.write("GET /bh HTTP/1.1\r\nHost: example.com\r\n\r\n")
      sleep(5) # wait for timeout
      assert_operator c.nread, :>, 0
      c
    end
    threads.each { |t| t.join(10) }
    assert_operator size, :>, threads[0].value.read.size
    assert_operator size, :>, threads[1].value.read.size
    msg = File.readlines(apperr.path)
    msg = msg.grep(/timeout on :wait_writable after 3s$/)
    assert_equal 2, msg.size
    threads.each { |t| t.value.close }
  ensure
    apperr.close! if apperr
    quit_wait(pid)
  end
end if `which curl 2>/dev/null`.strip =~ /curl/ &&
       `which md5sum 2>/dev/null`.strip =~ /md5sum/