feat(testing): add coverage

.github/workflows/docker.yml

@@ -211,7 +211,8 @@ jobs:
     uses: ./.github/workflows/tests.yml
     secrets: inherit
     with:
-      cmd: "pytest --durations=0 -m 'not (tool or mujoco)'"
+      cmd: "_DIMOS_COV=1 coverage run -m pytest --durations=0 -m 'not (tool or mujoco)' && coverage combine && coverage html && coverage report"
+      upload-coverage: true
       dev-image: ros-dev:${{ (needs.check-changes.outputs.python == 'true' || needs.check-changes.outputs.dev == 'true' || needs.check-changes.outputs.ros == 'true') && needs.ros-dev.result == 'success' && needs.check-changes.outputs.branch-tag || 'dev' }}
 
   run-mypy:

.github/workflows/tests.yml

@@ -10,6 +10,10 @@ on:
      cmd:
        required: true
        type: string
+     upload-coverage:
+       required: false
+       type: boolean
+       default: false
 
 permissions:
   contents: read
@@ -43,6 +47,13 @@ jobs:
         run: |
           /entrypoint.sh bash -c "source .venv/bin/activate && ${{ inputs.cmd }}"
 
+      - name: Upload coverage report
+        if: inputs.upload-coverage && !cancelled()
+        uses: actions/upload-artifact@v4
+        with:
+          name: coverage-report
+          path: htmlcov/
+
       - name: check disk space
         if: failure()
         run: |

.gitignore

@@ -72,3 +72,8 @@ CLAUDE.MD
 
 /.mcp.json
 *.speedscope.json
+
+# Coverage
+htmlcov/
+.coverage
+.coverage.*

bin/coverage-by-author

@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+
+from collections import defaultdict
+import os
+import subprocess
+import sys
+
+from coverage import CoverageData
+
+
+def get_repo_root():
+    result = subprocess.run(
+        ["git", "rev-parse", "--show-toplevel"],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    return result.stdout.strip()
+
+
+def blame_authors(filepath):
+    """Return {line_number: author} for every source line via git blame."""
+    try:
+        result = subprocess.run(
+            ["git", "blame", "--line-porcelain", filepath],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+    except subprocess.CalledProcessError:
+        return {}
+
+    authors = {}
+    current_line = None
+    for line in result.stdout.splitlines():
+        # First line of each hunk: <sha> <orig_line> <final_line> [<num_lines>]
+        parts = line.split()
+        if (
+            len(parts) >= 3
+            and len(parts[0]) == 40
+            and all(c in "0123456789abcdef" for c in parts[0])
+        ):
+            current_line = int(parts[2])
+        elif line.startswith("author "):
+            if current_line is not None:
+                authors[current_line] = line[len("author ") :]
+    return authors
+
+
+def main():
+    repo_root = get_repo_root()
+
+    cov_file = os.path.join(repo_root, ".coverage")
+    if not os.path.exists(cov_file):
+        print("Error: .coverage not found. Run bin/pytest-coverage first.", file=sys.stderr)
+        sys.exit(1)
+
+    data = CoverageData(basename=cov_file)
+    data.read()
+
+    stats = defaultdict(lambda: {"total": 0, "covered": 0})
+
+    for abs_path in sorted(data.measured_files()):
+        # Convert to repo-relative path for git blame
+        try:
+            rel_path = os.path.relpath(abs_path, repo_root)
+        except ValueError:
+            continue
+        if rel_path.startswith(".."):
+            continue
+
+        authors = blame_authors(rel_path)
+        if not authors:
+            continue
+
+        covered_lines = set(data.lines(abs_path) or [])
+
+        for line_no, author in authors.items():
+            stats[author]["total"] += 1
+            if line_no in covered_lines:
+                stats[author]["covered"] += 1
+
+    if not stats:
+        print("No coverage data with git blame information found.", file=sys.stderr)
+        sys.exit(1)
+
+    sorted_authors = sorted(
+        stats.items(),
+        key=lambda x: x[1]["covered"] / x[1]["total"] if x[1]["total"] else 0,
+        reverse=True,
+    )
+
+    total_all = sum(s["total"] for _, s in sorted_authors)
+    covered_all = sum(s["covered"] for _, s in sorted_authors)
+
+    # Print table
+    hdr = f"{'Author':<30} {'Lines':>7} {'Covered':>9} {'Coverage':>10}"
+    sep = "\u2500" * len(hdr)
+    print(hdr)
+    print(sep)
+    for author, s in sorted_authors:
+        pct = 100.0 * s["covered"] / s["total"] if s["total"] else 0
+        print(f"{author:<30} {s['total']:>7} {s['covered']:>9} {pct:>9.1f}%")
+    print(sep)
+    pct_all = 100.0 * covered_all / total_all if total_all else 0
+    print(f"{'TOTAL':<30} {total_all:>7} {covered_all:>9} {pct_all:>9.1f}%")
+
+
+if __name__ == "__main__":
+    main()

bin/pytest-coverage

@@ -0,0 +1,12 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+rm -f .coverage .coverage.*
+
+export _DIMOS_COV=1
+
+uv run coverage run -m pytest "$@" -m 'not tool'
+uv run coverage combine
+uv run coverage html
+uv run coverage report

dimos/conftest.py

@@ -43,6 +43,10 @@ def pytest_configure(config):
     config.addinivalue_line("markers", "skipif_no_alibaba: skip when ALIBABA_API_KEY is not set")
     config.addinivalue_line("markers", "skipif_no_ros: skip when ROS dependencies are not present")
 
+    # Propagate coverage collection to subprocesses.
+    if os.environ.get("_DIMOS_COV"):
+        os.environ["COVERAGE_PROCESS_START"] = str(config.rootpath / "pyproject.toml")
+
 
 @pytest.hookimpl()
 def pytest_collection_modifyitems(config, items):

dimos/core/test_native_module.py

@@ -165,6 +165,12 @@ def test_autoconnect(args_file: str) -> None:
 
         # Custom transport was applied
         assert native.pointcloud.transport.topic.topic == "/my/custom/lidar"
+
+        # Wait for the native subprocess to write the output file
+        for _ in range(50):
+            if Path(args_file).exists():
+                break
+            time.sleep(0.1)
     finally:
         coordinator.stop()
 

pyproject.toml

@@ -209,7 +209,7 @@ dev = [
     "pytest-mock==3.15.0",
     "pytest-env==1.1.5",
     "pytest-timeout==2.4.0",
-    "coverage>=7.0",  # Required for numba compatibility (coverage.types)
+    "coverage>=7.0",
     "requests-mock==1.12.1",
     "terminaltexteffects==0.12.2",
     "watchdog>=3.0.0",
@@ -397,6 +397,16 @@ addopts = "-v -r a -p no:warnings --color=yes -m 'not (tool or slow or mujoco)'"
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"
 
+[tool.coverage.run]
+source = ["dimos"]
+parallel = true
+sigterm = true
+concurrency = ["multiprocessing", "thread"]
+
+[tool.coverage.report]
+show_missing = true
+skip_empty = true
+
 [tool.largefiles]
 max_size_kb = 50
 ignore = [