Fix connection op for pods/log

lib/k8s/client/mint/request.ex

@@ -132,6 +132,7 @@ defmodule K8s.Client.Mint.Request do
   def map_frame({:binary, <<1, msg::binary>>}), do: {:stdout, msg}
   def map_frame({:binary, <<2, msg::binary>>}), do: {:stderr, msg}
   def map_frame({:binary, <<3, msg::binary>>}), do: {:error, msg}
+  def map_frame({:binary, <<msg::binary>>}), do: {:stdout, msg}
 
   @spec map_outgoing_frame({:stdin, binary()} | {:close, integer(), binary()} | :close | :exit) ::
           {:ok, :close | {:text, binary} | {:close, integer(), binary()}}

lib/k8s/operation.ex

@@ -30,6 +30,22 @@ defmodule K8s.Operation do
     apply: "application/apply-patch+yaml"
   }
 
+  @allowed_connect_params [
+    :command,
+    :container,
+    :follow,
+    :limitBytes,
+    :pretty,
+    :previous,
+    :sinceSeconds,
+    :stderr,
+    :stdin,
+    :stdout,
+    :tailLines,
+    :timestamps,
+    :tty
+  ]
+
   defstruct method: nil,
             verb: nil,
             api_version: nil,
@@ -212,9 +228,7 @@ defmodule K8s.Operation do
 
         verb === :connect ->
           [stdin: true, stdout: true, stderr: true, tty: false]
-          |> Keyword.merge(
-            Keyword.take(opts, [:stdin, :stdout, :stderr, :tty, :command, :container])
-          )
+          |> Keyword.merge(Keyword.take(opts, @allowed_connect_params))
 
         true ->
           []

test/k8s/client/runner/base_integration_test.exs

@@ -271,7 +271,7 @@ defmodule K8s.Client.Runner.BaseIntegrationTest do
 
   @tag :integration
   @tag :websocket
-  test "runs :connect operations and returns stdout", %{
+  test "runs :connect operations for pods/exec and returns stdout", %{
     conn: conn,
     labels: labels,
     test_id: test_id
@@ -303,6 +303,43 @@ defmodule K8s.Client.Runner.BaseIntegrationTest do
     assert response.stdout =~ "ok"
   end
 
+  @tag :integration
+  @tag :websocket
+  @tag :wip
+  @tail_lines 5
+  test "runs :connect operations for pods/log and returns stdout", %{
+    conn: conn,
+    labels: labels,
+    test_id: test_id
+  } do
+    {:ok, created_pod} =
+      build_pod("k8s-ex-#{test_id}", labels)
+      |> K8s.Client.create()
+      |> K8s.Client.put_conn(conn)
+      |> K8s.Client.run()
+
+    {:ok, _} =
+      K8s.Client.wait_until(conn, K8s.Client.get(created_pod),
+        find: ["status", "containerStatuses", Access.filter(&(&1["ready"] == true))],
+        eval: &match?([_ | _], &1),
+        timeout: 60
+      )
+
+    {:ok, response} =
+      K8s.Client.connect(
+        created_pod["apiVersion"],
+        "pods/log",
+        [namespace: K8s.Resource.namespace(created_pod), name: K8s.Resource.name(created_pod)],
+        tailLines: @tail_lines
+      )
+      |> K8s.Client.put_conn(conn)
+      |> K8s.Client.run()
+
+    assert String.printable?(response.stdout)
+    lines = String.split(response.stdout, "\n", trim: true)
+    assert @tail_lines == length(lines)
+  end
+
   @tag :integration
   @tag :websocket
   test "runs :connect operations and returns errors", %{