diff --git a/.github/testflinger-assets/test_binary_device_script.sh b/.github/testflinger-assets/test_binary_device_script.sh index 54516f22..e4adf2d5 100755 --- a/.github/testflinger-assets/test_binary_device_script.sh +++ b/.github/testflinger-assets/test_binary_device_script.sh @@ -29,4 +29,4 @@ set -e sudo kill ${JOURNAL_PID} || true # Exit with the test's exit status -exit $TEST_EXIT_STATUS +exit "$TEST_EXIT_STATUS" diff --git a/.github/testflinger-assets/test_snap_device_script.sh b/.github/testflinger-assets/test_snap_device_script.sh index 82bea702..cf6e8d9d 100755 --- a/.github/testflinger-assets/test_snap_device_script.sh +++ b/.github/testflinger-assets/test_snap_device_script.sh @@ -50,6 +50,12 @@ while sudo snap debug state /var/lib/snapd/state.json | grep -qE 'Doing|Undoing| sleep 10 done sudo snap connect fpgad:device-tree-overlays +echo " --- connecting to dfx-mgr-socket interface" +while sudo snap debug state /var/lib/snapd/state.json | grep -qE 'Doing|Undoing|Waiting'; do + echo " --- snapd internal tasks still running... waiting..." + sleep 10 +done +sudo snap connect fpgad:dfx-mgr-socket echo " --- connecting dbus interfaces" while sudo snap debug state /var/lib/snapd/state.json | grep -qE 'Doing|Undoing|Waiting'; do echo " --- snapd internal tasks still running... waiting..." @@ -60,11 +66,18 @@ sudo snap connect fpgad:cli-dbus fpgad:daemon-dbus echo "INFO: Done making necessary connections" echo "INFO: Running snap test script" -# NOTE: tarball contains "k24-starter-kits/..." and "k26-starter-kits/..." at tarball root from daemon/tests/test_data +# NOTE: test_data.gz contains "k24-starter-kits/..." and "k26-starter-kits/..." at tarball root from daemon/tests/test_data +# NOTE: tests.gz contains the test structure (common/, test_universal/, test_xlnx/, test_default/, etc.) mkdir -p fpgad/artifacts +echo " --- Extracting test data" tar -xzvf test_data.gz -C fpgad -sudo journalctl -f -n1 > fpgad/artifacts/journal.log 2>&1 & -JOURNAL_PID=$! -sudo python3 -u -m unittest ./snap_tests.py -v 2>&1 | tee fpgad/artifacts/snap_test.log -sudo kill ${JOURNAL_PID} || true +echo " --- Extracting tests" +mkdir -p tests +tar -xzvf tests.gz -C tests +echo " --- Saving timestamp for journal log retrieval" +TEST_START_TIME=$(date '+%Y-%m-%d %H:%M:%S') +echo " --- Running tests with unittest discovery" +sudo tests/snap_testing/test_snap.sh 2>&1 | tee fpgad/artifacts/snap_test.log +echo " --- Collecting journal logs since test start" +sudo journalctl --since "$TEST_START_TIME" -u "snap.fpgad*" > fpgad/artifacts/journal.log 2>&1 || true echo "INFO: Done running snap test script" diff --git a/.github/workflows/integration_tests.yaml.yml b/.github/workflows/integration_tests.yaml.yml index 26f3d21b..236ca0b9 100644 --- a/.github/workflows/integration_tests.yaml.yml +++ b/.github/workflows/integration_tests.yaml.yml @@ -132,12 +132,16 @@ jobs: tar -czvf test_data.gz -C daemon/tests/test_data/ . echo "::endgroup::" + echo "::group::Create tests tarball" + tar -czvf tests.gz -C tests/ . + echo "::endgroup::" + echo "::group::creating job.yaml" yq '.test_data.attachments = [ { "agent": "device_script.sh", "local": "'"$(pwd)"'/.github/testflinger-assets/test_snap_device_script.sh" }, { "agent": "fpgad.snap", "local": "'"$(pwd)"'/fpgad.snap" }, { "agent": "test_data.gz", "local": "'"$(pwd)"'/test_data.gz" }, - { "agent": "snap_tests.py", "local": "'"$(pwd)"'/tests/snap_tests.py" } + { "agent": "tests.gz", "local": "'"$(pwd)"'/tests.gz" } ]' .github/testflinger-assets/testflinger_job.yaml | tee job.yaml echo "::endgroup::" diff --git a/Cargo.lock b/Cargo.lock index 5f7b859f..f1aeb85f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -287,7 +287,7 @@ checksum = "784a4df722dc6267a04af36895398f59d21d07dce47232adf31ec0ff2fa45e67" [[package]] name = "fpgad" -version = "0.2.0" +version = "0.2.1" dependencies = [ "env_logger", "fdt", @@ -302,7 +302,7 @@ dependencies = [ [[package]] name = "fpgad_cli" -version = "0.2.0" +version = "0.2.1" dependencies = [ "clap", "env_logger", @@ -313,7 +313,7 @@ dependencies = [ [[package]] name = "fpgad_macros" -version = "0.2.0" +version = "0.2.1" dependencies = [ "quote", "syn", diff --git a/Cargo.toml b/Cargo.toml index aa20d386..28db6b2a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ resolver = "3" members = ["daemon", "cli", "fpgad_macros"] [workspace.package] -version = "0.2.0" +version = "0.2.1" edition = "2024" license = "GPL-3.0" homepage = "https://github.com/canonical/fpgad" diff --git a/cli/src/load.rs b/cli/src/load.rs index fdac8035..db9f96b5 100644 --- a/cli/src/load.rs +++ b/cli/src/load.rs @@ -124,6 +124,7 @@ async fn call_apply_overlay( /// /// # Arguments /// +/// * `platform_override` - Optional platform string to bypass platform detection /// * `dev_handle` - Optional [device handle](../index.html#device-handles) (e.g., "fpga0") /// * `file_path` - Path to the overlay file (.dtbo) /// * `overlay_handle` - Optional [overlay handle](../index.html#overlay-handles) for the overlay directory @@ -133,28 +134,40 @@ async fn call_apply_overlay( /// * `Err(zbus::Error)` - DBus communication error, device detection failure, or FpgadError. /// See [Error Handling](../index.html#error-handling) for details. async fn apply_overlay( + platform_override: &Option, dev_handle: &Option, file_path: &str, overlay_handle: &Option, ) -> Result { // Determine platform and overlay handle based on provided parameters - let (platform, overlay_handle_to_use) = match (dev_handle, overlay_handle) { - // Both are provided - (Some(dev), Some(overlay)) => (call_get_platform_type(dev).await?, overlay.clone()), + let (platform, overlay_handle_to_use) = match (platform_override, dev_handle, overlay_handle) { + // Platform override provided - use it regardless of other params + (Some(plat), _, Some(overlay)) => (plat.clone(), overlay.clone()), + (Some(plat), Some(dev), None) => (plat.clone(), dev.clone()), + (Some(plat), None, None) => { + let overlay = get_first_device_handle() + .await + .unwrap_or("overlay0".to_string()); + (plat.clone(), overlay) + } + + // No platform override - use detection logic + // Both device and overlay are provided + (None, Some(dev), Some(overlay)) => (call_get_platform_type(dev).await?, overlay.clone()), // dev_handle provided, overlay_handle not provided so use device name as overlay handle - (Some(dev), None) => { + (None, Some(dev), None) => { let platform = call_get_platform_type(dev).await?; (platform, dev.clone()) } // dev_handle not provided, so use first platform - (None, Some(overlay)) => { + (None, None, Some(overlay)) => { let platform = get_first_platform().await?; (platform, overlay.clone()) } // neither provided so get first device to and use its platform as platform and its name as // overlay_handle - (None, None) => { + (None, None, None) => { // this saves making two dbus calls by getting it all from the hashmap let platforms = call_get_platform_types().await?; let platform = platforms @@ -186,6 +199,7 @@ async fn apply_overlay( /// /// # Arguments /// +/// * `platform_override` - Optional platform string to bypass platform detection /// * `device_handle` - Optional [device handle](../index.html#device-handles) (e.g., "fpga0") /// * `file_path` - Path to the bitstream file /// @@ -194,6 +208,7 @@ async fn apply_overlay( /// * `Err(zbus::Error)` - DBus communication error, device detection failure, or FpgadError. /// See [Error Handling](../index.html#error-handling) for details. async fn load_bitstream( + platform_override: &Option, device_handle: &Option, file_path: &str, ) -> Result { @@ -201,7 +216,11 @@ async fn load_bitstream( None => get_first_device_handle().await?, Some(dev) => dev.clone(), }; - call_load_bitstream("", &dev, &sanitize_path_str(file_path)?, "").await + let platform_str = match platform_override { + None => "", + Some(plat) => plat.as_str(), + }; + call_load_bitstream(platform_str, &dev, &sanitize_path_str(file_path)?, "").await } /// Main handler for the load command. @@ -212,6 +231,7 @@ async fn load_bitstream( /// /// # Arguments /// +/// * `platform_override` - Optional platform string to bypass platform detection /// * `dev_handle` - Optional [device handle](../index.html#device-handles) /// * `sub_command` - The load subcommand specifying what to load (overlay or bitstream) /// @@ -220,13 +240,16 @@ async fn load_bitstream( /// * `Err(zbus::Error)` - DBus communication error, operation failure, or FpgadError. /// See [Error Handling](../index.html#error-handling) for details. pub async fn load_handler( + platform_override: &Option, dev_handle: &Option, sub_command: &LoadSubcommand, ) -> Result { match sub_command { - LoadSubcommand::Overlay { file, handle } => { - apply_overlay(dev_handle, file.as_ref(), handle).await + LoadSubcommand::Overlay { file, name } => { + apply_overlay(platform_override, dev_handle, file.as_ref(), name).await + } + LoadSubcommand::Bitstream { file } => { + load_bitstream(platform_override, dev_handle, file.as_ref()).await } - LoadSubcommand::Bitstream { file } => load_bitstream(dev_handle, file.as_ref()).await, } } diff --git a/cli/src/main.rs b/cli/src/main.rs index a9cb0722..9c0468f9 100644 --- a/cli/src/main.rs +++ b/cli/src/main.rs @@ -67,58 +67,69 @@ //! Usage: [snap run] fpgad [OPTIONS] //! //! OPTIONS: -//! -h, --help Print help -//! --handle fpga device `HANDLE` to be used for the operations. -//! Default value for this option is calculated in runtime -//! and the application picks the first available fpga device -//! in the system (under `/sys/class/fpga_manager/`) +//! -h, --help Print help +//! -p, --platform Platform override string (bypasses platform detection logic). +//! When provided, this platform string is passed directly to the +//! daemon instead of auto-detecting from the device handle. +//! Examples: "universal", "xlnx,zynqmp-pcap-fpga" +//! -d, --device FPGA device handle to be used for the operations. +//! Default value is calculated at runtime - the application +//! picks the first available FPGA device in the system +//! (under `/sys/class/fpga_manager/`). +//! Examples: "fpga0", "fpga1" +//! +//! SUBCOMMAND OPTIONS: +//! -n, --name (Used with load/remove overlay subcommands) +//! Name for the overlay directory in configfs +//! (under `/sys/kernel/config/device-tree/overlays/`). +//! If not provided, defaults to the device handle or "overlay0". //! //! COMMANDS: //! ├── load Load a bitstream or overlay -//! │ ├── overlay [--handle ] +//! │ ├── overlay [--name --platform ] //! │ │ Load overlay (.dtbo) into the system using the default OVERLAY_HANDLE //! │ │ (either the provided DEVICE_HANDLE or "overlay0") or provide -//! │ │ --handle: to name the overlay directory -//! │ └── bitstream +//! │ │ --name: to name the overlay directory +//! │ └── bitstream [--platform ] //! │ Load bitstream (e.g. `.bit.bin` file) into the FPGA //! │ //! ├── set //! │ Set an attribute/flag under `/sys/class/fpga_manager//` //! │ -//! ├── status [--handle ] +//! ├── status [--device --platform ] //! │ Show FPGA status (all devices and overlays) or provide -//! │ --handle: for a specific device status +//! │ --device: for a specific device status //! │ //! └── remove Remove an overlay or bitstream -//! ├── overlay [--handle ] +//! ├── overlay [--name --platform ] //! │ Removes the first overlay found (call repeatedly to remove all) or provide -//! │ --handle: to remove overlay previously loaded with given OVERLAY_HANDLE -//! └── bitstream +//! │ --name: to remove overlay previously loaded with given OVERLAY_HANDLE +//! └── bitstream [--name --platform ] //! Remove active bitstream from FPGA (bitstream removal is vendor specific) //! ``` //! //! ### Loading //! //! ```shell -//! fpgad [--handle=] load ( (overlay [--handle=]) | (bitstream ) ) +//! fpgad [--device=] [--platform=] load ( (overlay [--name=]) | (bitstream ) ) //! ``` //! //! ### Removing //! //! ```shell -//! fpgad [--handle=] remove ( ( overlay ) | ( bitstream ) ) +//! fpgad [--device=] [--platform=] remove ( ( overlay [--name=] ) | ( bitstream ) ) //! ``` //! //! ### Set //! //! ```shell -//! fpgad [--handle=] set ATTRIBUTE VALUE +//! fpgad [--device=] set ATTRIBUTE VALUE //! ``` //! //! ### Status //! //! ```shell -//! fpgad [--handle=] status +//! fpgad [--device=] [--platform=] status //! ``` //! //! ## examples (for testing) @@ -127,32 +138,36 @@ //! //! ```shell //! sudo ./target/debug/cli load bitstream /lib/firmware/k26-starter-kits.bit.bin -//! sudo ./target/debug/cli --handle=fpga0 load bitstream /lib/firmware/k26-starter-kits.bit.bin +//! sudo ./target/debug/cli --device=fpga0 load bitstream /lib/firmware/k26-starter-kits.bit.bin +//! sudo ./target/debug/cli --platform=universal load bitstream /lib/firmware/k26-starter-kits.bit.bin +//! sudo ./target/debug/cli --platform=xlnx load bitstream /lib/firmware/k26-starter-kits.bit.bin //! //! sudo ./target/debug/cli load overlay /lib/firmware/k26-starter-kits.dtbo -//! sudo ./target/debug/cli load overlay /lib/firmware/k26-starter-kits.dtbo --handle=overlay_handle -//! sudo ./target/debug/cli --handle=fpga0 load overlay /lib/firmware/k26-starter-kits.dtbo --handle=overlay_handle +//! sudo ./target/debug/cli load overlay /lib/firmware/k26-starter-kits.dtbo --name=overlay_handle +//! sudo ./target/debug/cli --device=fpga0 load overlay /lib/firmware/k26-starter-kits.dtbo --name=overlay_handle +//! sudo ./target/debug/cli --platform=universal load overlay /lib/firmware/k26-starter-kits.dtbo --name=overlay_handle +//! sudo ./target/debug/cli --platform=xlnx --device=fpga0 load overlay /lib/firmware/k26-starter-kits.dtbo --name=overlay_handle //! ``` //! //! ### Remove //! //! ```shell -//! sudo ./target/debug/cli --handle=fpga0 remove overlay -//! sudo ./target/debug/cli --handle=fpga0 remove overlay --handle=overlay_handle +//! sudo ./target/debug/cli --device=fpga0 remove overlay +//! sudo ./target/debug/cli --device=fpga0 remove overlay --name=overlay_handle //! ``` //! //! ### Set //! //! ```shell //! sudo ./target/debug/cli set flags 0 -//! sudo ./target/debug/cli --handle=fpga0 set flags 0 +//! sudo ./target/debug/cli --device=fpga0 set flags 0 //! ``` //! //! ### Status //! //! ```shell //! ./target/debug/cli status -//! ./target/debug/cli --handle=fpga0 status +//! ./target/debug/cli --device=fpga0 status //! ``` mod proxies; @@ -190,19 +205,28 @@ use std::error::Error; /// # Check status of all FPGA devices /// fpgad status /// -/// # Load an overlay with a specific handle -/// fpgad load overlay /lib/firmware/overlay.dtbo --handle=my_overlay +/// # Load an overlay with a specific name +/// fpgad load overlay /lib/firmware/overlay.dtbo --name=my_overlay /// /// ``` #[derive(Parser, Debug)] #[command(name = "fpga")] #[command(bin_name = "fpga")] struct Cli { - /// fpga device `HANDLE` to be used for the operations. - /// Default value for this option is calculated in runtime and the application - /// picks the first available fpga in the system (under /sys/class/fpga_manager) - #[arg(long = "handle")] - handle: Option, + /// Platform override string (bypasses platform detection logic). + /// When provided, this platform string is passed directly to the daemon + /// instead of auto-detecting from the device handle. + /// Examples: "universal", "xlnx,zynqmp-pcap-fpga" + #[arg(short = 'p', long = "platform")] + platform: Option, + + /// FPGA `device` handle to be used for the operations. + /// Default value is calculated at runtime - the application picks the first + /// available FPGA device in the system (under /sys/class/fpga_manager/). + /// Examples: "fpga0", "fpga1" + #[arg(short = 'd', long = "device")] + device: Option, + #[command(subcommand)] command: Commands, } @@ -215,6 +239,16 @@ struct Cli { /// /// Device tree overlays are typically loaded before or after bitstreams to properly configure /// the kernel's view of the FPGA's hardware interfaces and peripherals. +/// +/// # Examples +/// +/// ```shell +/// # Load a bitstream +/// fpgad load bitstream [-d= -p=] /lib/firmware/design.bit.bin +/// +/// # Load an overlay with a custom name +/// fpgad load overlay [-d= -p=] /lib/firmware/overlay.dtbo [-n=my_overlay] +/// ``` #[derive(Subcommand, Debug)] enum LoadSubcommand { /// Load overlay into the system @@ -222,10 +256,11 @@ enum LoadSubcommand { /// Overlay `FILE` to be loaded (typically .dtbo) file: String, - /// `HANDLE` for the overlay directory which will be created - /// under "/sys/kernel/config/device-tree/overlays" - #[arg(long = "handle")] - handle: Option, + /// Name for the overlay directory which will be created + /// under "/sys/kernel/config/device-tree/overlays/". + /// If not provided, defaults to the device handle or "overlay0". + #[arg(short = 'n', long = "name")] + name: Option, }, /// Load bitstream into the system Bitstream { @@ -237,28 +272,40 @@ enum LoadSubcommand { /// Subcommands for removing FPGA components. /// /// This enum defines the types of components that can be removed from an FPGA device: -/// - **Overlay**: Removes a device tree overlay by its handle. +/// - **Overlay**: Removes a device tree overlay by its name. /// - **Bitstream**: Intended to remove the currently loaded FPGA bitstream (vendor-specific -/// operation) +/// operation that may use slot identifiers on platforms like DFX Manager) /// /// Removing overlays is important for proper cleanup when reconfiguring the FPGA. /// Bitstream removal support depends on the FPGA vendor and platform capabilities. +/// +/// # Examples +/// +/// ```shell +/// # Remove the first overlay found +/// fpgad remove overlay +/// +/// # Remove a specific overlay by name +/// fpgad [-d=] [-p=] remove overlay -n=my_overlay +/// +/// # Remove a bitstream, if supported +/// fpgad [-d=] [-p=] remove bitstream -n=0 # for dfx-mgr slot 0 +/// ``` #[derive(Subcommand, Debug)] enum RemoveSubcommand { - /// Remove overlay with the `HANDLE` provided + /// Remove overlay with the name provided Overlay { - /// `HANDLE` is the handle that is given during `load` operation - /// it is different than device_handle which is being used for platform - /// detection logic. - #[arg(long = "handle")] - handle: Option, + /// Name of the overlay to remove (as given during `load` operation). + /// If not provided, removes the first overlay found in the system. + /// This is different from device_handle which is used for platform detection. + #[arg(short = 'n', long = "name")] + name: Option, }, - /// Remove bitstream loaded in given `HANDLE` to fpga command + /// Remove bitstream loaded in the given device Bitstream { - /// `HANDLE` is the handle that is given during `load` operation - /// TODO(Artie): document - for dfxmgr it will be the slot - use "" to allow remove latest - /// it is different than device_handle which is being used for platform - /// detection logic. + /// Handle/identifier for the bitstream to remove. + /// For DFX Manager platforms, this can be a slot ID. + /// Use empty string "" to remove the latest bitstream. #[arg(long = "handle")] handle: Option, }, @@ -274,6 +321,25 @@ enum RemoveSubcommand { /// /// Each command communicates with the fpgad daemon via DBus to perform privileged /// operations on FPGA devices managed through the Linux kernel's FPGA subsystem. +/// +/// # Examples +/// +/// ```shell +/// # Load a bitstream to a specific device +/// fpgad --device=fpga0 load bitstream /lib/firmware/design.bit.bin +/// +/// # Load an overlay with platform override +/// fpgad --platform=universal load overlay /lib/firmware/overlay.dtbo --name=my_overlay +/// +/// # Set flags for a device +/// fpgad --device=fpga0 set flags 0 +/// +/// # Get status for all devices +/// fpgad status +/// +/// # Remove an overlay by name +/// fpgad remove overlay --name=my_overlay +/// ``` #[derive(Subcommand, Debug)] enum Commands { /// Load a bitstream or an overlay for the given device handle @@ -331,10 +397,12 @@ async fn main() -> Result<(), Box> { let cli = Cli::parse(); debug!("parsed cli command with {cli:#?}"); let result = match cli.command { - Commands::Load { command } => load_handler(&cli.handle, &command).await, - Commands::Remove { command } => remove_handler(&cli.handle, &command).await, - Commands::Set { attribute, value } => set_handler(&cli.handle, &attribute, &value).await, - Commands::Status => status_handler(&cli.handle).await, + Commands::Load { command } => load_handler(&cli.platform, &cli.device, &command).await, + Commands::Remove { command } => remove_handler(&cli.platform, &cli.device, &command).await, + Commands::Set { attribute, value } => { + set_handler(&cli.platform, &cli.device, &attribute, &value).await + } + Commands::Status => status_handler(&cli.platform, &cli.device).await, }; match result { Ok(msg) => { diff --git a/cli/src/remove.rs b/cli/src/remove.rs index c0b2b7cb..8eb1cfa4 100644 --- a/cli/src/remove.rs +++ b/cli/src/remove.rs @@ -42,6 +42,7 @@ use zbus::Connection; /// /// # Arguments /// +/// * `platform_string` - Platform identifier string (empty string for auto-detection) /// * `device_handle` - Platform identifier for the [device](../index.html#device-handles) /// * `bitstream_handle` - the identifier of the bitstream (can be slot ID for dfx-mgr) TODO(Artie): update docs /// @@ -50,13 +51,14 @@ use zbus::Connection; /// * `Err(zbus::Error)` - DBus communication error, invalid handle(s), or FpgadError. /// See [Error Handling](../index.html#error-handling) for details. async fn call_remove_bitstream( + platform_string: &str, device_handle: &str, bitstream_handle: &str, ) -> Result { let connection = Connection::system().await?; let proxy = control_proxy::ControlProxy::new(&connection).await?; proxy - .remove_bitstream("", device_handle, bitstream_handle) + .remove_bitstream(platform_string, device_handle, bitstream_handle) .await } @@ -93,6 +95,7 @@ async fn call_remove_overlay( /// /// # Arguments /// +/// * `platform_override` - Optional platform string to bypass platform detection /// * `device_handle` - Optional [device handle](../index.html#device-handles) for platform detection /// * `overlay_handle` - Optional [overlay handle](../index.html#overlay-handles) of the specific overlay to remove /// @@ -101,12 +104,16 @@ async fn call_remove_overlay( /// * `Err(zbus::Error)` - DBus communication error, detection failure, or FpgadError. /// See [Error Handling](../index.html#error-handling) for details. async fn remove_overlay( + platform_override: &Option, device_handle: &Option, overlay_handle: &Option, ) -> Result { - let platform_string = match device_handle { - None => get_first_platform().await?, - Some(dev) => call_get_platform_type(dev).await?, + let platform_string = match platform_override { + Some(plat) => plat.clone(), + None => match device_handle { + None => get_first_platform().await?, + Some(dev) => call_get_platform_type(dev).await?, + }, }; let handle = match overlay_handle { Some(handle) => handle.clone(), @@ -123,9 +130,16 @@ async fn remove_overlay( /// vendor-specific and depends on platform capabilities that may be added /// through softener implementations in the future. /// +/// # Arguments +/// +/// * `platform_override` - Optional platform string to bypass platform detection +/// * `device_handle` - Optional [device handle](../index.html#device-handles) +/// * `bitstream_handle` - Optional bitstream/slot identifier +/// /// # Returns: `Result` /// * `Err(zbus::Error)` - Always returns "Not implemented" error async fn remove_bitstream( + platform_override: &Option, device_handle: &Option, bitstream_handle: &Option, ) -> Result { @@ -137,7 +151,11 @@ async fn remove_bitstream( Some(handle) => handle, None => "", }; - call_remove_bitstream(dev, handle).await + let platform_str = match platform_override { + None => "", + Some(plat) => plat.as_str(), + }; + call_remove_bitstream(platform_str, dev, handle).await } /// Main handler for the remove command. @@ -148,6 +166,7 @@ async fn remove_bitstream( /// /// # Arguments /// +/// * `platform_override` - Optional platform string to bypass platform detection /// * `dev_handle` - Optional [device handle](../index.html#device-handles) /// * `sub_command` - The remove subcommand specifying what to remove (overlay or bitstream) /// @@ -156,11 +175,16 @@ async fn remove_bitstream( /// * `Err(zbus::Error)` - DBus communication error, operation failure, or FpgadError. /// See [Error Handling](../index.html#error-handling) for details. pub async fn remove_handler( + platform_override: &Option, dev_handle: &Option, sub_command: &RemoveSubcommand, ) -> Result { match sub_command { - RemoveSubcommand::Overlay { handle } => remove_overlay(dev_handle, handle).await, - RemoveSubcommand::Bitstream { handle } => remove_bitstream(dev_handle, handle).await, + RemoveSubcommand::Overlay { name } => { + remove_overlay(platform_override, dev_handle, name).await + } + RemoveSubcommand::Bitstream { handle } => { + remove_bitstream(platform_override, dev_handle, handle).await + } } } diff --git a/cli/src/set.rs b/cli/src/set.rs index d2347414..26d5bed2 100644 --- a/cli/src/set.rs +++ b/cli/src/set.rs @@ -148,6 +148,7 @@ async fn call_write_property(property: &str, value: &str) -> Result Result, device_handle: &Option, attribute: &str, value: &str, ) -> Result { + // TODO(Artie): read and write property to be platform specific functionalities in daemon. let property_path = match device_handle { None => build_property_path(&get_first_device_handle().await?, attribute)?, Some(dev) => build_property_path(dev, attribute)?, diff --git a/cli/src/status.rs b/cli/src/status.rs index f2316512..4a322e4a 100644 --- a/cli/src/status.rs +++ b/cli/src/status.rs @@ -384,13 +384,15 @@ async fn get_full_status_message() -> Result { /// Main handler for the status command. /// /// Dispatches to the appropriate status query function based on whether a specific -/// device handle is provided. If a device handle is given, returns status for that -/// device only; otherwise returns comprehensive status for all devices and overlays. +/// device handle or platform override is provided. If a device handle is given, returns +/// status for that device only; if a platform override is given, returns status for that +/// platform; otherwise returns comprehensive status for all devices and overlays. /// This is the entry point called by the CLI's main function when a status command /// is issued. /// /// # Arguments /// +/// * `platform_override` - Optional platform string to bypass platform detection /// * `device_handle` - Optional [device handle](../index.html#device-handles) for querying a specific device /// /// # Returns: `Result` @@ -402,15 +404,25 @@ async fn get_full_status_message() -> Result { /// /// ```rust,ignore /// // Get status for all devices -/// let status = status_handler(&None).await?; +/// let status = status_handler(&None, &None).await?; /// /// // Get status for a specific device -/// let status = status_handler(&Some("fpga0".to_string())).await?; +/// let status = status_handler(&None, &Some("fpga0".to_string())).await?; +/// +/// // Get status for a specific platform +/// let status = status_handler(&Some("xlnx,zynqmp-pcap-fpga".to_string()), &None).await?; /// ``` -pub async fn status_handler(device_handle: &Option) -> Result { - let ret_string = match device_handle { - None => get_full_status_message().await?, - Some(dev) => get_fpga_state_message(dev.as_str()).await?, +pub async fn status_handler( + platform_override: &Option, + device_handle: &Option, +) -> Result { + let ret_string = match (platform_override, device_handle) { + // Platform override takes precedence + (Some(plat), _) => call_get_status_message(plat).await?, + // Then device handle + (None, Some(dev)) => get_fpga_state_message(dev.as_str()).await?, + // Default to full status + (None, None) => get_full_status_message().await?, }; Ok(ret_string) } diff --git a/daemon/src/comm/dbus/status_interface.rs b/daemon/src/comm/dbus/status_interface.rs index 1efa7eaf..57f720f0 100644 --- a/daemon/src/comm/dbus/status_interface.rs +++ b/daemon/src/comm/dbus/status_interface.rs @@ -20,6 +20,7 @@ use crate::platforms::platform::{list_fpga_managers, read_compatible_string}; use crate::platforms::platform::{platform_for_known_platform, platform_from_compat_or_device}; use crate::comm::dbus::{fs_read_property, validate_device_handle}; +use crate::error::FpgadError; use crate::system_io::fs_read_dir; use log::{error, info}; use zbus::{fdo, interface}; @@ -33,6 +34,9 @@ pub struct StatusInterface {} impl StatusInterface { async fn get_status_message(&self, platform_string: &str) -> Result { info!("get_fpga_state called with platform_string: {platform_string}"); + if platform_string.is_empty() { + return Err(FpgadError::Argument("Empty platform string - cannot determine how to get status message becuase cannot determine platform to use without platform string".to_string()).into()); + } let platform = platform_from_compat_or_device(platform_string, "")?; Ok(platform.status_message()?) } diff --git a/daemon/src/main.rs b/daemon/src/main.rs index f0eaee50..4316fc47 100644 --- a/daemon/src/main.rs +++ b/daemon/src/main.rs @@ -121,6 +121,17 @@ async fn main() -> Result<(), Box> { env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init(); register_platforms(); + // If running in a snap environment, try to start the softener daemon wrapper + #[cfg(feature = "softeners")] + { + if std::env::var("SNAP").is_ok() { + info!("SNAP environment detected, starting softener daemon wrapper"); + tokio::spawn(async { + softeners::softeners_thread::run_softener_daemons().await; + }); + } + } + // Upon load, the daemon will search each fpga device and determine what platform it is // based on its name in /sys/class/fpga_manager/{device}/name let status_interface = StatusInterface {}; diff --git a/daemon/src/softeners.rs b/daemon/src/softeners.rs index 514ea193..7204da3a 100644 --- a/daemon/src/softeners.rs +++ b/daemon/src/softeners.rs @@ -13,5 +13,6 @@ pub mod error; +pub mod softeners_thread; #[cfg(feature = "xilinx-dfx-mgr")] pub mod xilinx_dfx_mgr; diff --git a/daemon/src/softeners/softeners_thread.rs b/daemon/src/softeners/softeners_thread.rs new file mode 100644 index 00000000..74e04354 --- /dev/null +++ b/daemon/src/softeners/softeners_thread.rs @@ -0,0 +1,518 @@ +// This file is part of fpgad, an application to manage FPGA subsystem together with device-tree and kernel modules. +// +// Copyright 2026 Canonical Ltd. +// +// SPDX-License-Identifier: GPL-3.0-only +// +// fpgad is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License version 3, as published by the Free Software Foundation. +// +// fpgad is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranties of MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License along with this program. If not, see http://www.gnu.org/licenses/. + +//! Softener daemon lifecycle management. +//! +//! This module provides infrastructure for managing external daemon processes required +//! by vendor-specific FPGA softener implementations (such as dfx-mgrd for Xilinx devices). +//! It handles daemon discovery, startup, monitoring, and graceful shutdown. +//! +//! # Components +//! +//! * [`DaemonConfig`] - Configuration for individual daemon processes +//! * [`DaemonManager`] - Lifecycle manager that starts, monitors, and stops daemons +//! +//! # Daemon Discovery +//! +//! At startup, the manager checks for daemon binaries at configured paths (typically +//! under `$SNAP/usr/bin/`). Only daemons whose binaries are found will be started. +//! Socket files are monitored to confirm successful startup. +//! +//! # Lifecycle +//! +//! The daemon manager runs until the task is cancelled or the program exits. Status of all +//! daemons is checked every 5 seconds. When a daemon exits unexpectedly, the manager attempts +//! to restart it up to 5 times. If restart attempts fail repeatedly, that daemon is abandoned +//! while others continue running. All managed processes are automatically cleaned up when the +//! manager is dropped. +//! +//! # Logging +//! +//! Each daemon's stdout and stderr are redirected to log files under `$SNAP_COMMON/log/` +//! (or `/tmp/log/` if not in a snap). + +use crate::error::FpgadError; +use crate::softeners::error::FpgadSoftenerError; +use log::{error, info, warn}; +use std::collections::HashMap; +use std::env; +use std::fs::{self, OpenOptions}; +use std::os::unix::fs::FileTypeExt; // for "is_socket" +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::time::Duration; +use tokio::time::sleep; + +/// Configuration for a softener daemon. +/// +/// Defines the parameters needed to start and monitor a daemon process, including +/// binary location, socket path for health checking, and error handling. +/// +/// # Fields +/// +/// * `name` - Name of the daemon +/// * `binary_path` - Path to the daemon binary +/// * `socket_path` - Path to the daemon's socket file (used for startup verification) +/// * `start_timeout` - Timeout in seconds to wait for daemon startup +/// * `log_file` - Optional path to log file (will be auto-generated if None) +/// * `error_constructor` - Function to create the appropriate error type +#[derive(Debug, Clone)] +pub struct DaemonConfig { + /// Name of the daemon + pub name: String, + /// Path to the daemon binary + pub binary_path: PathBuf, + /// Path to the daemon's socket file + pub socket_path: PathBuf, + /// Timeout in seconds to wait for daemon startup + pub start_timeout: u64, + /// Optional path to log file (will be auto-generated if None) + pub log_file: Option, + /// Function to create the appropriate error type + pub error_constructor: fn(String) -> FpgadSoftenerError, +} + +impl DaemonConfig { + /// Create a new daemon configuration with auto-generated log file path. + /// + /// # Arguments + /// + /// * `name` - Name of the daemon + /// * `binary_path` - Path to the daemon binary + /// * `socket_path` - Path to the daemon's socket file + /// * `start_timeout` - Timeout in seconds to wait for daemon startup + /// * `error_constructor` - Function to create the appropriate error type + /// + /// # Returns: `Self` + /// * New DaemonConfig instance with auto-generated log file path + pub fn new( + name: String, + binary_path: PathBuf, + socket_path: PathBuf, + start_timeout: u64, + error_constructor: fn(String) -> FpgadSoftenerError, + ) -> Self { + Self { + name, + binary_path, + socket_path, + start_timeout, + log_file: None, + error_constructor, + } + } + + /// Get the log file path, creating default if not set. + fn log_file(&self) -> PathBuf { + if let Some(ref log_file) = self.log_file { + log_file.clone() + } else { + let snap_common = std::env::var("SNAP_COMMON").unwrap_or_else(|_| "/tmp".to_string()); + let log_dir = PathBuf::from(snap_common).join("log"); + // Create log directory if it doesn't exist + let _ = fs::create_dir_all(&log_dir); + log_dir.join(format!("{}.log", self.name)) + } + } +} + +/// Manages lifecycle of softener daemons. +pub struct DaemonManager { + daemons: Vec, + processes: HashMap, +} + +impl DaemonManager { + /// Maximum number of times to attempt restarting a failed daemon before giving up. + const MAX_RESTART_ATTEMPTS: u32 = 5; + /// Number of seconds to wait between monitoring loops + const MONITOR_RATE: Duration = Duration::from_secs(5); + /// Number of seconds to wait between restart attempts + const RESTART_DELAY: Duration = Duration::from_secs(1); + + /// Initialize with list of daemons to manage. + /// + /// # Arguments + /// + /// * `daemons` - Vector of daemon configurations to manage + /// + /// # Returns: `Self` + /// * New DaemonManager instance + pub fn new(daemons: Vec) -> Self { + Self { + daemons, + processes: HashMap::new(), + } + } + + /// Filter configured daemons to only those with available binaries. + fn filter_available_daemons(&self) -> Vec { + let mut available = Vec::new(); + + for daemon in &self.daemons { + if daemon.binary_path.is_file() { + info!( + "Detected {} at {}", + daemon.name, + daemon.binary_path.display() + ); + available.push(daemon.clone()); + } else { + info!( + "{} not found at {}, skipping", + daemon.name, + daemon.binary_path.display() + ); + } + } + + available + } + + /// Remove stale socket file if it exists. + fn cleanup_stale_socket(socket_path: &Path) { + if socket_path.exists() { + match fs::metadata(socket_path) { + Ok(metadata) => { + if metadata.file_type().is_socket() { + info!("Removing stale socket at {}", socket_path.display()); + if let Err(e) = fs::remove_file(socket_path) { + warn!("Could not remove stale socket: {}", e); + } + } + } + Err(e) => { + warn!("Warning: Could not stat socket file: {}", e); + } + } + } + } + + /// Start a daemon and wait for its socket to appear. + /// + /// # Arguments + /// + /// * `daemon` - Configuration for the daemon to start + /// + /// # Returns: `Result<(), FpgadError>` + /// * `Ok(())` - Daemon started successfully and socket appeared + /// * `Err(FpgadError::Softener)` - Failed to start daemon or socket didn't appear + fn start_daemon(&mut self, daemon: &DaemonConfig) -> Result<(), FpgadError> { + info!("Starting {}...", daemon.name); + + // Clean up any stale socket + Self::cleanup_stale_socket(&daemon.socket_path); + + let log_file_path = daemon.log_file(); + + // Open log file + let log_file = OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .open(&log_file_path) + .map_err(|e| { + FpgadError::Softener((daemon.error_constructor)(format!( + "Failed to open log file: {}", + e + ))) + })?; + + // Create stdio objects for log file + let log_stdout = Stdio::from(log_file.try_clone().map_err(|e| { + FpgadError::Softener((daemon.error_constructor)(format!( + "Failed to clone log file: {}", + e + ))) + })?); + let log_stderr = Stdio::from(log_file); + + // Start the daemon process + let child = Command::new(&daemon.binary_path) + .stdout(log_stdout) + .stderr(log_stderr) + .spawn() + .map_err(|e| { + FpgadError::Softener((daemon.error_constructor)(format!( + "Failed to spawn daemon: {}", + e + ))) + })?; + + let pid = child.id(); + info!("{} started with PID {}", daemon.name, pid); + info!("Logs will be written to: {}", log_file_path.display()); + + // store the process for later checking + self.processes.insert(daemon.name.clone(), child); + + // Wait for socket to appear + info!("Waiting for socket at {}...", daemon.socket_path.display()); + + for _ in 0..daemon.start_timeout { + // Check if socket exists + if daemon.socket_path.exists() { + match fs::metadata(&daemon.socket_path) { + Ok(metadata) if metadata.file_type().is_socket() => { + info!("{} socket detected - startup successful", daemon.name); + return Ok(()); + } + _ => {} + } + } + + // Check if process is still running + if let Some(process) = self.processes.get_mut(&daemon.name) + && let Ok(Some(status)) = process.try_wait() + { + error!( + "{} process terminated unexpectedly with status: {}", + daemon.name, status + ); + error!("Check logs at: {}", log_file_path.display()); + return Err(FpgadError::Softener((daemon.error_constructor)(format!( + "{} failed to start", + daemon.name + )))); + } + + std::thread::sleep(Duration::from_secs(1)); + } + + // Timeout reached + if !daemon.socket_path.exists() { + // TODO(Artie): what to do here? kill it instead of warn? + warn!( + "Socket didn't appear after {}s, but {} is still running", + daemon.start_timeout, daemon.name + ); + } + + Ok(()) + } + + /// Start all detected daemons. + /// + /// # Returns: `Result<(), String>` + /// * `Ok(())` - All daemons started successfully (or no daemons to start) + /// * `Err(String)` - Failed to start one or more daemons + pub fn check_and_start_daemons(&mut self) -> Result<(), String> { + let available = self.filter_available_daemons(); + + if available.is_empty() { + info!("No daemons to start"); + return Ok(()); + } + + for daemon in &available { + self.start_daemon(daemon).map_err(|e| e.to_string())?; + } + + Ok(()) + } + + /// Attempt to restart a failed daemon multiple times. + /// + /// This method repeatedly attempts to restart a daemon up to `MAX_RESTART_ATTEMPTS` times, + /// waiting 5 seconds between each attempt. If any restart succeeds, the function returns + /// early with success. If all attempts fail, returns an error. + /// + /// # Arguments + /// + /// * `daemon` - Configuration for the daemon to restart + /// + /// # Returns: `Result<(), FpgadError>` + /// * `Ok(())` - Daemon successfully restarted on at least one attempt + /// * `Err(FpgadError::Softener)` - All restart attempts failed + async fn try_restart_daemon(&mut self, daemon: &DaemonConfig) -> Result<(), FpgadError> { + for attempt in 0..Self::MAX_RESTART_ATTEMPTS { + warn!( + "Attempting to restart {} (attempt {}/{})", + daemon.name, + attempt, + Self::MAX_RESTART_ATTEMPTS + ); + + // Wait a bit before restarting + sleep(Self::RESTART_DELAY).await; + + match self.start_daemon(daemon) { + Ok(_) => { + info!("{} successfully restarted", daemon.name); + return Ok(()); + } + Err(e) => { + error!("Failed to restart {}: {}", daemon.name, e); + } + } + } + Err(FpgadError::Softener((daemon.error_constructor)(format!( + "Failed to restart {} after {} attempts", + daemon.name, + Self::MAX_RESTART_ATTEMPTS + )))) + } + + /// Monitor daemon processes and keep them alive. + /// + /// Continuously monitors all managed daemon processes. If a daemon dies unexpectedly, + /// attempts to restart it up to `MAX_RESTART_ATTEMPTS` times. If restart attempts are + /// exhausted, that daemon is abandoned while others continue running. This function + /// runs indefinitely until the task is cancelled. + pub async fn monitor_daemons(&mut self) { + if self.processes.is_empty() { + info!("No daemons to monitor"); + return; + } + + info!("Monitoring {} daemon(s)...", self.processes.len()); + + loop { + let mut daemons_to_restart = Vec::new(); + + // Check all processes and collect names of dead ones + for (name, process) in self.processes.iter_mut() { + match process.try_wait() { + Ok(Some(status)) => { + error!("{} process died unexpectedly with status: {}", name, status); + daemons_to_restart.push(name.clone()); + } + Ok(None) => {} + Err(e) => { + error!("Error checking process status for {}: {}", name, e); + } + } + } + + // Attempt to restart failed daemons - abandoning any that cannot be restarted + for daemon_name in daemons_to_restart { + // Remove the dead process + self.processes.remove(&daemon_name); + + if let Some(daemon) = self.daemons.iter().find(|d| d.name == daemon_name).cloned() { + self.try_restart_daemon(&daemon).await.unwrap_or_else(|e| { + error!("Abandoning daemon: {}", e); + }) + } else { + error!("Could not find daemon config for daemon with name {daemon_name}") + } + } + + sleep(Self::MONITOR_RATE).await; + } + } + + /// Clean up all managed processes. + /// + /// Terminates all running daemon processes and waits for them to exit. + /// This method is automatically called when the DaemonManager is dropped. + pub fn cleanup(&mut self) { + info!("Cleaning up daemon processes..."); + + for (name, mut process) in self.processes.drain() { + match process.try_wait() { + Ok(Some(_)) => { + // Already exited + } + Ok(None) => { + // Still running, terminate it + info!("Terminating {} (PID {})...", name, process.id()); + + if let Err(e) = process.kill() { + error!("Failed to kill {}: {}", name, e); + } else { + // Wait for process to exit + let _ = process.wait(); + } + } + Err(e) => { + error!("Error checking status of {}: {}", name, e); + } + } + } + } +} + +impl Drop for DaemonManager { + fn drop(&mut self) { + self.cleanup(); + } +} + +/// Get managed daemon configurations based on environment. +fn get_managed_daemons() -> Vec { + let prefix = if let Ok(snap_env) = env::var("SNAP_COMPONENTS") { + snap_env + "/dfx-mgr" + } else { + "".to_string() + }; + + vec![ + DaemonConfig::new( + "dfx-mgrd".to_string(), + PathBuf::from(format!("{}/usr/bin/dfx-mgrd", prefix)), + PathBuf::from("/run/dfx-mgrd.socket"), + 10, + FpgadSoftenerError::DfxMgr, + ), // Add more daemon configurations here as needed + // DaemonConfig::new( + // "another-daemon".to_string(), + // PathBuf::from(format!("{}/usr/bin/another-daemon", snap_path)), + // PathBuf::from("/run/another-daemon.socket"), + // ) + ] +} + +/// Run the softener daemon wrapper in the current task. +/// +/// This is the main entry point for the daemon management subsystem. It creates +/// a daemon manager with environment-appropriate configuration, starts all available +/// daemons, and monitors them until shutdown. This function should be called from +/// a spawned tokio task. +/// +/// # Environment Variables +/// +/// * `SNAP` - If set, daemon binaries are expected at `$SNAP/usr/bin/` +/// * `SNAP_COMMON` - If set, log files are written to `$SNAP_COMMON/log/` +/// +/// # Examples +/// +/// ```rust,no_run +/// # use daemon::softeners::softeners_thread::run_softener_daemons; +/// #[tokio::main] +/// async fn main() { +/// if std::env::var("SNAP").is_ok() { +/// tokio::spawn(async { +/// run_softener_daemons().await; +/// }); +/// } +/// } +/// ``` +pub async fn run_softener_daemons() { + info!("Starting softener daemon wrapper..."); + + let daemons = get_managed_daemons(); + let mut manager = DaemonManager::new(daemons); + + // Start all daemons + match manager.check_and_start_daemons() { + Ok(_) => { + // Monitor daemons until termination + manager.monitor_daemons().await; + } + Err(e) => { + error!("Failed to start one or more daemons: {}", e); + } + } + + info!("Daemon wrapper exiting"); +} diff --git a/daemon/src/softeners/xilinx_dfx_mgr.rs b/daemon/src/softeners/xilinx_dfx_mgr.rs index b9bfddd7..9a88f96f 100644 --- a/daemon/src/softeners/xilinx_dfx_mgr.rs +++ b/daemon/src/softeners/xilinx_dfx_mgr.rs @@ -213,9 +213,22 @@ pub fn load_overlay(bitstream_path: &Path, dtbo_path: &Path) -> Result Result { - let snap_env = env::var("SNAP").unwrap_or("".to_string()); + let prefix = if let Ok(snap_env) = env::var("SNAP_COMPONENTS") { + snap_env + "/dfx-mgr" + } else { + "".to_string() + }; + + let dfx_mgr_client_path = format!("{}/usr/bin/dfx-mgr-client", prefix); + + // Check if dfx-mgr-client exists before trying to run it + if !Path::new(&dfx_mgr_client_path).exists() { + return Err(FpgadSoftenerError::DfxMgr(format!( + "dfx-mgr-client not found at '{}'. Install the dfx-mgr component with: snap install fpgad+dfx-mgr.comp", + dfx_mgr_client_path + ))); + } - let dfx_mgr_client_path = format!("{}/usr/bin/dfx-mgr-client", snap_env); trace!("Calling dfx-mgr with args {:#?}", args); let output = std::process::Command::new(&dfx_mgr_client_path) .args(args) diff --git a/scripts/test_wrapper.sh b/scripts/test_wrapper.sh new file mode 100755 index 00000000..495605b0 --- /dev/null +++ b/scripts/test_wrapper.sh @@ -0,0 +1,55 @@ +#!/usr/bin/env bash +# +# This file is part of fpgad, an application to manage FPGA subsystem together with device-tree and kernel modules. +# +# Copyright 2026 Canonical Ltd. +# +# SPDX-License-Identifier: GPL-3.0-only +# +# fpgad is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License version 3, as published by the Free Software Foundation. +# +# fpgad is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranties of MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License along with this program. If not, see http://www.gnu.org/licenses/. +# + +# Check if the tests component is installed +if [ ! -d "$SNAP_COMPONENTS/test" ]; then + echo "Tests component not installed. Install with \"snap install fpgad+test.comp\"" + exit 1 +fi + +# Change to the user's current directory (from outside the snap) +# This allows tests to access files in the working directory +if [ -n "$SNAP_USER_COMMON" ]; then + # If test data is in SNAP_USER_COMMON, use that + if [ -d "$SNAP_USER_COMMON/fpgad" ]; then + cd "$SNAP_USER_COMMON" || return + echo "Test data found at $SNAP_USER_COMMON/fpgad" + echo "Changed working directory to: $SNAP_USER_COMMON" + else + echo "Test data directory not found at $SNAP_USER_COMMON/fpgad" + fi +else + echo "SNAP_USER_COMMON not set" +fi + +# Add snap_testing to PYTHONPATH so imports work +export PYTHONPATH="$SNAP_COMPONENTS/test:$PYTHONPATH" + +# Debug output +echo "SNAP_COMPONENTS: $SNAP_COMPONENTS" +if [ -d "$SNAP_COMPONENTS/dfx-mgr" ]; then + echo "dfx-mgr component found at: $SNAP_COMPONENTS/dfx-mgr" + if [ -f "$SNAP_COMPONENTS/dfx-mgr/usr/bin/dfx-mgr-client" ]; then + echo "dfx-mgr-client found: $SNAP_COMPONENTS/dfx-mgr/usr/bin/dfx-mgr-client" + else + echo "WARNING: dfx-mgr component directory exists but dfx-mgr-client not found" + fi +else + echo "dfx-mgr component NOT installed (xlnx tests will be skipped)" +fi + +# Run the snap tests +exec "$SNAP_COMPONENTS/test/test_snap.sh" "$@" + diff --git a/snap/hooks/component-dfx-mgr-install b/snap/hooks/component-dfx-mgr-install new file mode 100755 index 00000000..388114ee --- /dev/null +++ b/snap/hooks/component-dfx-mgr-install @@ -0,0 +1,21 @@ +#!/bin/sh +set -eu + +# Component install hook for dfx-mgr +# This hook runs when the dfx-mgr component is installed + +CONF_DIR="$SNAP_COMMON/etc/dfx-mgrd" +CONF_FILE="$CONF_DIR/daemon.conf" +DEFAULT_CONF="$SNAP_COMPONENTS/dfx-mgr/daemon.conf" + +mkdir -p "$CONF_DIR" + +# Only copy the default config if the user hasn't created/modified one +if [ ! -f "$CONF_FILE" ]; then + echo "Installing default dfx-mgrd daemon.conf" + cp "$DEFAULT_CONF" "$CONF_FILE" +else + echo "Preserving existing dfx-mgrd daemon.conf" + echo "Default config available at: $DEFAULT_CONF" +fi + diff --git a/snap/hooks/component-dfx-mgr-post-refresh b/snap/hooks/component-dfx-mgr-post-refresh new file mode 100755 index 00000000..42de698c --- /dev/null +++ b/snap/hooks/component-dfx-mgr-post-refresh @@ -0,0 +1,21 @@ +#!/bin/sh +set -eu + +# Component post-refresh hook for dfx-mgr +# This hook runs when the dfx-mgr component is updated + +CONF_DIR="$SNAP_COMMON/etc/dfx-mgrd" +CONF_FILE="$CONF_DIR/daemon.conf" +DEFAULT_CONF="$SNAP_COMPONENTS/dfx-mgr/daemon.conf" + +mkdir -p "$CONF_DIR" + +# Only restore the default config if the user's config is missing +if [ ! -f "$CONF_FILE" ]; then + echo "Restoring default dfx-mgrd daemon.conf" + cp "$DEFAULT_CONF" "$CONF_FILE" +else + echo "Preserving existing dfx-mgrd daemon.conf" + echo "Updated default config available at: $DEFAULT_CONF" +fi + diff --git a/snap/hooks/component-dfx-mgr-remove b/snap/hooks/component-dfx-mgr-remove new file mode 100755 index 00000000..f4d28793 --- /dev/null +++ b/snap/hooks/component-dfx-mgr-remove @@ -0,0 +1,20 @@ +#!/bin/sh +set -eu + +# Component remove hook for dfx-mgr +# This hook runs when the dfx-mgr component is removed + +CONF_DIR="$SNAP_COMMON/etc/dfx-mgrd" +CONF_FILE="$CONF_DIR/daemon.conf" + +if [ -f "$CONF_FILE" ]; then + echo "Removing dfx-mgrd daemon.conf" + rm -f "$CONF_FILE" +fi + +# Remove directory if empty +if [ -d "$CONF_DIR" ] && [ -z "$(ls -A "$CONF_DIR")" ]; then + echo "Removing empty dfx-mgrd config directory" + rmdir "$CONF_DIR" +fi + diff --git a/snap/hooks/install b/snap/hooks/install deleted file mode 100644 index 00f1a1e8..00000000 --- a/snap/hooks/install +++ /dev/null @@ -1,13 +0,0 @@ -#!/bin/sh -set -eu - -CONF_DIR="$SNAP_COMMON/etc/dfx-mgrd" -CONF_FILE="$CONF_DIR/daemon.conf" -DEFAULT_CONF="$SNAP/etc/dfx-mgrd/daemon.conf" - -mkdir -p "$CONF_DIR" - -if [ ! -f "$CONF_FILE" ]; then - echo "Initializing default daemon.conf" - cp "$DEFAULT_CONF" "$CONF_FILE" -fi \ No newline at end of file diff --git a/snap/hooks/post-refresh b/snap/hooks/post-refresh deleted file mode 100644 index 8e92d03a..00000000 --- a/snap/hooks/post-refresh +++ /dev/null @@ -1,13 +0,0 @@ -#!/bin/sh -set -eu - -CONF_DIR="$SNAP_COMMON/etc/dfx-mgrd" -CONF_FILE="$CONF_DIR/daemon.conf" -DEFAULT_CONF="$SNAP/etc/dfx-mgrd/daemon.conf" - -mkdir -p "$CONF_DIR" - -if [ ! -f "$CONF_FILE" ]; then - echo "Restoring default daemon.conf" - cp "$DEFAULT_CONF" "$CONF_FILE" -fi \ No newline at end of file diff --git a/snap/snapcraft.yaml b/snap/snapcraft.yaml index 884f9517..54d8803d 100644 --- a/snap/snapcraft.yaml +++ b/snap/snapcraft.yaml @@ -29,6 +29,24 @@ platforms: build-on: [riscv64] build-for: [riscv64] icon: snap/assets/icon.svg +environment: + # Set path to snap components + SNAP_COMPONENTS: /snap/$SNAP_INSTANCE_NAME/components/$SNAP_REVISION + # Add dfx-mgr libraries to library path (if component is installed) + LD_LIBRARY_PATH: $SNAP_COMPONENTS/dfx-mgr/usr/lib:$SNAP_COMPONENTS/dfx-mgr/usr/lib/$CRAFT_ARCH_TRIPLET_BUILD_FOR:$LD_LIBRARY_PATH +components: + dfx-mgr: + type: standard + summary: optional component to include dfx-mgr binaries + description: | + This component enables dfx-mgr integration for xilinx boards. + It provides dfx-mgr-client and dfx-mgrd. + FPGAd will run and monitor the dfx-mgrd binary and interface with it to enabled full dfx-mgr functionality \ + via FPGAd + test: + type: standard + summary: python integration and end-to-end tests to test cli + description: python integration and end-to-end tests to test cli slots: daemon-dbus: interface: dbus @@ -71,20 +89,17 @@ apps: - kernel-firmware-control - hardware-observe - dfx-mgr-socket + - device-tree-overlays activates-on: - daemon-dbus environment: RUST_LOG: trace - softeners: - command: bin/softener_wrapper - daemon: simple - restart-condition: on-failure + test: + command: bin/test_wrapper.sh plugs: + - cli-dbus - fpga - - kernel-firmware-control - - hardware-observe - - dfx-mgr-socket - - network-bind + - device-tree-overlays parts: version: plugin: nil @@ -117,22 +132,28 @@ parts: - dfx-mgr override-stage: | craftctl default - cat << 'EOF' > $SNAPCRAFT_PART_INSTALL/etc/dfx-mgrd/daemon.conf + cat << 'EOF' > daemon.conf { "firmware_location": [ "/var/snap/fpgad/current/provider-content/" ] } EOF - softener_wrapper: + organize: + # move everything, including the staged packages + "*": (component/dfx-mgr) + daemon.conf: (component/dfx-mgr) + test: + plugin: dump + source: tests + organize: + "snap_testing": (component/test) + test-wrapper: plugin: dump - source: snap/assets + source: scripts organize: - softener_wrapper: bin/softener_wrapper - stage: - - bin/softener_wrapper -# TODO: the daemon.conf should be copied into $SNAP_COMMON on install hook if - and only if - it doesn't already exist + "test_wrapper.sh": bin/test_wrapper.sh layout: - # For configuration files + # For dfx-mgr configuration files (when component is installed) /etc/dfx-mgrd: bind: $SNAP_COMMON/etc/dfx-mgrd diff --git a/tests/concurent_load/test_concurrent_load.sh b/tests/concurent_load/test_concurrent_load.sh index 6006bc5e..acde7dff 100755 --- a/tests/concurent_load/test_concurrent_load.sh +++ b/tests/concurent_load/test_concurrent_load.sh @@ -21,7 +21,7 @@ cmd2='sudo busctl call --system com.canonical.fpgad /com/canonical/fpgad/control ' # Spawn -for i in $(seq 1 $NUM_WORKERS); do +for _ in $(seq 1 "$NUM_WORKERS"); do eval "$cmd1" & eval "$cmd2" & done diff --git a/tests/snap_testing/QUICK_REFERENCE.sh b/tests/snap_testing/QUICK_REFERENCE.sh new file mode 100755 index 00000000..56ee24cd --- /dev/null +++ b/tests/snap_testing/QUICK_REFERENCE.sh @@ -0,0 +1,55 @@ +#!/bin/bash +# +# This file is part of fpgad, an application to manage FPGA subsystem together with device-tree and kernel modules. +# +# Copyright 2026 Canonical Ltd. +# +# SPDX-License-Identifier: GPL-3.0-only +# +# fpgad is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License version 3, as published by the Free Software Foundation. +# +# fpgad is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranties of MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License along with this program. If not, see http://www.gnu.org/licenses/. +# + +# Quick reference for running FPGA tests + +cat << 'EOF' +FPGA Test Runner - Quick Reference +==================================== + +Run All Tests: + cd tests && ./run_all.sh + python3 -m unittest discover -s tests -p "test_*.py" -v + +Run Platform-Specific Tests: + ./run_all.sh universal # Universal platform tests only + ./run_all.sh xlnx # Xilinx platform tests only + ./run_all.sh default # Help & CLI option tests + +Run Individual Test Files: + python3 -m unittest tests.test_universal.test_bitstream -v + python3 -m unittest tests.test_xlnx.test_overlay -v + python3 -m unittest tests.test_default.test_help -v + +Run Single Test: + python3 -m unittest tests.test_universal.test_bitstream.TestBitstreamUniversal.test_load_bitstream_local -v + +Directory Structure: + tests/ + ├── common/ # Shared base classes & utilities + ├── test_universal/ # Tests with --platform universal + ├── test_xlnx/ # Tests with --platform xlnx + └── test_default/ # Tests without platform flag + +Legacy (Deprecated): + python3 snap_tests.py # Old monolithic test file + +Requirements: + - Must run as root (tests interact with sysfs) + - fpgad snap must be installed + - FPGA hardware required (Xilinx Kria or compatible) + +EOF + diff --git a/tests/snap_testing/README.md b/tests/snap_testing/README.md new file mode 100644 index 00000000..4959076c --- /dev/null +++ b/tests/snap_testing/README.md @@ -0,0 +1,184 @@ +# FPGA Snap Tests + +This directory contains integration tests for the fpgad snap package. + +## Directory Structure + +``` +tests/ +├── common/ # Shared utilities and base classes +│ ├── base_test.py # Base test class with setUp/tearDown +│ └── helpers.py # Colors, TestData, utility functions +├── test_universal/ # Tests with --platform universal +│ ├── test_bitstream.py # Bitstream loading tests +│ ├── test_overlay.py # Overlay loading tests +│ ├── test_status.py # Status command tests +│ └── test_set.py # Set command tests +├── test_xlnx/ # Tests with --platform xlnx +│ ├── test_bitstream.py # Bitstream loading tests +│ ├── test_overlay.py # Overlay loading tests +│ ├── test_status.py # Status command tests +│ └── test_set.py # Set command tests +├── test_default/ # Tests without platform flag +│ ├── test_help.py # Help command tests +│ └── test_cli_options.py # CLI option tests (--device, --name, etc.) +├── run_all.sh # Convenience script to run tests +└── snap_tests.py # Legacy monolithic test file (deprecated) +``` + +## Running Tests + +### Running Snap Tests (Production) + +When testing the installed snap package, use the `fpgad.test` command: + +```bash +# Install the snap with test component +sudo snap install fpgad+test --dangerous ./fpgad_*.snap + +# Connect required interfaces (for daemon and tests) +sudo snap connect fpgad:fpga +sudo snap connect fpgad:kernel-firmware-control +sudo snap connect fpgad:hardware-observe +sudo snap connect fpgad:device-tree-overlays +sudo snap connect fpgad:dfx-mgr-socket +sudo snap connect fpgad:cli-dbus fpgad:daemon-dbus + +# Run all tests (MUST use sudo) +sudo fpgad.test + +# Run specific test suite +sudo fpgad.test universal # Universal platform tests only +sudo fpgad.test xlnx # Xilinx dfx-mgr tests only +sudo fpgad.test default # Default (no platform) tests only +``` + +**Important Notes:** +- Tests **MUST** be run with `sudo` as they need to write to sysfs +- All snap interfaces must be connected before running tests +- The test component must be installed (`fpgad+test`) +- **Ubuntu Core Limitation**: Tests will NOT work on Ubuntu Core. The `system-files` interface required for sysfs access (`/sys/class/fpga_manager/*` and `/sys/kernel/config/device-tree/overlays`) is not auto-connected on Ubuntu Core and cannot be manually connected on a strictly confined system. Tests must be run on Ubuntu Server or Desktop. + +### Running Tests During Development + +For development/debugging without building the snap: + +```bash +# From tests directory +python3 -m unittest discover -s . -p "test_*.py" -v + +# Or use the convenience script +./run_all.sh +``` + +### Run Platform-Specific Tests + +```bash +# Universal platform only +python3 -m unittest discover -s test_universal -p "test_*.py" -v +./run_all.sh universal + +# Xilinx platform only +python3 -m unittest discover -s test_xlnx -p "test_*.py" -v +./run_all.sh xlnx + +# Default (no platform flag) tests only +python3 -m unittest discover -s test_default -p "test_*.py" -v +./run_all.sh default +``` + +### Run Specific Test File + +```bash +python3 -m unittest tests.test_universal.test_bitstream +python3 -m unittest tests.test_xlnx.test_overlay +python3 -m unittest tests.test_default.test_help +``` + +### Run Single Test + +```bash +python3 -m unittest tests.test_universal.test_bitstream.TestBitstreamUniversal.test_load_bitstream_local +``` + +## Requirements + +### For Snap Tests +- **Root/sudo access** (tests interact with sysfs and FPGA hardware) +- FPGA hardware (Xilinx Kria or compatible) +- fpgad snap installed with test component +- **All snap interfaces connected** (provides necessary permissions) +- Test data files in snap provider content location +- **Ubuntu Server or Desktop** (tests will not work on Ubuntu Core due to system-files interface restrictions) + +### For Development Tests (without snap) +- **Root/sudo access** (direct sysfs interaction requires privileges) +- FPGA hardware (Xilinx Kria or compatible) +- fpgad daemon running +- Test data files in `daemon/tests/test_data/` +- Test data files copied to `fpgad/k26-starter-kits/` and `fpgad/k24-starter-kits/` (relative to test working directory) + +## Test Structure + +All test classes inherit from `FPGATestBase` which provides: +- Automatic cleanup of overlays before/after tests +- Reset of FPGA flags +- Common assertion helpers for process results +- Utility methods for FPGA operations + +### Test Data Setup + +The tests require bitstream (`.bit.bin`) and device tree overlay (`.dtbo`) files to be present in specific locations: + +**Source files** (repository): +- `daemon/tests/test_data/k26-starter-kits/` - Contains test files for Kria K26 +- `daemon/tests/test_data/k24-starter-kits/` - Contains test files for Kria KD240 + +**Expected locations** (at test runtime): +- `fpgad/k26-starter-kits/` - Relative to the test working directory +- `fpgad/k24-starter-kits/` - Relative to the test working directory + +The `local_snap_tests.sh` script automatically copies files from the source locations to the expected locations during test setup. If running tests manually, you must ensure these files are copied to the correct locations. + +## Migration from Legacy Tests + +The old `snap_tests.py` file is deprecated but kept for reference. All tests have been: +1. Split by platform (universal, xlnx, default) +2. Organized by functionality (bitstream, overlay, status, set, help, cli_options) +3. Refactored to use shared base classes and utilities + +To run the legacy tests: +```bash +python3 snap_tests.py +``` + +## CI/CD Integration + +The GitHub Actions workflow automatically: +1. Builds the snap +2. Deploys to test hardware via Testflinger +3. Runs all tests using unittest discovery +4. Collects and uploads test artifacts + +## Adding New Tests + +1. Determine which platform directory (test_universal, test_xlnx, or test_default) +2. Add test methods to existing test classes or create new test file +3. Inherit from `FPGATestBase` +4. Use provided assertion helpers +5. Follow existing naming conventions + +Example: +```python +from common.base_test import FPGATestBase + +class TestNewFeature(FPGATestBase): + PLATFORM = "universal" # or "xlnx", or omit for default + + def test_something(self): + proc = self.run_fpgad(["--platform", self.PLATFORM, "command"]) + self.assert_proc_succeeds(proc) +``` + +**Note:** Import paths use `from common.*` when tests are in the snap_testing directory. + diff --git a/tests/snap_testing/__init__.py b/tests/snap_testing/__init__.py new file mode 100644 index 00000000..c5962f99 --- /dev/null +++ b/tests/snap_testing/__init__.py @@ -0,0 +1,3 @@ +""" +FPGAd snap tests for CLI interactions +""" diff --git a/tests/snap_testing/common/__init__.py b/tests/snap_testing/common/__init__.py new file mode 100644 index 00000000..1ed54fdf --- /dev/null +++ b/tests/snap_testing/common/__init__.py @@ -0,0 +1 @@ +"""Common utilities for FPGA snap tests.""" diff --git a/tests/snap_testing/common/base_test.py b/tests/snap_testing/common/base_test.py new file mode 100644 index 00000000..38c423b2 --- /dev/null +++ b/tests/snap_testing/common/base_test.py @@ -0,0 +1,145 @@ +#!/usr/bin/env python3 +""" +Base test class for FPGA snap tests. +""" + +import os +import subprocess +import unittest +from pathlib import Path +from subprocess import CompletedProcess +from typing import List + +from .helpers import Colors, cleanup_applied_overlays, set_flags + + +class FPGATestBase(unittest.TestCase): + """Base test class with common setup, teardown, and assertion methods.""" + + def setUp(self): + """Run before each test.""" + self.cleanup_bitstreams() + cleanup_applied_overlays() + self.reset_flags() + + def tearDown(self): + """Run after each test.""" + self.cleanup_bitstreams() + cleanup_applied_overlays() + + @classmethod + def tearDownClass(cls): + """Run once after all tests in this class are finished.""" + cleanup_applied_overlays() + set_flags(0) + + def cleanup_bitstreams(self): + """Remove any loaded bitstreams to ensure clean test state.""" + # Try to remove bitstream - ignore errors if nothing to remove + try: + proc = self.run_fpgad(["remove", "bitstream"]) + # Don't assert success - it's OK if there's nothing to remove + if proc.returncode == 0: + print(f"{Colors.CYAN}[INFO]{Colors.RESET} Cleaned up loaded bitstream") + except Exception as e: + # Silently ignore - we're just trying to clean up + print( + f"{Colors.YELLOW}[WARN]{Colors.RESET} Could not cleanup bitstream: {e}" + ) + + # ============================================================ + # ===================== HELPER FUNCTIONS ===================== + # ============================================================ + + def assert_proc_succeeds(self, proc, msg=None): + """Assert that a process completed successfully, including stdout/stderr on failure.""" + if msg is None: + msg = f"Return code is {proc.returncode} when expecting 0" + full_msg = ( + f"{msg}\n" + f"Status code:\t{proc.returncode}\n" + f"stdout:\t{proc.stdout}\n" + f"stderr:\t{proc.stderr}" + ) + self.assertEqual(proc.returncode, 0, full_msg) + + def assert_proc_fails(self, proc, msg=None): + """Assert that a process failed, including stdout/stderr on failure.""" + if msg is None: + msg = f"Return code is {proc.returncode} when expecting nonzero" + full_msg = ( + f"{msg}\n" + f"Status code:\t{proc.returncode}\n" + f"stdout:\t{proc.stdout}\n" + f"stderr:\t{proc.stderr}" + ) + self.assertNotEqual(proc.returncode, 0, full_msg) + + def assert_in_proc_out( + self, substring: str, proc: CompletedProcess, msg: str = None + ): + """Assert that a substring exists in stdout, including stdout/stderr on failure.""" + if msg is None: + msg = f"'{substring}' not found in output." + full_msg = f"{msg}\nstdout:\t{proc.stdout}\nstderr:\t{proc.stderr}" + self.assertIn(substring, proc.stdout, full_msg) + + def assert_not_in_proc_out( + self, substring: str, proc: CompletedProcess, msg: str = None + ): + """Assert that a substring does not exist in stdout, including stdout/stderr on failure.""" + if msg is None: + msg = f"Undesired '{substring}' found in output." + full_msg = f"{msg}\nstdout:\t{proc.stdout}\nstderr:\t{proc.stderr}" + self.assertNotIn(substring, proc.stdout, full_msg) + + def assert_in_proc_err( + self, substring: str, proc: CompletedProcess, msg: str = None + ): + """Assert that a substring exists in stderr, including stdout/stderr on failure.""" + if msg is None: + msg = f"'{substring}' not found in stderr." + full_msg = f"{msg}\nstdout:\t{proc.stdout}\nstderr:\t{proc.stderr}" + self.assertIn(substring, proc.stderr, full_msg) + + @staticmethod + def get_fpga0_attribute(attr: str): + """Get an FPGA manager attribute value.""" + path = Path(f"/sys/class/fpga_manager/fpga0/{attr}") + with open(path, "r") as f: + real_attr = f.read().strip() + return real_attr + + def check_fpga0_attribute(self, attr: str, expected: str): + """Assert that an FPGA manager attribute contains the expected value.""" + path = Path(f"/sys/class/fpga_manager/fpga0/{attr}") + with open(path, "r") as f: + real_attr = f.read().strip() + self.assertIn(expected, real_attr) + + def reset_flags(self): + """Reset flags (to zero) using system calls, instead of fpgad.""" + print(f"{Colors.CYAN}[INFO]{Colors.RESET} Resetting fpga0's flags to 0") + set_flags(0) + + def run_fpgad(self, args: List[str]) -> subprocess.CompletedProcess[str]: + """ + Run the fpgad cli with provided args as a subprocess. + + :param args: list of arguments to provide to the fpgad cli call + :return: the completed process object, containing return code and captured output + """ + # When running in snap, call the CLI binary directly + # The alias 'fpgad' points to the CLI app, but inside snap we need the actual binary + if os.getenv("SNAP"): + cmd = [f"{os.getenv('SNAP')}/bin/fpgad_cli"] + args + else: + cmd = ["fpgad"] + args + + print(f"{Colors.CYAN}[INFO]{Colors.RESET} Running: {' '.join(cmd)}") + + proc = subprocess.run( + cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True + ) + + return proc diff --git a/tests/snap_testing/common/helpers.py b/tests/snap_testing/common/helpers.py new file mode 100644 index 00000000..50668f48 --- /dev/null +++ b/tests/snap_testing/common/helpers.py @@ -0,0 +1,164 @@ +#!/usr/bin/env python3 +""" +Helper utilities and data structures for FPGA snap tests. +""" + +import os +import shutil +from pathlib import Path + + +BAD_FLAGS = 223 + + +def is_dfx_mgr_available() -> bool: + """ + Check if the dfx-mgr component is installed and available. + + :return: True if dfx-mgr-client is available, False otherwise + """ + # Check if running in snap environment + snap_components = os.getenv("SNAP_COMPONENTS") + if snap_components: + dfx_mgr_client_path = f"{snap_components}/dfx-mgr/usr/bin/dfx-mgr-client" + exists = os.path.exists(dfx_mgr_client_path) + if not exists: + print(f"[DEBUG] dfx-mgr-client not found at: {dfx_mgr_client_path}") + print(f"[DEBUG] SNAP_COMPONENTS={snap_components}") + return exists + else: + # Not in snap environment, check system path + import shutil + + result = shutil.which("dfx-mgr-client") is not None + if not result: + print("[DEBUG] SNAP_COMPONENTS not set, dfx-mgr-client not in PATH") + return result + + +class Colors: + """ANSI color codes for terminal output.""" + + GREEN = "\033[92m" + RED = "\033[91m" + YELLOW = "\033[93m" + CYAN = "\033[96m" + RESET = "\033[0m" + + +class TestData: + """ + Container for copying files during tests. + + Defines the source and target locations for use with + copy_test_data_files and cleanup_test_data_files. + """ + + def __init__(self, source: Path, target: Path): + """ + Initialize test data paths. + + :param source: the path to the file which should be copied + :param target: the path to which the file should be copied/was copied to + """ + self.source = source + self.target = target + + +def copy_test_data_files(test_file: TestData) -> int: + """ + Copy a file from test_file.source to test_file.target. + + Use in conjunction with cleanup_test_data_files to test loading from a custom location. + + :param test_file: A TestData object which contains the relevant paths + :return: 0 on success, -1 on failure + """ + src = Path(test_file.source) + + if not src.exists(): + print(f"{Colors.YELLOW}[WARN]{Colors.RESET} Source file missing: {src}") + return -1 + + target = test_file.target + target_path = Path(target) + + # Ensure directory exists + target_path.parent.mkdir(parents=True, exist_ok=True) + + print(f"{Colors.CYAN}[INFO]{Colors.RESET} Copying {src} → {target_path}") + shutil.copy2(src, target_path) + return 0 + + +def cleanup_test_data_files(test_file: TestData) -> int: + """ + Clean up the file located at test_file.target. + + Use after copy_test_data_files. + + :param test_file: A TestData object which contains the location to which the file was originally copied + :return: 0 on success, -1 on failure + """ + target = test_file.target + path = Path(target) + print(f"{Colors.CYAN}[INFO]{Colors.RESET} deleting {test_file.target}") + if not path.exists(): + print( + f"{Colors.YELLOW}[WARN]{Colors.RESET} Missing file during cleanup: {path}" + ) + return -1 + try: + path.unlink() + except Exception as e: + print(f"{Colors.RED}[ERROR]{Colors.RESET} Failed to remove {path}: {e}") + return -1 + + return 0 + + +def cleanup_applied_overlays(): + """Remove all applied device tree overlays.""" + directory = "/sys/kernel/config/device-tree/overlays/" + print(f"{Colors.CYAN}[INFO]{Colors.RESET} Cleaning up applied overlays") + + for item in os.listdir(directory): + item_path = os.path.join(directory, item) + if os.path.isdir(item_path): + try: + os.rmdir(item_path) # Remove the directory itself + print( + f"{Colors.CYAN}[INFO]{Colors.RESET} Removed overlay directory at {item_path}" + ) + except PermissionError: + print( + f"{Colors.RED}[ERROR]{Colors.RESET} Permission denied removing {item_path}. Run as root." + ) + except OSError as e: + # This happens if the directory is not empty + print( + f"{Colors.RED}[ERROR]{Colors.RESET} Failed to remove {item_path}: {e}" + ) + + +def set_flags(flags: int = 0) -> None: + """ + Set FPGA manager flags directly via sysfs. + + :param flags: Flag value to write (default: 0) + """ + flags_path = r"/sys/class/fpga_manager/fpga0/flags" + try: + with open(flags_path, "w") as f: + f.write(f"{flags:X}") + print( + f"{Colors.CYAN}[INFO]{Colors.RESET} Successfully wrote {flags} to {flags_path}" + ) + except PermissionError: + print( + f"{Colors.RED}[ERROR]{Colors.RESET} Permission denied: you probably need to run as root" + ) + except FileNotFoundError: + print(f"{Colors.RED}[ERROR]{Colors.RESET} {flags_path} does not exist") + except Exception as e: + print(f"Error writing to {flags_path}: {e}") diff --git a/tests/snap_testing/test_default/__init__.py b/tests/snap_testing/test_default/__init__.py new file mode 100644 index 00000000..c1f18df9 --- /dev/null +++ b/tests/snap_testing/test_default/__init__.py @@ -0,0 +1 @@ +"""Tests for default behavior (without platform specification).""" diff --git a/tests/snap_testing/test_default/test_cli_options.py b/tests/snap_testing/test_default/test_cli_options.py new file mode 100644 index 00000000..0bf874c0 --- /dev/null +++ b/tests/snap_testing/test_default/test_cli_options.py @@ -0,0 +1,36 @@ +#!/usr/bin/env python3 +"""CLI option tests (--device, --platform, --name).""" + +from common.base_test import FPGATestBase + + +class TestCLIOptions(FPGATestBase): + """Test CLI options like --device, --platform, and --name.""" + + def test_load_bitstream_with_device_option(self): + """Test loading bitstream with explicit --device option.""" + path_str = "./fpgad/k26-starter-kits/k26_starter_kits.bit.bin" + proc = self.run_fpgad(["--device", "fpga0", "load", "bitstream", path_str]) + self.assert_proc_succeeds(proc) + # Accept either universal format or xlnx/dfx-mgr format + # Universal: "... loaded to 'fpga0' using firmware lookup path ..." + # xlnx: "Loaded with slot_handle 0" + success_indicators = ["loaded to", "Loaded with slot_handle"] + found = any(indicator in proc.stdout for indicator in success_indicators) + self.assertTrue( + found, f"Expected success indicator not found in output: {proc.stdout}" + ) + + def test_status_with_device_option(self): + """Test status command with explicit --device option.""" + proc = self.run_fpgad(["--device", "fpga0", "status"]) + self.assert_proc_succeeds(proc) + + def test_set_with_device_option(self): + """Test set command with explicit --device option.""" + proc = self.run_fpgad(["--device", "fpga0", "set", "flags", "0"]) + self.assert_proc_succeeds(proc) + self.assert_in_proc_out( + "0 written to /sys/class/fpga_manager/fpga0/flags", proc + ) + self.check_fpga0_attribute("flags", "0") diff --git a/tests/snap_testing/test_default/test_help.py b/tests/snap_testing/test_default/test_help.py new file mode 100644 index 00000000..e7fae9ee --- /dev/null +++ b/tests/snap_testing/test_default/test_help.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python3 +"""Help command tests (platform-independent).""" + +from common.base_test import FPGATestBase + + +class TestHelp(FPGATestBase): + """Test help command functionality.""" + + def test_help_main(self): + """Test main help command.""" + proc = self.run_fpgad(["help"]) + self.assert_proc_succeeds(proc) + + def test_help_main_as_flag(self): + """Test help as --help flag.""" + proc = self.run_fpgad(["--help"]) + self.assert_proc_succeeds(proc) + + def test_help_set(self): + """Test help for set command.""" + proc = self.run_fpgad(["help", "set"]) + self.assert_proc_succeeds(proc) + + def test_help_remove(self): + """Test help for remove command.""" + proc = self.run_fpgad(["help", "remove"]) + self.assert_proc_succeeds(proc) + + def test_help_remove_overlay(self): + """Test help for remove overlay subcommand.""" + proc = self.run_fpgad(["help", "remove", "overlay"]) + self.assert_proc_succeeds(proc) + + def test_help_remove_bitstream(self): + """Test help for remove bitstream subcommand.""" + proc = self.run_fpgad(["help", "remove", "bitstream"]) + self.assert_proc_succeeds(proc) + + def test_help_load(self): + """Test help for load command.""" + proc = self.run_fpgad(["help", "load"]) + self.assert_proc_succeeds(proc) + + def test_help_load_bitstream(self): + """Test help for load bitstream subcommand.""" + proc = self.run_fpgad(["help", "load", "bitstream"]) + self.assert_proc_succeeds(proc) + + def test_help_load_overlay(self): + """Test help for load overlay subcommand.""" + proc = self.run_fpgad(["help", "load", "overlay"]) + self.assert_proc_succeeds(proc) diff --git a/tests/snap_testing/test_snap.sh b/tests/snap_testing/test_snap.sh new file mode 100755 index 00000000..28fb2a0a --- /dev/null +++ b/tests/snap_testing/test_snap.sh @@ -0,0 +1,52 @@ +#!/usr/bin/env bash +# +# This file is part of fpgad, an application to manage FPGA subsystem together with device-tree and kernel modules. +# +# Copyright 2026 Canonical Ltd. +# +# SPDX-License-Identifier: GPL-3.0-only +# +# fpgad is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License version 3, as published by the Free Software Foundation. +# +# fpgad is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranties of MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License along with this program. If not, see http://www.gnu.org/licenses/. +# + +# Convenience script to run all FPGA snap tests + +set -e + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +echo "==========================================" +echo "Running FPGA Snap Tests" +echo "Working directory: $(pwd)" +echo "==========================================" + +# Check if specific test suite is requested +if [ $# -eq 0 ]; then + echo "Running ALL tests..." + python3 -m unittest discover -s "$SCRIPT_DIR" -p "test_*.py" -v +elif [ "$1" == "universal" ]; then + echo "Running UNIVERSAL platform tests..." + python3 -m unittest discover -s "$SCRIPT_DIR/test_universal" -p "test_*.py" -v +elif [ "$1" == "xlnx" ]; then + echo "Running xilinx dfx-mgr platform tests..." + python3 -m unittest discover -s "$SCRIPT_DIR/test_xlnx" -p "test_*.py" -v +elif [ "$1" == "default" ]; then + echo "Running DEFAULT (no platform) tests..." + python3 -m unittest discover -s "$SCRIPT_DIR/test_default" -p "test_*.py" -v +else + echo "Usage: $0 [universal|xlnx|default]" + echo " No argument: run all tests" + echo " universal: run only --platform universal tests" + echo " xlnx: run only --platform xlnx tests" + echo " default: run only tests without platform flag" + exit 1 +fi + +echo "==========================================" +echo "Tests completed successfully!" +echo "==========================================" + diff --git a/tests/snap_testing/test_universal/__init__.py b/tests/snap_testing/test_universal/__init__.py new file mode 100644 index 00000000..8403efd9 --- /dev/null +++ b/tests/snap_testing/test_universal/__init__.py @@ -0,0 +1 @@ +"""Universal platform tests.""" diff --git a/tests/snap_testing/test_universal/test_bitstream.py b/tests/snap_testing/test_universal/test_bitstream.py new file mode 100644 index 00000000..de6612ba --- /dev/null +++ b/tests/snap_testing/test_universal/test_bitstream.py @@ -0,0 +1,73 @@ +#!/usr/bin/env python3 +"""Bitstream loading tests for universal platform.""" + +import os +from pathlib import Path +from common.base_test import FPGATestBase +from common.helpers import ( + BAD_FLAGS, + set_flags, +) + + +class TestBitstreamUniversal(FPGATestBase): + """Test bitstream loading operations with --platform universal.""" + + PLATFORM = "universal" + + def test_load_bitstream_local(self): + """Test loading bitstream from local relative path.""" + path_str = "./fpgad/k26-starter-kits/k26_starter_kits.bit.bin" + proc = self.run_fpgad( + ["--platform", self.PLATFORM, "load", "bitstream", path_str] + ) + self.assert_proc_succeeds(proc) + # Check for key parts of success message (output includes quotes) + self.assertIn("loaded to", proc.stdout) + self.assertIn("fpga0", proc.stdout) + + def test_load_bitstream_home_fullpath(self): + """Test loading bitstream from full absolute path.""" + prefix = Path(os.getcwd()) + path = prefix.joinpath("fpgad/k26-starter-kits/k26_starter_kits.bit.bin") + proc = self.run_fpgad( + ["--platform", self.PLATFORM, "load", "bitstream", str(path)] + ) + self.assert_proc_succeeds(proc) + self.assert_in_proc_out("loaded to", proc) + self.assert_in_proc_out("fpga0", proc) + + def test_load_bitstream_path_not_exist(self): + """Test loading bitstream from non-existent path fails.""" + proc = self.run_fpgad( + [ + "--platform", + self.PLATFORM, + "load", + "bitstream", + "/this/path/is/fake.bit.bin", + ] + ) + self.assert_proc_fails(proc) + self.assert_in_proc_err("FpgadError::IOWrite:", proc) + + def test_load_bitstream_containing_dir(self): + """Test loading bitstream with directory path fails.""" + prefix = Path(os.getcwd()) + path = prefix.joinpath("fpgad/k26-starter-kits/") + proc = self.run_fpgad( + ["--platform", self.PLATFORM, "load", "bitstream", str(path)] + ) + self.assert_proc_fails(proc) + self.assert_in_proc_err("FpgadError::IOWrite:", proc) + + def test_load_bitstream_bad_flags(self): + """Test loading bitstream with invalid flags fails.""" + prefix = Path(os.getcwd()) + path = prefix.joinpath("fpgad/k26-starter-kits/k26_starter_kits.bit.bin") + set_flags(BAD_FLAGS) + proc = self.run_fpgad( + ["--platform", self.PLATFORM, "load", "bitstream", str(path)] + ) + self.assert_proc_fails(proc) + self.assert_in_proc_err("FpgadError::IOWrite:", proc) diff --git a/tests/snap_testing/test_universal/test_overlay.py b/tests/snap_testing/test_universal/test_overlay.py new file mode 100644 index 00000000..a8bb025f --- /dev/null +++ b/tests/snap_testing/test_universal/test_overlay.py @@ -0,0 +1,34 @@ +#!/usr/bin/env python3 +"""Overlay loading tests for universal platform.""" + +from pathlib import Path + +from common.base_test import FPGATestBase + + +class TestOverlayUniversal(FPGATestBase): + """Test overlay loading operations with --platform universal.""" + + PLATFORM = "universal" + + def test_load_overlay_bad_path(self): + """Test loading overlay from non-existent path fails.""" + overlay_path = Path("/path/does/not/exist") + proc = self.run_fpgad( + ["--platform", self.PLATFORM, "load", "overlay", str(overlay_path)] + ) + self.assert_proc_fails(proc) + self.assert_not_in_proc_out("loaded via", proc) + # Actual error is IOCreate permission denied + self.assert_in_proc_err("FpgadError::", proc) + + def test_load_overlay_missing_bitstream(self): + """Test loading overlay without corresponding bitstream fails.""" + overlay_path = Path("./fpgad/k26-starter-kits/k26_starter_kits.dtbo") + proc = self.run_fpgad( + ["--platform", self.PLATFORM, "load", "overlay", str(overlay_path)] + ) + self.assert_proc_fails(proc) + self.assert_not_in_proc_out("loaded via", proc) + # Actual error is IOCreate permission denied + self.assert_in_proc_err("FpgadError::", proc) diff --git a/tests/snap_testing/test_universal/test_set.py b/tests/snap_testing/test_universal/test_set.py new file mode 100644 index 00000000..73516e50 --- /dev/null +++ b/tests/snap_testing/test_universal/test_set.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python3 +"""Set command tests for universal platform.""" + +from common.base_test import FPGATestBase + + +class TestSetUniversal(FPGATestBase): + """Test set command with --platform universal.""" + + PLATFORM = "universal" + + def test_set_flags_nonzero(self): + """Test setting flags to non-zero value.""" + proc = self.run_fpgad(["--platform", self.PLATFORM, "set", "flags", "20"]) + self.assert_proc_succeeds(proc) + self.assert_in_proc_out( + "20 written to /sys/class/fpga_manager/fpga0/flags", proc + ) + self.check_fpga0_attribute("flags", "20") + + def test_set_flags_string(self): + """Test setting flags with invalid string value fails.""" + proc = self.run_fpgad(["--platform", self.PLATFORM, "set", "flags", "zero"]) + self.assert_proc_fails(proc) + self.check_fpga0_attribute("flags", "0") + + def test_set_state(self): + """Test setting state (read-only attribute) fails.""" + old = self.get_fpga0_attribute("state") + + proc = self.run_fpgad(["--platform", self.PLATFORM, "set", "state", "0"]) + self.assert_proc_fails(proc) + self.check_fpga0_attribute("state", old) + + def test_set_flags_float(self): + """Test setting flags with float value fails.""" + proc = self.run_fpgad(["--platform", self.PLATFORM, "set", "flags", "0.2"]) + self.assert_proc_fails(proc) + self.check_fpga0_attribute("flags", "0") + + def test_set_flags_zero(self): + """Test setting flags to zero.""" + proc = self.run_fpgad(["--platform", self.PLATFORM, "set", "flags", "0"]) + self.assert_proc_succeeds(proc) + self.assert_in_proc_out( + "0 written to /sys/class/fpga_manager/fpga0/flags", proc + ) + self.check_fpga0_attribute("flags", "0") diff --git a/tests/snap_testing/test_universal/test_status.py b/tests/snap_testing/test_universal/test_status.py new file mode 100644 index 00000000..6ef1d591 --- /dev/null +++ b/tests/snap_testing/test_universal/test_status.py @@ -0,0 +1,33 @@ +#!/usr/bin/env python3 +"""Status command tests for universal platform.""" + +from common.base_test import FPGATestBase + + +class TestStatusUniversal(FPGATestBase): + """Test status command with --platform universal.""" + + PLATFORM = "universal" + + def test_status_executes(self): + """Test status command executes successfully.""" + proc = self.run_fpgad(["--platform", self.PLATFORM, "status"]) + self.assert_proc_succeeds(proc) + + def test_status_with_bitstream(self): + """Test status shows operating state after loading bitstream.""" + load_proc = self.run_fpgad( + [ + "--platform", + self.PLATFORM, + "load", + "bitstream", + "./fpgad/k26-starter-kits/k26_starter_kits.bit.bin", + ] + ) + self.assert_proc_succeeds( + load_proc, "Failed to load a bitstream before checking status." + ) + status_proc = self.run_fpgad(["--platform", self.PLATFORM, "status"]) + self.assert_proc_succeeds(status_proc) + self.assert_in_proc_out("operating", status_proc) diff --git a/tests/snap_testing/test_xlnx/__init__.py b/tests/snap_testing/test_xlnx/__init__.py new file mode 100644 index 00000000..ef05c18b --- /dev/null +++ b/tests/snap_testing/test_xlnx/__init__.py @@ -0,0 +1 @@ +"""Xilinx platform tests.""" diff --git a/tests/snap_testing/test_xlnx/test_bitstream.py b/tests/snap_testing/test_xlnx/test_bitstream.py new file mode 100644 index 00000000..56e6dca9 --- /dev/null +++ b/tests/snap_testing/test_xlnx/test_bitstream.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 +"""Bitstream loading tests for xlnx platform.""" + +import os +import unittest +from pathlib import Path +from common.base_test import FPGATestBase +from common.helpers import ( + BAD_FLAGS, + set_flags, + is_dfx_mgr_available, +) + + +@unittest.skipUnless( + is_dfx_mgr_available(), + "dfx-mgr component not installed. Install with: snap install fpgad+dfx-mgr.comp", +) +class TestBitstreamXlnx(FPGATestBase): + """Test bitstream loading operations with --platform xlnx.""" + + PLATFORM = "xlnx" + + def test_load_bitstream_local(self): + """Test loading bitstream from local relative path.""" + path_str = "./fpgad/k26-starter-kits/k26_starter_kits.bit.bin" + proc = self.run_fpgad( + ["--platform", self.PLATFORM, "load", "bitstream", path_str] + ) + self.assert_proc_succeeds(proc) + # dfx-mgr output format: "Loaded with slot_handle X" + self.assertIn("Loaded with slot_handle", proc.stdout) + + def test_load_bitstream_home_fullpath(self): + """Test loading bitstream from full absolute path.""" + prefix = Path(os.getcwd()) + path = prefix.joinpath("fpgad/k26-starter-kits/k26_starter_kits.bit.bin") + proc = self.run_fpgad( + ["--platform", self.PLATFORM, "load", "bitstream", str(path)] + ) + self.assert_proc_succeeds(proc) + self.assert_in_proc_out("Loaded with slot_handle", proc) + + def test_load_bitstream_path_not_exist(self): + """Test loading bitstream from non-existent path fails.""" + proc = self.run_fpgad( + [ + "--platform", + self.PLATFORM, + "load", + "bitstream", + "/this/path/is/fake.bit.bin", + ] + ) + self.assert_proc_fails(proc) + # dfx-mgr returns error via softener + self.assert_in_proc_err("FpgadError::Softener:", proc) + + def test_load_bitstream_containing_dir(self): + """Test loading bitstream with directory path fails.""" + prefix = Path(os.getcwd()) + path = prefix.joinpath("fpgad/k26-starter-kits/") + proc = self.run_fpgad( + ["--platform", self.PLATFORM, "load", "bitstream", str(path)] + ) + self.assert_proc_fails(proc) + # dfx-mgr returns error via softener + self.assert_in_proc_err("FpgadError::Softener:", proc) + + def test_load_bitstream_bad_flags(self): + """Test loading bitstream with invalid flags fails.""" + prefix = Path(os.getcwd()) + path = prefix.joinpath("fpgad/k26-starter-kits/k26_starter_kits.bit.bin") + set_flags(BAD_FLAGS) + proc = self.run_fpgad( + ["--platform", self.PLATFORM, "load", "bitstream", str(path)] + ) + # Note: dfx-mgr may succeed even with bad flags, just check it doesn't crash + # The bad flags are for the fpga_manager, not dfx-mgr + if proc.returncode != 0: + self.assert_in_proc_err("FpgadError", proc) diff --git a/tests/snap_testing/test_xlnx/test_overlay.py b/tests/snap_testing/test_xlnx/test_overlay.py new file mode 100644 index 00000000..49e44cc4 --- /dev/null +++ b/tests/snap_testing/test_xlnx/test_overlay.py @@ -0,0 +1,40 @@ +#!/usr/bin/env python3 +"""Overlay loading tests for xlnx platform.""" + +import unittest +from pathlib import Path + +from common.base_test import FPGATestBase +from common.helpers import is_dfx_mgr_available + + +@unittest.skipUnless( + is_dfx_mgr_available(), + "dfx-mgr component not installed. Install with: snap install fpgad+dfx-mgr.comp", +) +class TestOverlayXlnx(FPGATestBase): + """Test overlay loading operations with --platform xlnx.""" + + PLATFORM = "xlnx" + + def test_load_overlay_bad_path(self): + """Test loading overlay from non-existent path fails.""" + overlay_path = Path("/path/does/not/exist") + proc = self.run_fpgad( + ["--platform", self.PLATFORM, "load", "overlay", str(overlay_path)] + ) + self.assert_proc_fails(proc) + self.assert_not_in_proc_out("loaded via", proc) + # dfx-mgr returns error via softener + self.assert_in_proc_err("FpgadError::Softener:", proc) + + def test_load_overlay_missing_bitstream(self): + """Test loading overlay without corresponding bitstream fails.""" + overlay_path = Path("./fpgad/k26-starter-kits/k26_starter_kits.dtbo") + proc = self.run_fpgad( + ["--platform", self.PLATFORM, "load", "overlay", str(overlay_path)] + ) + self.assert_proc_fails(proc) + self.assert_not_in_proc_out("loaded via", proc) + # dfx-mgr returns error via softener (permission denied or file read error) + self.assert_in_proc_err("FpgadError::Softener:", proc) diff --git a/tests/snap_testing/test_xlnx/test_set.py b/tests/snap_testing/test_xlnx/test_set.py new file mode 100644 index 00000000..5184e716 --- /dev/null +++ b/tests/snap_testing/test_xlnx/test_set.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python3 +"""Set command tests for xlnx platform.""" + +from common.base_test import FPGATestBase + + +class TestSetXlnx(FPGATestBase): + """Test set command with --platform xlnx.""" + + PLATFORM = "xlnx" + + def test_set_flags_nonzero(self): + """Test setting flags to non-zero value.""" + proc = self.run_fpgad(["--platform", self.PLATFORM, "set", "flags", "20"]) + self.assert_proc_succeeds(proc) + self.assert_in_proc_out( + "20 written to /sys/class/fpga_manager/fpga0/flags", proc + ) + self.check_fpga0_attribute("flags", "20") + + def test_set_flags_string(self): + """Test setting flags with invalid string value fails.""" + proc = self.run_fpgad(["--platform", self.PLATFORM, "set", "flags", "zero"]) + self.assert_proc_fails(proc) + self.check_fpga0_attribute("flags", "0") + + def test_set_state(self): + """Test setting state (read-only attribute) fails.""" + old = self.get_fpga0_attribute("state") + + proc = self.run_fpgad(["--platform", self.PLATFORM, "set", "state", "0"]) + self.assert_proc_fails(proc) + self.check_fpga0_attribute("state", old) + + def test_set_flags_float(self): + """Test setting flags with float value fails.""" + proc = self.run_fpgad(["--platform", self.PLATFORM, "set", "flags", "0.2"]) + self.assert_proc_fails(proc) + self.check_fpga0_attribute("flags", "0") + + def test_set_flags_zero(self): + """Test setting flags to zero.""" + proc = self.run_fpgad(["--platform", self.PLATFORM, "set", "flags", "0"]) + self.assert_proc_succeeds(proc) + self.assert_in_proc_out( + "0 written to /sys/class/fpga_manager/fpga0/flags", proc + ) + self.check_fpga0_attribute("flags", "0") diff --git a/tests/snap_testing/test_xlnx/test_status.py b/tests/snap_testing/test_xlnx/test_status.py new file mode 100644 index 00000000..a99b3894 --- /dev/null +++ b/tests/snap_testing/test_xlnx/test_status.py @@ -0,0 +1,66 @@ +#!/usr/bin/env python3 +"""Status command tests for xlnx platform.""" + +import unittest + +from common.base_test import FPGATestBase +from common.helpers import is_dfx_mgr_available + + +@unittest.skipUnless( + is_dfx_mgr_available(), + "dfx-mgr component not installed. Install with: snap install fpgad+dfx-mgr.comp", +) +class TestStatusXlnx(FPGATestBase): + """Test status command with --platform xlnx.""" + + PLATFORM = "xlnx" + + def test_status_executes(self): + """Test status command executes successfully.""" + proc = self.run_fpgad(["--platform", self.PLATFORM, "status"]) + self.assert_proc_succeeds(proc) + + def test_status_with_bitstream(self): + """Test status shows operating state after loading bitstream.""" + load_proc = self.run_fpgad( + [ + "--platform", + self.PLATFORM, + "load", + "bitstream", + "./fpgad/k26-starter-kits/k26_starter_kits.bit.bin", + ] + ) + self.assert_proc_succeeds( + load_proc, "Failed to load a bitstream before checking status." + ) + + status_proc = self.run_fpgad(["--platform", self.PLATFORM, "status"]) + self.assert_proc_succeeds(status_proc) + # dfx-mgr returns table with "slot->handle" and bitstream filename + self.assert_in_proc_out("slot->handle", status_proc) + self.assert_in_proc_out("k26_starter_kits.bit.bin", status_proc) + + def test_status_failed_overlay(self): + """Test status shows error after failed overlay load.""" + load_proc = self.run_fpgad( + [ + "--platform", + self.PLATFORM, + "load", + "overlay", + "./fpgad/k26-starter-kits/k26_starter_kits.dtbo", + ] + ) + self.assertNotEqual( + load_proc.returncode, + 0, + "Overlay load succeeded and therefore test has failed.", + ) + + proc = self.run_fpgad(["--platform", self.PLATFORM, "status"]) + self.assert_proc_succeeds(proc) + # dfx-mgr returns empty table when nothing is loaded, not "error" + # Just check it returns successfully + self.assert_in_proc_out("slot->handle", proc) diff --git a/tests/snap_tests.py b/tests/snap_tests.py deleted file mode 100644 index 07c94d13..00000000 --- a/tests/snap_tests.py +++ /dev/null @@ -1,630 +0,0 @@ -#!/usr/bin/env python3 -""" -FPGA snap test framework using a well-defined named test type -with human-readable output. -""" - -import os -import subprocess -import unittest -from pathlib import Path -from subprocess import CompletedProcess -from typing import List - -import shutil - -BAD_FLAGS = 223 - - -class Colors: - GREEN = "\033[92m" - RED = "\033[91m" - YELLOW = "\033[93m" - CYAN = "\033[96m" - RESET = "\033[0m" - - -class TestFPGAdCLI(unittest.TestCase): - def setUp(self): - """ - Runs before each tests in this class. - """ - self.cleanup_applied_overlays() - self.reset_flags() - - @classmethod - def tearDownClass(cls): - """ - Runs once after all tests in this class are finished - """ - cls.cleanup_applied_overlays() - cls.set_flags(0) - - # ============================================================ - # ======================= USEFUL DATA ======================== - # ============================================================ - - class TestData: - def __init__(self, source: Path, target: Path): - """ - Useful container for copying files during tests. Defines the source and target locations for use with - copy_test_data_files and cleanup_test_data_files - :param source: the path to the file which should be copied - :param target: the path to which the file should be copied/was copied to - """ - self.source = source - self.target = target - - # ============================================================ - # ===================== HELPER FUNCTIONS ===================== - # ============================================================ - def assert_proc_succeeds(self, proc, msg=None): - """Assert that a process completed successfully, including stdout/stderr on failure.""" - if msg is None: - msg = f"Return code is {proc.returncode} when expecting 0" - full_msg = ( - f"{msg}\n" - f"Status code:\t{proc.returncode}\n" - f"stdout:\t{proc.stdout}\n" - f"stderr:\t{proc.stderr}" - ) - self.assertEqual(proc.returncode, 0, full_msg) - - def assert_proc_fails(self, proc, msg=None): - """Assert that a process completed successfully, including stdout/stderr on failure.""" - if msg is None: - msg = f"Return code is {proc.returncode} when expecting nonzero" - full_msg = ( - f"{msg}\n" - f"Status code:\t{proc.returncode}\n" - f"stdout:\t{proc.stdout}\n" - f"stderr:\t{proc.stderr}" - ) - self.assertNotEqual(proc.returncode, 0, full_msg) - - def assert_in_proc_out( - self, substring: str, proc: CompletedProcess, msg: str = None - ): - """Assert that a substring exists in output, including stdout/stderr on failure.""" - if msg is None: - msg = f"'{substring}' not found in output." - full_msg = f"{msg}\nstdout:\t{proc.stdout}\nstderr:\t{proc.stderr}" - self.assertIn(substring, proc.stdout, full_msg) - - def assert_not_in_proc_out( - self, substring: str, proc: CompletedProcess, msg: str = None - ): - """Assert that a substring exists in output, including stdout/stderr on failure.""" - if msg is None: - msg = f"Undesired '{substring}' found in output." - full_msg = f"{msg}\nstdout:\t{proc.stdout}\nstderr:\t{proc.stderr}" - self.assertNotIn(substring, proc.stdout, full_msg) - - def assert_in_proc_err( - self, substring: str, proc: CompletedProcess, msg: str = None - ): - """Assert that a substring exists in output, including stdout/stderr on failure.""" - if msg is None: - msg = f"'{substring}' not found in stderr." - full_msg = f"{msg}\nstdout:\t{proc.stdout}\nstderr:\t{proc.stderr}" - self.assertIn(substring, proc.stderr, full_msg) - - @staticmethod - def get_fpga0_attribute(attr: str): - path = Path(f"/sys/class/fpga_manager/fpga0/{attr}") - - with open(path, "r") as f: - real_attr = f.read().strip() - return real_attr - - def check_fpga0_attribute(self, attr: str, expected: str): - path = Path(f"/sys/class/fpga_manager/fpga0/{attr}") - - with open(path, "r") as f: - real_attr = f.read().strip() - self.assertIn(expected, real_attr) - - @staticmethod - def copy_test_data_files(test_file: TestData) -> int: - """ - Copied a file from test_file.source to test_file.target, use to, e.g., copy a bitstream. - Use in conjunction with cleanup_test_data_files to test loading from a custom location. - :rtype: int - :param test_file: A TestData object which contains the relevant paths - :return: 0 on success, -1 on failure - """ - src = Path(test_file.source) - - if not src.exists(): - print(f"{Colors.YELLOW}[WARN]{Colors.RESET} Source file missing: {src}") - return -1 - - target = test_file.target - target_path = Path(target) - - # Ensure directory exists - target_path.parent.mkdir(parents=True, exist_ok=True) - - print(f"{Colors.CYAN}[INFO]{Colors.RESET} Copying {src} → {target_path}") - shutil.copy2(src, target_path) - return 0 - - @staticmethod - def cleanup_test_data_files(test_file: TestData) -> int: - """ - Cleans up the file located at test_file.target, use after copy_test_data_files - :return: - :rtype: int - :param test_file: A TestData object which contains the location to which the file was originally copied - :return: 0 on success, -1 on failure - """ - target = test_file.target - path = Path(target) - print(f"{Colors.CYAN}[INFO]{Colors.RESET} deleting {test_file.target}") - if not path.exists(): - print( - f"{Colors.YELLOW}[WARN]{Colors.RESET} Missing file during cleanup: {path}" - ) - return -1 - try: - path.unlink() - except Exception as e: - print(f"{Colors.RED}[ERROR]{Colors.RESET} Failed to remove {path}: {e}") - return -1 - - return 0 - - def load_bitstream(self, path: Path) -> CompletedProcess[str]: - """ - One line wrapper for calling fpgad to load a bitstream - (may have more functionality added in future) - :rtype: CompletedProcess[str] - :param path: path to the bitstream to load - :return: - """ - return self.run_fpgad(["load", "bitstream", str(path)]) - - def load_overlay(self, path: Path) -> CompletedProcess[str]: - """ - One line wrapper for calling fpgad to load an overlay - (may have more functionality added in future) - :rtype: CompletedProcess[str] - :param path: path to the overlay to load - :return: the completed process object, containing return code and captured output - """ - return self.run_fpgad(["load", "overlay", str(path)]) - - @staticmethod - def cleanup_applied_overlays(): - directory = "/sys/kernel/config/device-tree/overlays/" - print(f"{Colors.CYAN}[INFO]{Colors.RESET} Cleaning up applied overlays") - - for item in os.listdir(directory): - item_path = os.path.join(directory, item) - if os.path.isdir(item_path): - try: - os.rmdir(item_path) # Remove the directory itself - print( - f"{Colors.CYAN}[INFO]{Colors.RESET} Removed overlay directory at {item_path}" - ) - except PermissionError: - print( - f"{Colors.RED}[ERROR]{Colors.RESET} Permission denied removing {item_path}. Run as root." - ) - except OSError as e: - # This happens if the directory is not empty - print( - f"{Colors.RED}[ERROR]{Colors.RESET} Failed to remove {item_path}: {e}" - ) - - @staticmethod - def set_flags(flags: int = 0) -> None: - flags_path = r"/sys/class/fpga_manager/fpga0/flags" - try: - with open(flags_path, "w") as f: - f.write(f"{flags:X}") - print( - f"{Colors.CYAN}[INFO]{Colors.RESET} Successfully wrote {flags} to {flags_path}" - ) - except PermissionError: - print( - f"{Colors.RED}[ERROR]{Colors.RESET} Permission denied: you probably need to run as root" - ) - except FileNotFoundError: - print(f"{Colors.RED}[ERROR]{Colors.RESET} {flags_path} does not exist") - except Exception as e: - print(f"Error writing to {flags_path}: {e}") - - def reset_flags(self): - """ - Reset flags (to zero) using system calls, instead of fpgad. - :return: the completed process object, containing return code and captured output - """ - print(f"{Colors.CYAN}[INFO]{Colors.RESET} Resetting fpga0's flags to 0") - self.set_flags(0) - - def run_fpgad(self, args: List[str]) -> subprocess.CompletedProcess[str]: - """ - Run the fpgad cli with provided args as a subprocess - :rtype: subprocess.CompletedProcess[str] - :param args: list of arguments to provide to the fpgad cli call - :return: the completed process object, containing return code and captured output - """ - cmd = ["fpgad"] + args - print(f"{Colors.CYAN}[INFO]{Colors.RESET} Running: {' '.join(cmd)}") - - proc = subprocess.run( - cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True - ) - - return proc - - # ============================================================ - # ===================== TEST DEFINITIONS ===================== - # ============================================================ - - # -------------------------------------------------------- - # load bitstream tests - # -------------------------------------------------------- - - def test_load_bitstream_local(self): - path_str = "./fpgad/k26-starter-kits/k26_starter_kits.bit.bin" - proc = self.load_bitstream(Path(path_str)) - self.assert_proc_succeeds(proc) - self.assertIn("loaded to fpga0 using firmware lookup path", proc.stdout) - - def test_load_bitstream_home_fullpath(self): - prefix = Path(os.getcwd()) - path = prefix.joinpath("fpgad/k26-starter-kits/k26_starter_kits.bit.bin") - - proc = self.load_bitstream(path) - self.assert_proc_succeeds(proc) - self.assert_in_proc_out("loaded to fpga0 using firmware lookup path", proc) - - def test_load_bitstream_lib_firmware(self): - test_file_paths = self.TestData( - source=Path("./fpgad/k26-starter-kits/k26_starter_kits.bit.bin"), - target=Path("/lib/firmware/k26-starter-kits.bit.bin"), - ) - try: - self.copy_test_data_files(test_file_paths) != 0 - except Exception as e: - print( - f"Failed to copy {test_file_paths.source} to {test_file_paths.target}" - ) - raise e - - proc = self.load_bitstream(test_file_paths.target) - - try: - self.cleanup_test_data_files(test_file_paths) - except Exception as e: - print(f"Failed to clean up {test_file_paths.target}") - raise e - - self.assert_proc_succeeds(proc) - self.assert_in_proc_out("loaded to fpga0 using firmware lookup path", proc) - - def test_load_bitstream_lib_firmware_xilinx(self): - test_file_paths = self.TestData( - source=Path("./fpgad/k26-starter-kits/k26_starter_kits.bit.bin"), - target=Path( - "/lib/firmware/xilinx/k26_starter_kits/k26-starter-kits.bit.bin" - ), - ) - try: - self.copy_test_data_files(test_file_paths) != 0 - except Exception as e: - print( - f"Failed to copy {test_file_paths.source} to {test_file_paths.target}" - ) - raise e - - proc = self.load_bitstream(test_file_paths.target) - - try: - self.cleanup_test_data_files(test_file_paths) - except Exception as e: - print(f"Failed to clean up {test_file_paths.target}") - raise e - - self.assert_proc_succeeds(proc) - self.assert_in_proc_out("loaded to fpga0 using firmware lookup path", proc) - - def test_load_bitstream_path_not_exist(self): - proc = self.load_bitstream(Path("/this/path/is/fake.bit.bin")) - self.assert_proc_fails(proc) - self.assert_in_proc_err("FpgadError::IOWrite:", proc) - - def test_load_bitstream_containing_dir(self): - prefix = Path(os.getcwd()) - path = prefix.joinpath("fpgad/k26-starter-kits/") - - proc = self.load_bitstream(path) - self.assert_proc_fails(proc) - self.assert_in_proc_err("FpgadError::IOWrite:", proc) - - def test_load_bitstream_bad_flags(self): - prefix = Path(os.getcwd()) - path = prefix.joinpath("fpgad/k26-starter-kits/k26_starter_kits.bit.bin") - - self.set_flags(BAD_FLAGS) - - proc = self.load_bitstream(path) - self.assert_proc_fails(proc) - self.assert_in_proc_err("FpgadError::IOWrite:", proc) - - # -------------------------------------------------------- - # load overlay tests - # -------------------------------------------------------- - - ### overlay cases: - # load from relative path - # load from /lib/firmware - # load from full path not in /lib/firmware - # fail to load from bad path - # fail to load due to bad flags - - def test_load_overlay_local(self): - # Necessary due to bad dtbo content from upstream - test_file_paths = self.TestData( - source=Path("./fpgad/k26-starter-kits/k26_starter_kits.bit.bin"), - target=Path("./fpgad/k26-starter-kits/k26-starter-kits.bit.bin"), - ) - - try: - self.copy_test_data_files(test_file_paths) != 0 - except Exception as e: - print( - f"Failed to copy {test_file_paths.source} to {test_file_paths.target}" - ) - raise e - overlay_path = Path("./fpgad/k26-starter-kits/k26_starter_kits.dtbo") - proc = self.load_overlay(overlay_path) - try: - self.cleanup_test_data_files(test_file_paths) - except Exception as e: - print(f"Failed to clean up {test_file_paths.target}") - raise e - - self.assert_proc_succeeds(proc) - self.assert_in_proc_out("loaded via", proc) - - def test_load_overlay_lib_firmware(self): - # Necessary due to bad dtbo content from upstream - test_file_paths = [ - self.TestData( - source=Path("./fpgad/k26-starter-kits/k26_starter_kits.bit.bin"), - target=Path("/lib/firmware/k26-starter-kits.bit.bin"), - ), - self.TestData( - source=Path("./fpgad/k26-starter-kits/k26_starter_kits.dtbo"), - target=Path("/lib/firmware/k26-starter-kits.dtbo"), - ), - ] - for file in test_file_paths: - try: - self.copy_test_data_files(file) != 0 - except Exception as e: - print(f"Failed to copy {file.source} to {file.target}") - raise e - - overlay_path = Path("/lib/firmware/k26-starter-kits.dtbo") - proc = self.load_overlay(overlay_path) - for file in test_file_paths: - try: - self.cleanup_test_data_files(file) - except Exception as e: - print(f"Failed to clean up {file.target}") - raise e - - self.assert_proc_succeeds(proc) - self.assert_in_proc_out("loaded via", proc) - - def test_load_overlay_full_path(self): - # Necessary due to bad dtbo content from upstream - test_file_paths = self.TestData( - source=Path("./fpgad/k26-starter-kits/k26_starter_kits.bit.bin"), - target=Path("./fpgad/k26-starter-kits/k26-starter-kits.bit.bin"), - ) - - try: - self.copy_test_data_files(test_file_paths) != 0 - except Exception as e: - print( - f"Failed to copy {test_file_paths.source} to {test_file_paths.target}" - ) - raise e - prefix = Path(os.getcwd()) - overlay_path = prefix.joinpath("fpgad/k26-starter-kits/k26_starter_kits.dtbo") - proc = self.load_overlay(overlay_path) - try: - self.cleanup_test_data_files(test_file_paths) - except Exception as e: - print(f"Failed to clean up {test_file_paths.target}") - raise e - - self.assert_proc_succeeds(proc) - self.assert_in_proc_out("loaded via", proc) - - def test_load_overlay_bad_path(self): - overlay_path = Path("/path/does/not/exist") - proc = self.load_overlay(overlay_path) - self.assert_proc_fails(proc) - self.assert_not_in_proc_out("loaded via", proc) - self.assert_in_proc_err("FpgadError::OverlayStatus:", proc) - - def test_load_overlay_missing_bitstream(self): - # TODO: if the dtbo gets fixed, then this test needs to be re-written. - overlay_path = Path("./fpgad/k26-starter-kits/k26_starter_kits.dtbo") - proc = self.load_overlay(overlay_path) - self.assert_proc_fails(proc) - self.assert_not_in_proc_out("loaded via", proc) - self.assert_in_proc_err("FpgadError::OverlayStatus:", proc) - - def test_load_overlay_bad_flags(self): - self.set_flags(BAD_FLAGS) - # Necessary due to bad dtbo content from upstream - test_file_paths = self.TestData( - source=Path("./fpgad/k26-starter-kits/k26_starter_kits.bit.bin"), - target=Path("./fpgad/k26-starter-kits/k26-starter-kits.bit.bin"), - ) - - try: - self.copy_test_data_files(test_file_paths) != 0 - except Exception as e: - print( - f"Failed to copy {test_file_paths.source} to {test_file_paths.target}" - ) - raise e - overlay_path = Path("./fpgad/k26-starter-kits/k26_starter_kits.dtbo") - proc = self.load_overlay(overlay_path) - try: - self.cleanup_test_data_files(test_file_paths) - except Exception as e: - print(f"Failed to clean up {test_file_paths.target}") - raise e - - self.assert_proc_fails(proc) - self.assert_not_in_proc_out("loaded via", proc) - self.assert_in_proc_err("FpgadError::OverlayStatus:", proc) - - # -------------------------------------------------------- - # status tests - # -------------------------------------------------------- - - def test_status_executes(self): - proc = self.run_fpgad(["status"]) - self.assert_proc_succeeds(proc) - - def test_status_with_bitstream(self): - load_proc = self.load_bitstream( - Path("./fpgad/k26-starter-kits/k26_starter_kits.bit.bin") - ) - self.assert_proc_succeeds( - load_proc, "Failed to load a bitstream before checking status." - ) - - status_proc = self.run_fpgad(["status"]) - self.assert_proc_succeeds(status_proc) - self.assert_in_proc_out("operating", status_proc) - - def test_status_with_overlay(self): - test_file_paths = self.TestData( - source=Path("./fpgad/k26-starter-kits/k26_starter_kits.bit.bin"), - target=Path("./fpgad/k26-starter-kits/k26-starter-kits.bit.bin"), - ) - try: - self.copy_test_data_files(test_file_paths) != 0 - except Exception as e: - print( - f"Failed to copy {test_file_paths.source} to {test_file_paths.target}" - ) - raise e - load_proc = self.load_overlay( - Path("./fpgad/k26-starter-kits/k26_starter_kits.dtbo") - ) - self.cleanup_test_data_files(test_file_paths) - self.assert_proc_succeeds(load_proc) - - status_proc = self.run_fpgad(["status"]) - self.assert_proc_succeeds(status_proc) - self.assert_in_proc_out("applied", status_proc) - self.assert_in_proc_out("operating", status_proc) - self.assert_in_proc_out("k26_starter_kits.dtbo", status_proc) - self.assert_not_in_proc_out("error", status_proc) - - def test_status_failed_overlay(self): - load_proc = self.load_overlay( - Path("./fpgad/k26-starter-kits/k26_starter_kits.dtbo") - ) - self.assertNotEqual( - load_proc.returncode, - 0, - "Overlay load succeeded and therefore test has failed.", - ) - - proc = self.run_fpgad(["status"]) - self.assert_proc_succeeds(proc) - self.assert_in_proc_out("error", proc) - - # -------------------------------------------------------- - # set tests - # -------------------------------------------------------- - - def test_set_flags_nonzero(self): - proc = self.run_fpgad(["set", "flags", "20"]) - self.assert_proc_succeeds(proc) - self.assert_in_proc_out( - "20 written to /sys/class/fpga_manager/fpga0/flags", proc - ) - self.check_fpga0_attribute("flags", "20") - - def test_set_flags_string(self): - proc = self.run_fpgad(["set", "flags", "zero"]) - self.assert_proc_fails(proc) - self.check_fpga0_attribute("flags", "0") - - def test_set_state(self): - old = self.get_fpga0_attribute("state") - - proc = self.run_fpgad(["set", "state", "0"]) - self.assert_proc_fails(proc) - self.check_fpga0_attribute("state", old) - - def test_set_flags_float(self): - proc = self.run_fpgad(["set", "flags", "0.2"]) - self.assert_proc_fails(proc) - self.check_fpga0_attribute("flags", "0") - - def test_set_flags_zero(self): - proc = self.run_fpgad(["set", "flags", "0"]) - self.assert_proc_succeeds(proc) - self.assert_in_proc_out( - "0 written to /sys/class/fpga_manager/fpga0/flags", proc - ) - self.check_fpga0_attribute("flags", "0") - - # -------------------------------------------------------- - # help tests - # -------------------------------------------------------- - - def test_help_main(self): - proc = self.run_fpgad(["help"]) - self.assert_proc_succeeds(proc) - - def test_help_main_as_flag(self): - proc = self.run_fpgad(["--help"]) - self.assert_proc_succeeds(proc) - - def test_help_set(self): - proc = self.run_fpgad(["help", "set"]) - self.assert_proc_succeeds(proc) - - def test_help_remove(self): - proc = self.run_fpgad(["help", "remove"]) - self.assert_proc_succeeds(proc) - - def test_help_remove_overlay(self): - proc = self.run_fpgad(["help", "remove", "overlay"]) - self.assert_proc_succeeds(proc) - - def test_help_remove_bitstream(self): - proc = self.run_fpgad(["help", "remove", "bitstream"]) - self.assert_proc_succeeds(proc) - - def test_help_load(self): - proc = self.run_fpgad(["help", "load"]) - self.assert_proc_succeeds(proc) - - def test_help_load_bitstream(self): - proc = self.run_fpgad(["help", "load", "bitstream"]) - self.assert_proc_succeeds(proc) - - def test_help_load_overlay(self): - proc = self.run_fpgad(["help", "load", "overlay"]) - self.assert_proc_succeeds(proc) - - -if __name__ == "__main__": - unittest.main()