coverage.fetcher

Collect .gcda coverage from embedded (Zephyr LLEXT) targets over the console.

Unix hosts write .gcda to a filesystem that otto fetches with GcdaFetcher. Embedded RTOS targets have no filesystem: a coverage-instrumented LLEXT extension built against NASA’s embedded-gcov dumps its counters as an ASCII hexdump over the serial console (call_fn cov_dump → __gcov_exit). This module reconstructs the binary .gcda files from that capture, mirroring embedded-gcov’s serial_split.awk + xxd -r, so everything downstream (lcov merge → report) is reused unchanged.

The on-wire format, per source file:

Emitting <N> bytes for <path>.gcda
00000000: 61 64 63 67 2a 32 32 42 ...
...
<path>.gcda
Gcov End
otto.coverage.fetcher.embedded.decode_cov_dump(text)

Decode an embedded-gcov serial cov_dump capture.

Parameters:

text – Raw console output containing one or more emitted file blocks. ANSI colour codes and unrelated lines are tolerated.

Return type:

dict[str, bytes]

Returns:

Mapping of .gcda basename → reconstructed binary contents.

class otto.coverage.fetcher.embedded.EmbeddedGcdaCollector(staging_root, dump_command, pattern=None)

Bases: object

Collect coverage from embedded (Zephyr LLEXT) hosts over the console.

The embedded analogue of GcdaFetcher: each host gets its own staging_root/<host.id>/ subdirectory of decoded .gcda files, so the downstream merge/report layer treats embedded and Unix hosts identically.

dump_command is the console command that triggers the product’s coverage dump (e.g. llext call_fn <ext> cov_dump → __gcov_exit).

async collect_all()

Dump, decode and stage coverage from every matching embedded host.

Returns a mapping of host id → per-host staging directory. Non-embedded hosts, hosts with no coverage data, and failed dumps are omitted.

Return type:

dict[str, Path]

async otto.coverage.fetcher.embedded.collect_embedded_coverage(cov_config, staging_root, pattern=None)

Collect embedded coverage per the [coverage.embedded] config section.

Reads the product’s extension name from cov_config['embedded']['extension'] and dumps it via llext call_fn <ext> cov_dump (the conventional embedded-gcov __gcov_exit trigger) on every embedded host whose id matches pattern.

Parameters:

  • cov_config – The repo’s [coverage] table.

  • staging_root – Directory under which per-host .gcda is staged.

  • pattern – Optional compiled regex (the repo-declared [coverage].hosts selector) matched against each host’s id; None collects from every embedded host in the lab.

Returns {host_id: staging_dir}, or {} when no embedded coverage is configured (so the Unix-only path is unaffected).

Return type:

dict[str, Path]

Fetch .gcda files from remote hosts using otto’s file transfer.

Uses UnixHost.get() which supports SCP, SFTP, FTP, and netcat with progress tracking and multi-hop SSH chains.

class otto.coverage.fetcher.remote.GcdaFetcher(staging_root, pattern=None)

Bases: object

Fetch .gcda files from the configured lab hosts into a local staging area.

Each host gets its own subdirectory under staging_root so files from different hosts never collide before the merge step:

staging_root/
    host1_ne/
        foo.gcda
        subdir/bar.gcda
    host2_ne/
        foo.gcda

Hosts are selected via all_hosts(), optionally filtered by a compiled regex pattern matched against each host’s id.

async fetch_all(gcda_remote_dir)

Fetch .gcda files from every matching host concurrently.

Parameters:

gcda_remote_dir – Absolute path on each remote host where .gcda files are located (e.g. /var/coverage/myproduct).

Return type:

dict[str, Path]

Returns:

Mapping of host id → local staging directory containing its fetched .gcda files. Hosts with no files or failed transfers are omitted from the result.

async clean_remote(gcda_remote_dir)

Delete .gcda files from every matching remote host concurrently.

Should be called before a test run to ensure clean coverage data, and optionally after collection to save disk space.

Return type:

None