feat: streamed results

doc/neotest.txt

@@ -76,7 +76,7 @@ neotest.setup({user_config})                                 *neotest.setup()*
           skipped = "NeotestSkipped",
           target = "NeotestTarget",
           test = "NeotestTest",
-          unknown = "NeotestUnknown",
+          unknown = "NeotestUnknown"
         },
         icons = {
           child_indent = "│",

lua/neotest/async.lua

@@ -18,6 +18,25 @@ end
 local async_wrapper = {
   api = proxy_vim("api"),
   fn = proxy_vim("fn"),
+  lib = {
+    first = function(...)
+      local functions = { ... }
+      local send_ran, await_ran = plen_async.control.channel.oneshot()
+      local result, ran
+      for _, func in ipairs(functions) do
+        plen_async.run(function()
+          local func_result = func()
+          if not ran then
+            result = func_result
+            ran = true
+            send_ran()
+          end
+        end)
+      end
+      await_ran()
+      return result
+    end,
+  },
 }
 if false then
   -- For type checking

lua/neotest/client/init.lua

@@ -55,33 +55,24 @@ function NeotestClient:run_tree(tree, args)
     return
   end
   self._state:update_running(adapter_id, root.id, pos_ids)
-  local all_results = {}
-  local success, error = pcall(
+  local success, all_results = pcall(
     self._runner.run_tree,
     self._runner,
     tree,
     args,
     adapter,
     function(results)
-      for pos_id, result in pairs(results) do
-        all_results[pos_id] = result
-      end
-      self._state:update_results(adapter_id, results)
+      self._state:update_results(adapter_id, results, true)
     end
   )
   if not success then
-    lib.notify(("%s: %s"):format(adapter.name, error), "warn")
+    lib.notify(("%s: %s"):format(adapter.name, all_results), "warn")
     all_results = {}
     for _, pos in tree:iter() do
       all_results[pos.id] = { status = "skipped" }
     end
   end
-  if root.type ~= "test" then
-    self._runner:fill_results(tree, all_results)
-  end
-  if root.type == "test" or root.type == "namespace" then
-    all_results[root.path] = nil
-  end
+
   self._state:update_results(adapter_id, all_results)
 end
 
@@ -299,7 +290,7 @@ function NeotestClient:_get_adapter(position_id, adapter_id, refresh)
   else
     for _, adapter in ipairs(self._adapters) do
       local root = self._state:positions(adapter.name)
-      if vim.startswith(position_id, root:data().path) then
+      if root and vim.startswith(position_id, root:data().path) then
         return adapter.name, adapter
       end
     end

lua/neotest/client/runner.lua

@@ -20,18 +20,46 @@ end
 ---@param args table
 ---@param adapter neotest.Adapter
 function TestRunner:run_tree(tree, args, adapter, on_results)
-  local results = {}
-  local results_callback = function(results_)
-    on_results(results_)
-    for pos_id, result in ipairs(results_) do
-      results[pos_id] = result
+  local all_results = {}
+  local results_callback = function(root, results, output_path)
+    local function fill_results(missing_results)
+      for pos_id, result in pairs(missing_results) do
+        results[pos_id] = result
+      end
+    end
+
+    fill_results(self:_missing_results(root, results, not output_path))
+
+    if output_path then
+      for _, pos in root:iter() do
+        if not results[pos.id] and not all_results[pos.id] then
+          results[pos.id] = {
+            status = "failed",
+            errors = {},
+            output = output_path,
+          }
+        end
+      end
+
+      for _, result in pairs(results) do
+        if not result.output then
+          result.output = output_path
+        end
+      end
+    end
+
+    for pos_id, result in pairs(results) do
+      all_results[pos_id] = result
     end
+    fill_results(self:_missing_results(tree, all_results, true))
+    on_results(results)
   end
 
   args = vim.tbl_extend("keep", args or {}, { strategy = "integrated" })
 
   self:_run_tree(tree, args, adapter, results_callback)
-  on_results(results)
+
+  return all_results
 end
 
 function TestRunner:_run_tree(tree, args, adapter, results_callback)
@@ -44,92 +72,31 @@ function TestRunner:_run_tree(tree, args, adapter, results_callback)
   self:_run_spec(spec, tree, args, adapter, results_callback)
 end
 
-function TestRunner:_stream_queue()
-  local sender, receiver = async.control.channel.mpsc()
-
-  local producer = function(output_stream)
-    local orig = ""
-    local pending_data = nil
-    for data in output_stream do
-      orig = orig .. data
-      local ends_with_newline = vim.endswith(data, "\n")
-      local next_lines = vim.split(data, "\n", { plain = true, trimempty = true })
-      if pending_data then
-        next_lines[1] = pending_data .. next_lines[1]
-        pending_data = nil
-      end
-      if not ends_with_newline then
-        pending_data = table.remove(next_lines, #next_lines)
-      end
-      for _, line in ipairs(next_lines) do
-        sender.send(line)
-      end
-    end
-  end
-
-  local consumer = function()
-    return receiver.recv()
-  end
-  return producer, consumer
-end
-
 ---@param spec neotest.RunSpec
 ---@param adapter neotest.Adapter
 function TestRunner:_run_spec(spec, tree, args, adapter, results_callback)
   local position = tree:data()
   spec.strategy =
     vim.tbl_extend("force", spec.strategy or {}, config.strategies[args.strategy] or {})
-    spec.env = vim.tbl_extend("force", spec.env or {}, args.env or {})
-    spec.cwd = args.cwd or spec.cwd
-    if vim.tbl_isempty(spec.env or {}) then
-      spec.env = nil
-    end
-
+  spec.env = vim.tbl_extend("force", spec.env or {}, args.env or {})
+  spec.cwd = args.cwd or spec.cwd
+  if vim.tbl_isempty(spec.env or {}) then
+    spec.env = nil
+  end
 
   local proc_key = self:_create_process_key(adapter.name, position.id)
-  local producer, consumer = self:_stream_queue()
 
-  local process_result = self._processes:run(proc_key, spec, args, spec.stream and producer)
-  if spec.stream then
-    async.run(function()
-      for stream_results in spec.stream(consumer) do
-        results_callback(stream_results)
+  local stream_processor = spec.stream
+    and function(stream)
+      for stream_results in spec.stream(stream) do
+        results_callback(tree, stream_results)
       end
-    end)
-  end
-
-  local results = adapter.results(spec, process_result, tree)
-
-  if vim.tbl_isempty(results) then
-    results_callback(self:_fill_empty_results(tree, process_result.output))
-    return
-  end
-
-  self:fill_results(tree, results)
-
-  for _, result in pairs(results) do
-    if not result.output then
-      result.output = process_result.output
     end
-  end
+  local process_result = self._processes:run(proc_key, spec, args, stream_processor)
 
-  results_callback(results)
-end
+  local results = adapter.results(spec, process_result, tree)
 
-function TestRunner:_fill_empty_results(tree, output_path)
-  if #tree:children() == 0 then
-    return { [tree:data().id] = { status = "skipped", output = output_path } }
-  end
-  local results = {}
-  logger.warn("Results returned were empty, setting all positions to failed")
-  for _, pos in tree:iter() do
-    results[pos.id] = {
-      status = "failed",
-      errors = {},
-      output = output_path,
-    }
-  end
-  return results
+  results_callback(tree, results, process_result.output)
 end
 
 function TestRunner:_run_broken_down_tree(tree, args, adapter, results_callback)
@@ -211,7 +178,13 @@ end
 ---@async
 ---@param tree neotest.Tree
 ---@param results table<string, neotest.Result>
-function TestRunner:fill_results(tree, results)
+---@param partial? boolean
+function TestRunner:_missing_results(tree, results, partial)
+  local new_results = setmetatable({}, {
+    __index = function(_, key)
+      return results[key]
+    end,
+  })
   local root = tree:data()
   local missing_tests = {}
   for _, node in tree:iter_nodes() do
@@ -223,7 +196,6 @@ function TestRunner:fill_results(tree, results)
         if not lib.positions.contains(root, parent_pos) then
           break
         end
-
         local parent_result = results[parent_pos.id]
         local pos_result = results[pos.id]
         if not parent_result then
@@ -240,14 +212,21 @@ function TestRunner:fill_results(tree, results)
           parent_result.errors = vim.list_extend(parent_result.errors or {}, pos_result.errors)
         end
 
-        results[parent_pos.id] = parent_result
+        new_results[parent_pos.id] = parent_result
       end
     else
       if pos.type == "test" then
         missing_tests[#missing_tests + 1] = pos.id
       end
     end
   end
+  if partial then
+    for _, test_id in ipairs(missing_tests) do
+      for parent in tree:get_key(test_id):iter_parents() do
+        new_results[parent:data().id] = nil
+      end
+    end
+  end
 
   local root_result = results[root.id]
   for _, node in tree:iter_nodes() do
@@ -256,22 +235,18 @@ function TestRunner:fill_results(tree, results)
       if pos.type == "file" then
         -- Files not being present means that they were skipped (probably)
         if not results[pos.id] and root_result then
-          results[pos.id] = { status = "skipped", output = root_result.output }
+          new_results[pos.id] = { status = "skipped", output = root_result.output }
         end
       else
         -- Tests and namespaces not being present means that they failed to even start, count as root result
         if not results[pos.id] and root_result then
-          results[pos.id] = { status = root_result.status, output = root_result.output }
+          new_results[pos.id] = { status = root_result.status, output = root_result.output }
         end
       end
     end
   end
 
-  for _, test_id in ipairs(missing_tests) do
-    for parent in tree:get_key(test_id):iter_parents() do
-      results[parent:data().id] = nil
-    end
-  end
+  return new_results
 end
 
 return function(processes)

lua/neotest/client/state/init.lua

@@ -61,15 +61,17 @@ function NeotestClientState:update_positions(adapter_id, tree)
 end
 
 ---@param results table<string, neotest.Result>
-function NeotestClientState:update_results(adapter_id, results)
+function NeotestClientState:update_results(adapter_id, results, partial)
   logger.debug("New results for adapter", adapter_id)
   logger.trace(results)
   self._results[adapter_id] = vim.tbl_extend("force", self._results[adapter_id] or {}, results)
   if not self._running[adapter_id] then
     self._running[adapter_id] = {}
   end
-  for id, _ in pairs(results) do
-    self._running[adapter_id][id] = nil
+  if not partial then
+    for id, _ in pairs(results) do
+      self._running[adapter_id][id] = nil
+    end
   end
   self._events:emit(NeotestEvents.RESULTS, adapter_id, results)
 end