jyapayne/switch-pico
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Restore profile editor management on the native Joy-Con hub

Browse source
This commit is contained in:
Joey Yakimowich-Payne 2026-09-13 08:35:10 -06:00
commit dd69f3d487
30 changed files with 1849 additions and 348 deletions
Download patch file Download diff file Expand all files Collapse all files

352
tests/native_hub_management_test.cpp Normal file
View file

@ -0,0 +1,352 @@
#include "usb/usb_configuration_management.h"
#include "usb/native_hub/native_hub.h"
#include "platform/pico/pico_profile_storage.h"
#include "adapter/adapter_usb_mode.h"
#include "bootsel.h"
#include <algorithm>
#include <array>
#include <cstdlib>
#include <cstring>
#include <iostream>
#include <vector>
extern "C" {
void native_test_initialize(void);
void native_test_drain(void);
bool native_test_setup(uint8_t, const tusb_control_request_t*, bool);
bool native_test_out(uint8_t, const uint8_t*, uint16_t, bool);
bool native_test_in(uint8_t, uint8_t*, uint16_t*, bool);
void native_test_bus_reset(bool);
}
namespace {
using namespace UsbConfigurationManagement;
std::array<uint8_t, PROFILE_STORAGE_ARENA_COUNT * PROFILE_STORAGE_ARENA_SIZE> flash;
uint32_t programs = 0;
uint32_t erases = 0;
uint32_t bootsel_calls = 0;
std::array<uint8_t, 64> child_identity[2];
void require(bool condition, const char* message) {
if (!condition) { std::cerr << message << '\n'; std::exit(1); }
}
[[noreturn]] void unexpected_mutation() {
require(false, "profile editor invoked an unrelated configuration/pairing/reboot mutation");
std::abort();
}
uint16_t u16(const std::vector<uint8_t>& data, size_t offset) {
return data.at(offset) | (static_cast<uint16_t>(data.at(offset + 1)) << 8);
}
uint32_t u32(const std::vector<uint8_t>& data, size_t offset) {
return u16(data, offset) | (static_cast<uint32_t>(u16(data, offset + 2)) << 16);
}
void put16(std::vector<uint8_t>& data, size_t offset, uint16_t value) {
data.at(offset) = value; data.at(offset + 1) = value >> 8;
}
void put32(std::vector<uint8_t>& data, size_t offset, uint32_t value) {
put16(data, offset, value); put16(data, offset + 2, value >> 16);
}
tusb_control_request_t request(Operation op, bool input, uint16_t length) {
tusb_control_request_t setup{};
setup.bmRequestType = input ? 0xc0 : 0x40;
setup.bRequest = static_cast<uint8_t>(op);
setup.wValue = kRequestValue; setup.wIndex = kRequestIndex; setup.wLength = length;
return setup;
}
std::vector<uint8_t> envelope(Operation op, const std::vector<uint8_t>& payload) {
std::vector<uint8_t> bytes(kRequestHeaderSize + payload.size());
memcpy(bytes.data(), "SPMG", 4);
bytes[4] = kProtocolVersion; bytes[5] = static_cast<uint8_t>(op);
put16(bytes, 8, payload.size());
put32(bytes, 12, configuration_crc32(payload.data(), payload.size()));
std::copy(payload.begin(), payload.end(), bytes.begin() + kRequestHeaderSize);
return bytes;
}
void acknowledge(uint8_t slot = 0, bool drain = true) {
uint8_t packet[64]; uint16_t length = 0xffff;
require(native_test_in(slot, packet, &length, drain) && length == 0,
"OUT transfer did not complete with a real zero-length status packet");
}
std::vector<uint8_t> receive(uint8_t slot = 0) {
std::vector<uint8_t> bytes;
for (;;) {
uint8_t packet[64]; uint16_t length = 0;
require(native_test_in(slot, packet, &length, true), "control IN packet was not available");
bytes.insert(bytes.end(), packet, packet + length);
if (length < 64) break;
}
require(native_test_out(slot, nullptr, 0, true), "control IN status OUT was rejected");
return bytes;
}
std::vector<uint8_t> read_operation(Operation op) {
const auto setup = request(op, true, kMaximumResponseSize);
require(native_test_setup(0, &setup, true), "root management read stalled");
auto bytes = receive();
require(bytes.size() >= kResponseHeaderSize && memcmp(bytes.data(), "SPMG", 4) == 0 &&
bytes[5] == static_cast<uint8_t>(op) &&
bytes.size() == kResponseHeaderSize + u16(bytes, 8), "management read envelope is corrupt");
require(u32(bytes, 16) == configuration_crc32(bytes.data() + kResponseHeaderSize,
bytes.size() - kResponseHeaderSize), "multi-packet response CRC is corrupt");
return bytes;
}
void write_operation(Operation op, const std::vector<uint8_t>& payload, bool ack = true) {
const auto bytes = envelope(op, payload);
const auto setup = request(op, false, bytes.size());
require(native_test_setup(0, &setup, true), "root management write setup stalled");
for (size_t offset = 0; offset < bytes.size(); offset += 64) {
require(native_test_out(0, bytes.data() + offset,
std::min<size_t>(64, bytes.size() - offset), true), "management OUT packet stalled");
}
if (ack) acknowledge();
}
std::vector<uint8_t> identity_payload(uint8_t profile) {
std::vector<uint8_t> payload(15);
require(controller_identity_encode(controller_identity_global(), payload.data(), 14), "global identity did not encode");
payload[14] = profile; return payload;
}
std::vector<uint8_t> encoded_profile(uint8_t destination) {
ControllerProfile profile = controller_profile_default(controller_identity_global(), 0);
profile.button_map[0] = destination;
std::vector<uint8_t> bytes(CONTROLLER_PROFILE_ENCODED_SIZE);
require(controller_profile_encode(profile, bytes.data(), bytes.size()), "edited profile is invalid");
return bytes;
}
void begin_profile(uint32_t transaction, const std::vector<uint8_t>& profile) {
std::vector<uint8_t> payload(28);
put32(payload, 0, transaction);
require(controller_identity_encode(controller_identity_global(), payload.data() + 4, 14), "profile owner did not encode");
put16(payload, 20, CONTROLLER_PROFILE_SCHEMA_VERSION); put16(payload, 22, profile.size());
put32(payload, 24, configuration_crc32(profile.data(), profile.size()));
write_operation(Operation::kProfileBegin, payload);
// Selecting an existing owner is part of normal editor navigation and
// must not persist anything while a profile is merely being staged.
write_operation(Operation::kProfileSelect, identity_payload(0));
}
std::vector<uint8_t> chunk_payload(uint32_t transaction, const std::vector<uint8_t>& profile, size_t offset) {
const size_t count = std::min(kMaximumChunkSize, profile.size() - offset);
std::vector<uint8_t> payload(8 + count);
put32(payload, 0, transaction); put16(payload, 4, offset); put16(payload, 6, count);
std::copy_n(profile.data() + offset, count, payload.data() + 8);
return payload;
}
void stage_profile(uint32_t transaction, const std::vector<uint8_t>& profile) {
begin_profile(transaction, profile);
for (size_t offset = 0; offset < profile.size(); offset += kMaximumChunkSize)
write_operation(Operation::kProfileChunk, chunk_payload(transaction, profile, offset));
}
std::vector<uint8_t> transaction_payload(uint32_t id) { std::vector<uint8_t> data(4); put32(data, 0, id); return data; }
void require_profile(const std::vector<uint8_t>& expected) {
auto bytes = read_operation(Operation::kProfileRead);
require(bytes[6] == static_cast<uint8_t>(Status::kOk) &&
std::vector<uint8_t>(bytes.begin() + kResponseHeaderSize, bytes.end()) == expected,
"host readback differs from the durable selected profile");
}
void read_child(uint8_t slot) {
tusb_control_request_t setup{};
setup.bmRequestType = 0xc0; setup.bRequest = 3; setup.wLength = 128;
require(native_test_setup(slot, &setup, true), "native child identity stalled");
const auto bytes = receive(slot);
require(bytes == std::vector<uint8_t>(child_identity[slot - 1].begin(), child_identity[slot - 1].end()),
"native child identity leaked root or sibling vendor bytes");
}
void test_profile_transport() {
const uint32_t programs_before = programs, erases_before = erases;
const auto original = encoded_profile(0);
const auto edited = encoded_profile(4);
auto info = read_operation(Operation::kInfo);
require(info[kResponseHeaderSize] == 0 && info[kResponseHeaderSize + 1] == 72 &&
info[kResponseHeaderSize + 2] == 0 && info[kResponseHeaderSize + 4] == 5 &&
info[kResponseHeaderSize + 5] == 7, "native INFO does not describe the fixed image");
auto list = read_operation(Operation::kProfileList);
require(list[kResponseHeaderSize] == 1 && list.size() > 64, "root catalog omitted the global profile owner");
auto playtest = read_operation(Operation::kProfilePlaytest);
require(playtest[kResponseHeaderSize] == 0 && playtest[kResponseHeaderSize + 1] == 0xff,
"disconnected playtest fabricated controller input");
require_profile(original);
require(programs == programs_before && erases == erases_before, "editor reads wrote saved storage");
for (uint8_t slot : {1, 2}) {
const auto management = request(Operation::kProfileList, true, kMaximumResponseSize);
require(!native_test_setup(slot, &management, true), "native child accepted regular management");
read_child(slot);
}
for (Operation op : {Operation::kModeSet, Operation::kReboot}) {
const auto setup = request(op, false, kRequestHeaderSize + (op == Operation::kModeSet ? 5 : 4));
require(!native_test_setup(0, &setup, true), "fixed native image accepted mode switching");
}
begin_profile(1, edited);
const auto chunk = envelope(Operation::kProfileChunk, chunk_payload(1, edited, 0));
const auto setup = request(Operation::kProfileChunk, false, chunk.size());
require(native_test_setup(0, &setup, true) && native_test_out(0, chunk.data(), 64, true), "first full OUT packet failed");
read_child(1); read_child(2);
const auto child_management = request(Operation::kInfo, true, kMaximumResponseSize);
require(!native_test_setup(1, &child_management, true), "child INFO was accepted during a root write");
require(native_test_out(0, chunk.data() + 64, chunk.size() - 64, true), "interleaved child requests corrupted root OUT tail");
acknowledge();
for (size_t offset = kMaximumChunkSize; offset < edited.size(); offset += kMaximumChunkSize)
write_operation(Operation::kProfileChunk, chunk_payload(1, edited, offset));
write_operation(Operation::kProfileCommit, transaction_payload(1), false);
profile_service_task_on_storage_core(1000);
require(programs == programs_before && erases == erases_before, "profile persisted before its status ACK");
// Host sends its next SETUP before Core0 drains the already completed ACK.
acknowledge(0, false);
const auto next = request(Operation::kInfo, true, kMaximumResponseSize);
require(native_test_setup(0, &next, false), "next root SETUP was rejected");
native_test_drain(); receive();
profile_service_task_on_storage_core(1000);
require_profile(edited);
auto status = read_operation(Operation::kProfileTransactionStatus);
require(status[6] == static_cast<uint8_t>(Status::kOk) && u32(status, kResponseHeaderSize) == 1,
"genuine queued ACK lost its commit when the next SETUP arrived");
// Reopen the actual storage journal, not the service's published cache.
ProfileStorage reopened;
ControllerProfile stored{}, sibling{};
require(reopened.initialize(pico_profile_storage_io()) &&
reopened.get(controller_identity_global(), 0, &stored) == ProfileStorageResult::kOk &&
reopened.get(controller_identity_global(), 1, &sibling) == ProfileStorageResult::kOk &&
stored.button_map[0] == 4 && sibling.button_map[0] == 0,
"profile journal lost the edit or modified an unrelated profile");
}
void test_interrupted_transactions() {
const auto retained = encoded_profile(4), replacement = encoded_profile(5);
const uint32_t programs_before = programs, erases_before = erases;
stage_profile(2, replacement);
write_operation(Operation::kProfileCommit, transaction_payload(2), false);
// No status token completed: a fresh root SETUP aborts the write.
read_operation(Operation::kInfo);
profile_service_task_on_storage_core(2000);
require_profile(retained);
require(programs == programs_before && erases == erases_before, "aborted status committed stale profile data");
// The shared service intentionally retains incomplete receives. An
// explicit wrong-ID commit ends one with OutOfOrder, never a flash write.
write_operation(Operation::kProfileCommit, transaction_payload(0));
begin_profile(3, replacement);
const auto chunk = envelope(Operation::kProfileChunk, chunk_payload(3, replacement, 0));
const auto setup = request(Operation::kProfileChunk, false, chunk.size());
require(native_test_setup(0, &setup, true) && native_test_out(0, chunk.data(), 64, true), "aborted-packet setup failed");
read_operation(Operation::kInfo);
require(!native_test_out(0, chunk.data() + 64, chunk.size() - 64, true), "new SETUP left an aborted OUT tail armed");
require(native_test_setup(0, &setup, true) && native_test_out(0, chunk.data(), 64, true), "short-packet setup failed");
require(!native_test_out(0, chunk.data() + 64, chunk.size() - 65, true), "short OUT tail was accepted");
auto status = read_operation(Operation::kProfileTransactionStatus);
require(u16(status, kResponseHeaderSize + 4) == 0, "short OUT appended stale bytes to a profile");
auto corrupt = chunk;
corrupt[12] ^= 1;
require(native_test_setup(0, &setup, true) && native_test_out(0, corrupt.data(), 64, true), "bad-CRC setup failed");
require(!native_test_out(0, corrupt.data() + 64, corrupt.size() - 64, true), "bad CRC acquired a status ACK");
write_operation(Operation::kProfileCommit, transaction_payload(0));
stage_profile(4, replacement);
write_operation(Operation::kProfileCommit, transaction_payload(4), false);
// The SIE captured ACK, but a bus reset revoked the queued transaction
// before the service consumed it. A SETUP must not be conflated with reset.
acknowledge(0, false);
native_test_bus_reset(true);
profile_service_task_on_storage_core(3000);
require_profile(retained);
require(programs == programs_before && erases == erases_before, "bus reset committed a stale queued profile ACK");
write_operation(Operation::kProfileCommit, transaction_payload(0));
stage_profile(5, replacement);
write_operation(Operation::kProfileCommit, transaction_payload(5));
profile_service_task_on_storage_core(3000);
require_profile(replacement);
// A repeated status token is unarmed and cannot dispatch the commit twice.
uint8_t packet[64]; uint16_t length;
require(!native_test_in(0, packet, &length, true), "completed request retained a second status ACK");
}
void test_private_bootsel() {
const auto bytes = envelope(Operation::kBootselReboot, {});
const auto setup = request(Operation::kBootselReboot, false, bytes.size());
for (uint8_t slot : {0, 1, 2}) {
require(native_test_setup(slot, &setup, true), "private BOOTSEL setup stalled");
require(!native_test_out(slot, bytes.data(), bytes.size() - 1, true), "short BOOTSEL was accepted");
probe_bootsel_task(100); probe_bootsel_task(200);
require(bootsel_calls == 0, "short BOOTSEL rebooted the device");
require(native_test_setup(slot, &setup, true) && native_test_out(slot, bytes.data(), bytes.size(), true),
"valid private BOOTSEL envelope failed");
// An unrelated identity/INFO SETUP cancels an unacknowledged BOOTSEL.
if (slot) read_child(slot); else read_operation(Operation::kInfo);
probe_bootsel_task(300); probe_bootsel_task(400);
require(bootsel_calls == 0, "unacknowledged BOOTSEL rebooted the device");
}
require(native_test_setup(2, &setup, true) && native_test_out(2, bytes.data(), bytes.size(), true),
"validated child BOOTSEL failed");
acknowledge(2);
probe_bootsel_task(500); probe_bootsel_task(549);
require(bootsel_calls == 0, "BOOTSEL did not retain the post-ACK delay");
probe_bootsel_task(550);
require(bootsel_calls == 1, "validated child BOOTSEL did not reach ROM after the delay");
}
bool flash_read(void*, uint8_t arena, size_t offset, uint8_t* data, size_t size) {
if (arena >= PROFILE_STORAGE_ARENA_COUNT || offset > PROFILE_STORAGE_ARENA_SIZE ||
size > PROFILE_STORAGE_ARENA_SIZE - offset) return false;
memcpy(data, flash.data() + arena * PROFILE_STORAGE_ARENA_SIZE + offset, size); return true;
}
bool flash_erase(void*, uint8_t arena) {
if (arena >= PROFILE_STORAGE_ARENA_COUNT) return false;
++erases; memset(flash.data() + arena * PROFILE_STORAGE_ARENA_SIZE, 0xff, PROFILE_STORAGE_ARENA_SIZE); return true;
}
bool flash_program(void*, uint8_t arena, size_t offset, const uint8_t* data, size_t size) {
if (arena >= PROFILE_STORAGE_ARENA_COUNT || size != PROFILE_STORAGE_PAGE_SIZE ||
offset % PROFILE_STORAGE_PAGE_SIZE || offset + size > PROFILE_STORAGE_ARENA_SIZE) return false;
++programs;
uint8_t* destination = flash.data() + arena * PROFILE_STORAGE_ARENA_SIZE + offset;
for (size_t i = 0; i < size; ++i) destination[i] &= data[i];
return memcmp(destination, data, size) == 0;
}
} // namespace
ProfileStorageIo pico_profile_storage_io() {
return {nullptr, PROFILE_STORAGE_ARENA_SIZE, PROFILE_STORAGE_SECTOR_SIZE,
PROFILE_STORAGE_PAGE_SIZE, flash_read, flash_erase, flash_program};
}
uint32_t configuration_crc32(const uint8_t* data, size_t size) { return profile_storage_crc32(data, size); }
void configuration_service_snapshot(ConfigurationServiceSnapshot* out) {
*out = {}; out->state = ConfigurationServiceState::kReady;
out->configuration = adapter_configuration_default();
}
ConfigurationTransactionStatus configuration_service_begin(uint32_t, uint16_t, size_t, uint32_t) { unexpected_mutation(); }
ConfigurationTransactionStatus configuration_service_append(uint32_t, size_t, const uint8_t*, size_t) { unexpected_mutation(); }
ConfigurationTransactionStatus configuration_service_commit(uint32_t) { unexpected_mutation(); }
ConfigurationTransactionStatus configuration_service_reset(uint32_t) { unexpected_mutation(); }
ConfigurationTransactionStatus configuration_service_set_mode(uint32_t, AdapterRequestedMode, const AdapterModeAvailability&) { unexpected_mutation(); }
const AdapterModeAvailability& adapter_usb_mode_availability() { unexpected_mutation(); }
bool adapter_reboot_for_mode_transaction(uint32_t) { unexpected_mutation(); }
bool adapter_reboot_to_bootsel() { unexpected_mutation(); }
void bluepad32_input_backend_request_pairing_snapshot() { unexpected_mutation(); }
uint32_t bluepad32_input_backend_clear_pairings() { unexpected_mutation(); }
void bluepad32_input_backend_pairing_snapshot(Bluepad32PairingSnapshot* out) { *out = {}; }
void bluepad32_input_backend_playtest_snapshot(uint8_t, Bluepad32PlaytestSnapshot* out) { *out = {}; }
void bluepad32_input_backend_diagnostics(Bluepad32BackendDiagnostics* out) { *out = {}; }
bool bluepad32_input_backend_identify(const ControllerIdentity&) { return false; }
bool bluepad32_input_backend_set_wii_orientation(const ControllerIdentity&, uint32_t, bool) { return false; }
bool bluepad32_input_backend_capture_start(uint8_t, uint32_t, const CaptureOptions&) { return false; }
bool bluepad32_input_backend_capture_stop(uint32_t) { return false; }
bool bluepad32_input_backend_capture_page(uint32_t, uint16_t, Bluepad32CaptureSnapshot*) { return false; }
extern "C" void reset_usb_boot(uint32_t, uint32_t) { ++bootsel_calls; }
extern "C" bool tud_vendor_control_xfer_cb(uint8_t slot, uint8_t stage, const tusb_control_request_t* setup) {
if (probe_management_vendor_control(slot, stage, setup)) return true;
if (slot < 1 || slot > 2 || setup->bmRequestType != 0xc0 ||
setup->bRequest != 3 || setup->wValue || setup->wIndex) return false;
return stage != CONTROL_STAGE_SETUP || native_hub_control_xfer(slot, setup,
child_identity[slot - 1].data(), child_identity[slot - 1].size());
}
int main() {
static_assert(sizeof(tusb_control_request_t) == 8);
flash.fill(0xff); child_identity[0].fill(0x31); child_identity[1].fill(0x72);
profile_service_prepare(); profile_service_initialize_on_storage_core();
native_test_initialize();
test_profile_transport();
test_interrupted_transactions();
test_private_bootsel();
std::cout << "native root management packet and persistence regressions passed\n";
}

2
tests/native_hub_stubs/hardware/clocks.h Normal file
View file

@ -0,0 +1,2 @@
#pragma once
#include "hardware_stub.h"

2
tests/native_hub_stubs/hardware/irq.h Normal file
View file

@ -0,0 +1,2 @@
#pragma once
#include "hardware_stub.h"

2
tests/native_hub_stubs/hardware/resets.h Normal file
View file

@ -0,0 +1,2 @@
#pragma once
#include "hardware_stub.h"

2
tests/native_hub_stubs/hardware/structs/sio.h Normal file
View file

@ -0,0 +1,2 @@
#pragma once
#include "hardware_stub.h"

2
tests/native_hub_stubs/hardware/structs/usb.h Normal file
View file

@ -0,0 +1,2 @@
#pragma once
#include "hardware_stub.h"

2
tests/native_hub_stubs/hardware/structs/usb_dpram.h Normal file
View file

@ -0,0 +1,2 @@
#pragma once
#include "hardware_stub.h"

2
tests/native_hub_stubs/hardware/sync.h Normal file
View file

@ -0,0 +1,2 @@
#pragma once
#include "hardware_stub.h"

2
tests/native_hub_stubs/hardware/watchdog.h Normal file
View file

@ -0,0 +1,2 @@
#pragma once
#include "hardware_stub.h"

109
tests/native_hub_stubs/hardware_stub.h Normal file
View file

@ -0,0 +1,109 @@
#pragma once
#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>
#include <string.h>
#define __not_in_flash_func(name) name
#define __force_inline inline __attribute__((always_inline))
#define __dmb() ((void)0)
typedef struct { unsigned unused; } spin_lock_t;
static inline uint32_t save_and_disable_interrupts(void) { return 0; }
static inline void restore_interrupts(uint32_t flags) { (void)flags; }
static inline uint32_t spin_lock_blocking(spin_lock_t* lock) { (void)lock; return 0; }
static inline void spin_unlock(spin_lock_t* lock, uint32_t flags) { (void)lock; (void)flags; }
static inline bool spin_try_lock_unsafe(spin_lock_t* lock) { (void)lock; return true; }
static inline void spin_unlock_unsafe(spin_lock_t* lock) { (void)lock; }
static inline int spin_lock_claim_unused(bool required) { (void)required; return 0; }
static inline spin_lock_t* spin_lock_instance(unsigned index) {
static spin_lock_t lock; (void)index; return &lock;
}
static inline void hw_clear_bits(volatile uint32_t* address, uint32_t bits) { *address &= ~bits; }
static inline void hw_set_bits(volatile uint32_t* address, uint32_t bits) { *address |= bits; }
typedef struct {
volatile uint32_t ints, sie_status, buf_status, dev_addr_ctrl, inte;
volatile uint32_t ep_stall_arm, muxing, phy_direct, phy_direct_override;
volatile uint32_t pwr, main_ctrl, sie_ctrl, ep_nak_stall_status;
volatile uint32_t ep_tx_error, ep_rx_error;
} usb_hw_t;
typedef struct { volatile uint32_t in, out; } usb_pair_t;
typedef struct {
uint8_t setup_packet[8];
usb_pair_t ep_ctrl[15];
usb_pair_t ep_buf_ctrl[16];
uint8_t ep0_buf_a[64];
uint8_t padding[3776];
} usb_device_dpram_t;
typedef struct { volatile uint32_t mtime, gpio_hi_oe_clr; } sio_hw_t;
extern usb_hw_t native_test_usb;
extern usb_device_dpram_t native_test_dpram;
extern sio_hw_t native_test_sio;
#define usb_hw (&native_test_usb)
#define usb_dpram (&native_test_dpram)
#define sio_hw (&native_test_sio)
#define USBCTRL_DPRAM_BASE ((uintptr_t)usb_dpram)
#define USB_DPRAM_SIZE sizeof(*usb_dpram)
#define USB_BUF_CTRL_LEN_MASK 0x3ffu
#define USB_BUF_CTRL_AVAIL (1u << 10)
#define USB_BUF_CTRL_STALL (1u << 11)
#define USB_BUF_CTRL_SEL (1u << 12)
#define USB_BUF_CTRL_DATA1_PID (1u << 13)
#define USB_BUF_CTRL_LAST (1u << 14)
#define USB_BUF_CTRL_FULL (1u << 15)
#define EP_CTRL_ENABLE_BITS (1u << 31)
#define EP_CTRL_INTERRUPT_PER_BUFFER (1u << 29)
#define EP_CTRL_INTERRUPT_ON_NAK (1u << 16)
#define EP_CTRL_BUFFER_TYPE_LSB 26
#define USB_INTS_BUS_RESET_BITS (1u << 0)
#define USB_INTS_SETUP_REQ_BITS (1u << 1)
#define USB_INTS_DEV_SUSPEND_BITS (1u << 2)
#define USB_INTS_DEV_RESUME_FROM_HOST_BITS (1u << 3)
#define USB_INTS_BUFF_STATUS_BITS (1u << 4)
#define USB_SIE_STATUS_SETUP_REC_BITS (1u << 0)
#define USB_SIE_STATUS_BUS_RESET_BITS (1u << 1)
#define USB_SIE_STATUS_SUSPENDED_BITS (1u << 2)
#define USB_SIE_STATUS_RESUME_BITS (1u << 3)
#define USB_USB_MUXING_TO_PHY_BITS 1u
#define USB_USB_MUXING_SOFTCON_BITS 2u
#define USB_USB_MUXING_USBPHY_AS_GPIO_BITS 4u
#define SIO_GPIO_HI_IN_USB_DP_BITS 1u
#define SIO_GPIO_HI_IN_USB_DM_BITS 2u
#define USB_USBPHY_DIRECT_DP_PULLUP_EN_BITS 1u
#define USB_USBPHY_DIRECT_OVERRIDE_DP_PULLUP_EN_OVERRIDE_EN_BITS 1u
#define USB_USB_PWR_VBUS_DETECT_BITS 1u
#define USB_USB_PWR_VBUS_DETECT_OVERRIDE_EN_BITS 2u
#define USB_MAIN_CTRL_CONTROLLER_EN_BITS 1u
#define USB_SIE_CTRL_EP0_INT_1BUF_BITS 1u
#define USB_SIE_CTRL_PULLUP_EN_BITS 2u
#define RESETS_RESET_USBCTRL_BITS 1u
#define USBCTRL_IRQ 0u
#define clk_sys 0u
static inline uint32_t clock_get_hz(unsigned clock) { (void)clock; return 240000000u; }
static inline void reset_block(uint32_t mask) { (void)mask; }
static inline void unreset_block_wait(uint32_t mask) { (void)mask; }
static inline void multicore_launch_core1(void (*entry)(void)) { (void)entry; }
static inline void irq_set_exclusive_handler(unsigned irq, void (*fn)(void)) { (void)irq; (void)fn; }
static inline void irq_set_priority(unsigned irq, unsigned priority) { (void)irq; (void)priority; }
static inline void irq_set_enabled(unsigned irq, bool enabled) { (void)irq; (void)enabled; }
static inline uint32_t time_us_32(void) { return 1000000u; }
static inline bool watchdog_enable_caused_reboot(void) { return false; }
static inline void watchdog_enable(uint32_t ms, bool pause) { (void)ms; (void)pause; }
static inline void watchdog_update(void) {}
static inline void stdio_init_all(void) {}
static inline void sleep_ms(uint32_t ms) { (void)ms; }
static inline void tight_loop_contents(void) {}
static inline void pico_get_unique_board_id_string(char* buffer, size_t size) {
if (size) buffer[0] = '\0';
}
#ifdef __cplusplus
extern "C" {
#endif
void reset_usb_boot(uint32_t gpio_mask, uint32_t disable_mask);
int probe_debug_printf(const char* format, ...);
#ifdef __cplusplus
}
#endif

2
tests/native_hub_stubs/pico/bootrom.h Normal file
View file

@ -0,0 +1,2 @@
#pragma once
#include "hardware_stub.h"

2
tests/native_hub_stubs/pico/multicore.h Normal file
View file

@ -0,0 +1,2 @@
#pragma once
#include "hardware_stub.h"

2
tests/native_hub_stubs/pico/stdlib.h Normal file
View file

@ -0,0 +1,2 @@
#pragma once
#include "hardware_stub.h"

2
tests/native_hub_stubs/pico/unique_id.h Normal file
View file

@ -0,0 +1,2 @@
#pragma once
#include "hardware_stub.h"

45
tests/native_hub_stubs/tusb.h Normal file
View file

@ -0,0 +1,45 @@
#pragma once
#include <stdbool.h>
#include <stdint.h>
enum {
CONTROL_STAGE_SETUP = 0, CONTROL_STAGE_DATA = 1, CONTROL_STAGE_ACK = 2,
TUSB_REQ_RCPT_DEVICE = 0, TUSB_REQ_TYPE_VENDOR = 2,
TUSB_DIR_OUT = 0, TUSB_DIR_IN = 1,
TUSB_XFER_BULK = 2, TUSB_XFER_INTERRUPT = 3,
TUSB_REQ_GET_STATUS = 0, TUSB_REQ_CLEAR_FEATURE = 1, TUSB_REQ_SET_FEATURE = 3,
TUSB_REQ_SET_ADDRESS = 5, TUSB_REQ_GET_DESCRIPTOR = 6,
TUSB_REQ_GET_CONFIGURATION = 8, TUSB_REQ_SET_CONFIGURATION = 9,
TUSB_REQ_GET_INTERFACE = 10, TUSB_REQ_SET_INTERFACE = 11,
TUSB_DESC_DEVICE = 1, TUSB_DESC_CONFIGURATION = 2, TUSB_DESC_STRING = 3,
HID_REPORT_TYPE_OUTPUT = 2,
};
typedef uint8_t hid_report_type_t;
typedef struct __attribute__((packed)) {
union {
uint8_t bmRequestType;
struct __attribute__((packed)) {
uint8_t recipient : 5;
uint8_t type : 2;
uint8_t direction : 1;
} bmRequestType_bit;
};
uint8_t bRequest;
uint16_t wValue, wIndex, wLength;
} tusb_control_request_t;
#ifdef __cplusplus
extern "C" {
#endif
bool tud_control_xfer(uint8_t rhport, const tusb_control_request_t* request, void* buffer, uint16_t length);
bool tud_control_status(uint8_t rhport, const tusb_control_request_t* request);
bool tud_vendor_control_xfer_cb(uint8_t rhport, uint8_t stage, const tusb_control_request_t* request);
void tud_vendor_rx_cb(uint8_t instance, const uint8_t* buffer, uint16_t length);
void tud_vendor_tx_cb(uint8_t instance, uint32_t length);
const uint8_t* tud_hid_descriptor_report_cb(uint8_t instance);
uint16_t tud_hid_get_report_cb(uint8_t instance, uint8_t report_id, hid_report_type_t type, uint8_t* buffer, uint16_t length);
void tud_hid_set_report_cb(uint8_t instance, uint8_t report_id, hid_report_type_t type, const uint8_t* buffer, uint16_t length);
void tud_hid_report_complete_cb(uint8_t instance, const uint8_t* buffer, uint16_t length);
#ifdef __cplusplus
}
#endif

104
tests/native_hub_transport_fixture.c Normal file
View file

@ -0,0 +1,104 @@
#include "hardware_stub.h"
#include "usb/native_hub/native_hub.c"
usb_hw_t native_test_usb;
usb_device_dpram_t native_test_dpram;
sio_hw_t native_test_sio;
void probe_router_init(uint32_t hz) { (void)hz; }
void probe_router_core1(void) {}
void probe_router_publish(const uint8_t values[PROBE_ROUTER_SLOTS], uint8_t slot) { (void)values; (void)slot; }
void probe_router_enable(bool enabled) { (void)enabled; }
bool probe_router_set_phase(uint32_t phase) { (void)phase; return true; }
void probe_router_snapshot(probe_router_stats* snapshot) { memset(snapshot,0,sizeof(*snapshot)); snapshot->ready = 1; }
int probe_debug_printf(const char* format, ...) { (void)format; return 0; }
const uint8_t* native_joycon_device_descriptor(uint8_t instance) { (void)instance; return hub_device; }
const uint8_t* native_joycon_configuration_descriptor(uint8_t instance) { (void)instance; return hub_configuration; }
const uint16_t* native_joycon_string_descriptor(uint8_t instance, uint8_t index, uint16_t language) {
(void)instance; (void)language; return hub_string(index);
}
void native_joycon_usb_reset(uint8_t instance) { (void)instance; }
const uint8_t* tud_hid_descriptor_report_cb(uint8_t instance) { (void)instance; return NULL; }
uint16_t tud_hid_get_report_cb(uint8_t instance, uint8_t id, hid_report_type_t type, uint8_t* data, uint16_t length) {
(void)instance; (void)id; (void)type; (void)data; (void)length; return 0;
}
void tud_hid_set_report_cb(uint8_t instance, uint8_t id, hid_report_type_t type, const uint8_t* data, uint16_t length) {
(void)instance; (void)id; (void)type; (void)data; (void)length;
}
void tud_hid_report_complete_cb(uint8_t instance, const uint8_t* data, uint16_t length) { (void)instance; (void)data; (void)length; }
void tud_vendor_rx_cb(uint8_t instance, const uint8_t* data, uint16_t length) { (void)instance; (void)data; (void)length; }
void tud_vendor_tx_cb(uint8_t instance, uint32_t length) { (void)instance; (void)length; }
void native_test_initialize(void) {
memset(devices,0,sizeof(devices));
memset(ports,0,sizeof(ports));
memset(usb_hw,0,sizeof(*usb_hw));
memset(usb_dpram,0,sizeof(*usb_dpram));
event_head = event_tail = 0;
failed = bus_suspended = bank_restore_pending = false;
bank_lock = spin_lock_instance(0);
active_device = default_device = 0;
addresses[0] = 0; addresses[1] = 1; addresses[2] = 2;
started = root_configured_once = true;
}
static bool select_slot(uint8_t slot) {
if (!native_hub_select_device(addresses[slot],slot,UINT32_MAX / 2)) return false;
restore_selected_bank();
return true;
}
void native_test_drain(void) { native_hub_task(); }
bool native_test_setup(uint8_t slot, const tusb_control_request_t* request, bool drain) {
if (!select_slot(slot)) return false;
memcpy(usb_dpram->setup_packet,request,sizeof(*request));
usb_hw->sie_status = USB_SIE_STATUS_SETUP_REC_BITS;
usb_hw->ints = USB_INTS_SETUP_REQ_BITS;
usb_interrupt();
usb_hw->ints = 0;
if (drain) native_hub_task();
return !failed && devices[slot].control.stage != STALLED;
}
bool native_test_out(uint8_t slot, const uint8_t* data, uint16_t length, bool drain) {
if (!select_slot(slot)) return false;
uint32_t value = buffer_regs()[1];
if (!(value & USB_BUF_CTRL_AVAIL) || (value & USB_BUF_CTRL_STALL) ||
length > (value & USB_BUF_CTRL_LEN_MASK)) return false;
if (length) copy_to_usb(usb_dpram->ep0_buf_a,data,length);
buffer_regs()[1] = (value & ~(USB_BUF_CTRL_AVAIL | USB_BUF_CTRL_LEN_MASK)) | length;
usb_hw->buf_status = 2;
usb_hw->ints = USB_INTS_BUFF_STATUS_BITS;
usb_interrupt();
usb_hw->ints = 0;
if (drain) native_hub_task();
return !failed && devices[slot].control.stage != STALLED;
}
bool native_test_in(uint8_t slot, uint8_t* data, uint16_t* length, bool drain) {
if (!select_slot(slot)) return false;
uint32_t value = buffer_regs()[0];
if (!(value & USB_BUF_CTRL_AVAIL) || !(value & USB_BUF_CTRL_FULL) ||
(value & USB_BUF_CTRL_STALL)) return false;
*length = value & USB_BUF_CTRL_LEN_MASK;
if (*length) copy_from_usb(data,usb_dpram->ep0_buf_a,*length);
buffer_regs()[0] = value & ~USB_BUF_CTRL_AVAIL;
usb_hw->buf_status = 1;
usb_hw->ints = USB_INTS_BUFF_STATUS_BITS;
usb_interrupt();
usb_hw->ints = 0;
if (drain) native_hub_task();
return !failed && devices[slot].control.stage != STALLED;
}
void native_test_bus_reset(bool drain) {
usb_hw->sie_status = USB_SIE_STATUS_BUS_RESET_BITS;
usb_hw->ints = USB_INTS_BUS_RESET_BITS;
usb_interrupt();
usb_hw->ints = 0;
if (drain) native_hub_task();
// Assign fixture addresses after reset, independently of EP0 state.
addresses[0] = 0; addresses[1] = 1; addresses[2] = 2;
}

31
tests/switch2_native_gamepad_bridge_test.cpp
View file

@ -358,6 +358,36 @@ void wii_bias_and_independent_sensor_freshness() {
}
}
void nunchuk_buttons_map_to_native_left_shoulders() {
++source.controller.connection_generation;
source.controller.state = {};
source.accel_valid = source.gyro_valid = false;
profile = controller_profile_default(controller_identity_global(), 0);
// The real Wii parser maps Nunchuk C to west and Z to north. These are
// ordinary profile inputs, not the unrelated Switch2 extra "C" control.
profile.button_map[static_cast<unsigned>(ControllerProfileLogicalButton::kWest)] =
CONTROLLER_PROFILE_LEFT_TRIGGER_CONTROL;
profile.button_map[static_cast<unsigned>(ControllerProfileLogicalButton::kNorth)] =
static_cast<uint8_t>(ControllerProfileLogicalButton::kLeftShoulder);
source.controller.state.button_west = true; // C -> ZL.
publish(false); pair();
assert(reports[0][2] == 0 && reports[1][2] == 0x20);
source.controller.state.button_west = false;
source.controller.state.button_north = true; // Z -> L.
publish(false); pair();
assert(reports[0][2] == 0 && reports[1][2] == 0x10);
source.controller.state.button_right_shoulder = true; // Remote 2 -> R.
publish(false); pair();
assert(reports[0][2] == 0x10 && reports[1][2] == 0x10); // Real L+R across the pair.
source.controller.state.button_west = true;
publish(false); pair();
assert(reports[0][2] == 0x10 && reports[1][2] == 0x30);
source.controller.state = {};
publish(false); pair();
assert(reports[0][2] == 0 && reports[1][2] == 0); // No sticky synthetic chord.
no_mouse_or_rails();
}
} // namespace
int main() {
@ -372,5 +402,6 @@ int main() {
if (SWITCH2_BRIDGE_IMU_TARGET_MASK == 3) real_motion_admission_and_loss();
selected_motion_target_keeps_both_control_halves();
wii_bias_and_independent_sensor_freshness();
nunchuk_buttons_map_to_native_left_shoulders();
return 0;
}

610
tests/test_config_manager.py
View file

@ -41,6 +41,7 @@ class FakeDevice:
bus = 1
address = 7
port_numbers = (1,)
firmware_version = (0, 2, 0)
def __init__(self) -> None:
self.configuration = struct.pack(
@ -300,9 +301,7 @@ class FakeDevice:
request,
bytes(
[
0,
2,
0,
*self.firmware_version,
2,
self.active_mode,
self.capabilities,
@ -376,7 +375,9 @@ class FakeDevice:
make_response(
request,
self.profiles[self.selected_profile],
schema=struct.unpack_from("<H", self.profiles[self.selected_profile])[0],
schema=struct.unpack_from(
"<H", self.profiles[self.selected_profile]
)[0],
generation=self.profile_generation,
)
)
@ -718,7 +719,11 @@ def native_rumble_configuration(
) -> bytes:
return (
struct.pack(
"<HBBB3x", 90, config_manager.REQUESTED_MODE_XINPUT, len(identities), joycon_mode
"<HBBB3x",
90,
config_manager.REQUESTED_MODE_XINPUT,
len(identities),
joycon_mode,
)
+ b"".join(identity.to_bytes() for identity in identities)
+ bytes((16 - len(identities)) * 14)
@ -770,7 +775,11 @@ def test_native_rumble_configuration_canonical_wire_round_trip(schema: int) -> N
config_manager.write_configuration(
device,
config_manager.AdapterConfiguration(
90, 0, 0, config_manager.REQUESTED_MODE_XINPUT, tuple(reversed(identities)),
90,
0,
0,
config_manager.REQUESTED_MODE_XINPUT,
tuple(reversed(identities)),
schema_version=schema,
),
1.0,
@ -830,7 +839,8 @@ def test_native_rumble_configuration_rejects_invalid_lists(malformation: str) ->
@pytest.mark.parametrize(
"malformation", ("duplicate", "overflow", "global", "ble", "pair", "vendor", "product")
"malformation",
("duplicate", "overflow", "global", "ble", "pair", "vendor", "product"),
)
def test_native_rumble_write_rejects_invalid_approvals_before_transaction(
malformation: str,
@ -865,11 +875,16 @@ def test_native_rumble_write_rejects_invalid_approvals_before_transaction(
@pytest.mark.parametrize(
("schema", "joycon_mode"),
((3, config_manager.JOYCON_MODE_PAIRED), (4, config_manager.JOYCON_MODE_INDIVIDUAL)),
(
(3, config_manager.JOYCON_MODE_PAIRED),
(4, config_manager.JOYCON_MODE_INDIVIDUAL),
),
)
def test_native_rumble_cli_approval_is_physical_and_preserves_other_settings(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],
schema: int, joycon_mode: int,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
schema: int,
joycon_mode: int,
) -> None:
device = FakeDevice()
device.configuration_schema = schema
@ -1038,8 +1053,10 @@ def test_configuration_transaction_and_reset() -> None:
),
)
def test_joycon_mode_legacy_read_and_write_refusal(
schema: int, payload: bytes,
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],
schema: int,
payload: bytes,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
device = FakeDevice()
device.configuration_schema = schema
@ -1047,7 +1064,9 @@ def test_joycon_mode_legacy_read_and_write_refusal(
monkeypatch.setattr(config_manager, "_candidate_devices", lambda: (device,))
assert config_manager.main(["joycon-mode", "--json"]) == 0
assert json.loads(capsys.readouterr().out) == {
"mode": "paired", "generation": 3, "supported": False,
"mode": "paired",
"generation": 3,
"supported": False,
}
configuration = config_manager.read_configuration(device)
assert configuration.joycon_mode == config_manager.JOYCON_MODE_PAIRED
@ -1058,7 +1077,8 @@ def test_joycon_mode_legacy_read_and_write_refusal(
assert capsys.readouterr().out == ""
with pytest.raises(config_manager.ConfigManagerError):
config_manager.write_configuration(
device, replace(configuration, joycon_mode=config_manager.JOYCON_MODE_INDIVIDUAL),
device,
replace(configuration, joycon_mode=config_manager.JOYCON_MODE_INDIVIDUAL),
1.0,
)
assert device.configuration == payload
@ -1097,12 +1117,15 @@ def test_joycon_mode_rejects_invalid_values_before_transaction(mode: object) ->
with pytest.raises(config_manager.ConfigManagerError):
config_manager.set_joycon_mode(device, mode, 1.0)
with pytest.raises(config_manager.ConfigManagerError):
config_manager.write_configuration(device, replace(before, joycon_mode=mode), 1.0)
config_manager.write_configuration(
device, replace(before, joycon_mode=mode), 1.0
)
assert not device.out_requests
def test_joycon_mode_cli_commits_and_reads_without_reboot_or_profile_changes(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
device = FakeDevice()
device.configuration_schema = 4
@ -1140,7 +1163,8 @@ def test_joycon_mode_cli_commits_and_reads_without_reboot_or_profile_changes(
def test_joycon_mode_cli_commit_failure_does_not_claim_new_preference(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
device = FakeDevice()
device.configuration_schema = 4
@ -1627,19 +1651,53 @@ def test_requested_and_active_mode_response_validation() -> None:
config_manager.read_info(device)
def test_native_hub_reports_capabilities_but_refuses_mode_changes() -> None:
device = FakeDevice()
device.firmware_version = (0, 72, 0)
device.active_mode = 5
device.capabilities = 7
before = config_manager.read_configuration(device)
info = config_manager.read_info(device)
assert info.firmware_version == (0, 72, 0)
assert info.mode_name() == "Native Joy-Con hub"
assert info.capability_names() == ("input", "rumble", "motion")
with pytest.raises(config_manager.ConfigManagerError, match="fixed USB output"):
config_manager.configure_mode(device, before.requested_mode, 1.0)
with pytest.raises(config_manager.ConfigManagerError, match="fixed USB output"):
config_manager.set_mode(device, config_manager.REQUESTED_MODE_SWITCH, 1.0)
with pytest.raises(config_manager.ConfigManagerError, match="fixed USB output"):
config_manager.request_reboot(device, 1)
assert config_manager.read_configuration(device) == before
assert device.out_requests == []
config_manager.request_bootsel_reboot(device)
assert device.bootsel_reboot_requested
@pytest.mark.parametrize(
("left_type", "right_type", "flags"), [(0, 0, 1), (1, 0, 3), (0, 1, 5), (1, 1, 7)]
)
def test_joycon_pair_wire_round_trip_preserves_both_typed_members(
left_type: int, right_type: int, flags: int,
left_type: int,
right_type: int,
flags: int,
) -> None:
left = config_manager.ControllerIdentity(
True, config_manager.TRANSPORT_BLE, left_type,
bytes.fromhex("C10203040506"), 0x057E, 0x2067,
True,
config_manager.TRANSPORT_BLE,
left_type,
bytes.fromhex("C10203040506"),
0x057E,
0x2067,
)
right = config_manager.ControllerIdentity(
True, config_manager.TRANSPORT_BLE, right_type,
bytes.fromhex("D11213141516"), 0x057E, 0x2066,
True,
config_manager.TRANSPORT_BLE,
right_type,
bytes.fromhex("D11213141516"),
0x057E,
0x2066,
)
pair = config_manager.ControllerIdentity.make_joycon_pair(left, right)
wire = bytes((flags, 3)) + left.address + right.address
@ -1653,34 +1711,51 @@ def test_joycon_pair_wire_round_trip_preserves_both_typed_members(
def test_joycon_pair_distinguishes_address_types_without_normalizing_members() -> None:
left = config_manager.ControllerIdentity(
True, config_manager.TRANSPORT_BLE, 0,
bytes.fromhex("C10203040506"), 0x057E, 0x2067,
True,
config_manager.TRANSPORT_BLE,
0,
bytes.fromhex("C10203040506"),
0x057E,
0x2067,
)
right = replace(left, address_type=1, product_id=0x2066)
pair = config_manager.ControllerIdentity.make_joycon_pair(left, right)
assert pair.to_bytes() == bytes.fromhex("0503C10203040506C10203040506")
assert pair.joycon_pair_members() == (left, right)
with pytest.raises(config_manager.ConfigManagerError):
config_manager.ControllerIdentity.make_joycon_pair(left, replace(right, address_type=0))
config_manager.ControllerIdentity.make_joycon_pair(
left, replace(right, address_type=0)
)
with pytest.raises(config_manager.ConfigManagerError):
config_manager.ControllerIdentity.make_joycon_pair(right, left)
@pytest.mark.parametrize("wire", [
"0003102030405060C12233445566", # Stable flag missing.
"0D03102030405060C12233445566", # Reserved flag.
"0303102030405060C12233445566", # Left random address is not static.
"0503C12233445566102030405060", # Right random address is not static.
"0103102030405060102030405060", # Duplicate typed members.
])
@pytest.mark.parametrize(
"wire",
[
"0003102030405060C12233445566", # Stable flag missing.
"0D03102030405060C12233445566", # Reserved flag.
"0303102030405060C12233445566", # Left random address is not static.
"0503C12233445566102030405060", # Right random address is not static.
"0103102030405060102030405060", # Duplicate typed members.
],
)
def test_joycon_pair_rejects_malformed_wire(wire: str) -> None:
with pytest.raises(config_manager.ConfigManagerError):
config_manager.ControllerIdentity.from_bytes(bytes.fromhex(wire))
@pytest.mark.parametrize("malformation", [
"global", "classic", "vendor", "model", "address_type", "random_address",
])
@pytest.mark.parametrize(
"malformation",
[
"global",
"classic",
"vendor",
"model",
"address_type",
"random_address",
],
)
def test_joycon_pair_requires_legitimate_ble_members(malformation: str) -> None:
pair = config_manager.ControllerIdentity.from_bytes(
bytes.fromhex("0503102030405060C12233445566")
@ -1698,14 +1773,19 @@ def test_joycon_pair_requires_legitimate_ble_members(malformation: str) -> None:
config_manager.ControllerIdentity.make_joycon_pair(left, invalid)
@pytest.mark.parametrize("fields", [
{"stable": False},
{"vendor_id": 0x045E},
{"product_id": 0x2066},
{"address_type": 2},
{"partner_address_type": 3},
])
def test_joycon_pair_rejects_inconsistent_in_memory_identity(fields: dict[str, object]) -> None:
@pytest.mark.parametrize(
"fields",
[
{"stable": False},
{"vendor_id": 0x045E},
{"product_id": 0x2066},
{"address_type": 2},
{"partner_address_type": 3},
],
)
def test_joycon_pair_rejects_inconsistent_in_memory_identity(
fields: dict[str, object],
) -> None:
pair = config_manager.ControllerIdentity.from_bytes(
bytes.fromhex("0503102030405060C12233445566")
)
@ -1713,10 +1793,13 @@ def test_joycon_pair_rejects_inconsistent_in_memory_identity(fields: dict[str, o
replace(pair, **fields)
@pytest.mark.parametrize("fields", [
{"partner_address_type": 1},
{"partner_address": bytes.fromhex("C12233445566")},
])
@pytest.mark.parametrize(
"fields",
[
{"partner_address_type": 1},
{"partner_address": bytes.fromhex("C12233445566")},
],
)
def test_physical_identity_cannot_hide_pair_members(fields: dict[str, object]) -> None:
with pytest.raises(config_manager.ConfigManagerError):
replace(native_rumble_identity(), **fields)
@ -1724,13 +1807,16 @@ def test_physical_identity_cannot_hide_pair_members(fields: dict[str, object]) -
def test_pairing_inventory_rejects_logical_profile_owners() -> None:
device = FakeDevice()
device.records = [(config_manager.TRANSPORT_JOYCON_PAIR, 0, bytes.fromhex("102030405060"))]
device.records = [
(config_manager.TRANSPORT_JOYCON_PAIR, 0, bytes.fromhex("102030405060"))
]
with pytest.raises(config_manager.ConfigManagerError):
config_manager.read_pairings(device)
def test_profile_cli_lists_both_pair_addresses_but_native_inventory_omits_pair(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
device = FakeDevice()
pair = config_manager.ControllerIdentity.from_bytes(
@ -1870,8 +1956,14 @@ def test_identity_and_profile_binary_json_round_trip() -> None:
legacy_json_object["schema_version"] = config_manager.PROFILE_LEGACY_SCHEMA_VERSION
legacy_json_object["size"] = config_manager.PROFILE_LEGACY_SIZE
for field in (
"shortcuts", "shift", "turbo_settings", "extra_button_map", "swing",
"nunchuk_swing", "combined_swing", "combination_window_ms",
"shortcuts",
"shift",
"turbo_settings",
"extra_button_map",
"swing",
"nunchuk_swing",
"combined_swing",
"combination_window_ms",
):
del legacy_json_object[field]
del legacy_json_object["motion_toggle_chord"]
@ -1936,8 +2028,14 @@ def test_schema5_full_macro_stream_migrates_bytes_and_json(monkeypatch) -> None:
obj["schema_version"] = 5
obj["size"] = 256
for field in (
"shortcuts", "shift", "turbo_settings", "extra_button_map", "swing",
"nunchuk_swing", "combined_swing", "combination_window_ms",
"shortcuts",
"shift",
"turbo_settings",
"extra_button_map",
"swing",
"nunchuk_swing",
"combined_swing",
"combination_window_ms",
):
del obj[field]
for macro in obj["macros"]:
@ -1997,31 +2095,53 @@ def test_set_b_sparse_settings_and_macro_modes_round_trip() -> None:
device.profiles[(device.stable_identity.to_bytes(), 1)] = bytes(legacy_wire)
assert config_manager.read_profile(device, device.stable_identity, 1) == profile
device.profile_aliases[device.stable_identity.to_bytes()] = "Custom controller"
old_listing = config_manager.parse_profile_list(config_manager.parse_response(
make_response(config_manager.OP_PROFILE_LIST, device._profile_list_payload(), schema=6),
config_manager.OP_PROFILE_LIST,
))
assert old_listing[1] == config_manager.ProfileListEntry(device.stable_identity, 1, "Custom controller")
old_listing = config_manager.parse_profile_list(
config_manager.parse_response(
make_response(
config_manager.OP_PROFILE_LIST, device._profile_list_payload(), schema=6
),
config_manager.OP_PROFILE_LIST,
)
)
assert old_listing[1] == config_manager.ProfileListEntry(
device.stable_identity, 1, "Custom controller"
)
@pytest.mark.parametrize("version", [7, 8, 9])
def test_schema7_extra_controls_keep_output_channels_and_wire_layout(version: int) -> None:
def test_schema7_extra_controls_keep_output_channels_and_wire_layout(
version: int,
) -> None:
obj = custom_profile().to_json_object()
obj["extra_button_map"] = dict(zip(
config_manager.EXTRA_BUTTONS,
("south", "right_trigger", "left_trigger", None, "dpad_left", "start", "capture"),
))
obj["extra_button_map"] = dict(
zip(
config_manager.EXTRA_BUTTONS,
(
"south",
"right_trigger",
"left_trigger",
None,
"dpad_left",
"start",
"capture",
),
)
)
obj["shift"]["mode"] = "hold"
obj["shift"]["modifier"] = "gl"
obj["shift"]["extra_button_map"] = dict(zip(
config_manager.EXTRA_BUTTONS,
("east", None, "west", "north", "system", "dpad_up", "dpad_right"),
))
obj["shift"]["extra_button_map"] = dict(
zip(
config_manager.EXTRA_BUTTONS,
("east", None, "west", "north", "system", "dpad_up", "dpad_right"),
)
)
obj["shortcuts"]["modifier"] = "left_sr"
obj["shortcuts"]["profiles"][0] = "south"
obj["switching_chord"] = ["left_trigger", "c", "right_sr"]
obj["motion_toggle_chord"] = ["right_trigger", "gl", "left_sl"]
for index, names in enumerate((["c", "gl"], ["gr"], ["left_sl", "left_sr"], ["right_sl", "right_sr"])):
for index, names in enumerate(
(["c", "gl"], ["gr"], ["left_sl", "left_sr"], ["right_sl", "right_sr"])
):
obj["macros"][index]["trigger"] = names
obj["macros"][index]["cancel"] = config_manager.EXTRA_BUTTONS[index + 3]
obj["schema_version"] = version
@ -2040,7 +2160,12 @@ def test_schema7_extra_controls_keep_output_channels_and_wire_layout(version: in
assert encoded[358:364] == bytes((3, 4, 24, 96, 65, 10))
assert config_manager.ControllerProfile.from_bytes(encoded) == profile
assert config_manager.ControllerProfile.from_json(profile.to_json()) == profile
assert len(profile.button_map) == len(profile.shift.button_map) == len(profile.turbo_modes) == 16
assert (
len(profile.button_map)
== len(profile.shift.button_map)
== len(profile.turbo_modes)
== 16
)
assert profile.swing.button == config_manager.PROFILE_NONE_BUTTON
legacy_wire = bytearray(encoded)
struct.pack_into("<H", legacy_wire, 0, version)
@ -2051,20 +2176,29 @@ def test_schema7_extra_controls_keep_output_channels_and_wire_layout(version: in
device = FakeDevice()
device.profiles[(device.stable_identity.to_bytes(), 1)] = bytes(legacy_wire)
assert config_manager.read_profile(device, device.stable_identity, 1) == profile
listing = config_manager.parse_profile_list(config_manager.parse_response(
make_response(config_manager.OP_PROFILE_LIST, device._profile_list_payload(), schema=version),
config_manager.OP_PROFILE_LIST,
))
listing = config_manager.parse_profile_list(
config_manager.parse_response(
make_response(
config_manager.OP_PROFILE_LIST,
device._profile_list_payload(),
schema=version,
),
config_manager.OP_PROFILE_LIST,
)
)
assert listing[1].identity == device.stable_identity
assert listing[1].active_profile_index == 1
@pytest.mark.parametrize("path", [
("button_map", "south"),
("extra_button_map", "c"),
("triggers", "left", "output"),
("shift", "extra_button_map", "c"),
])
@pytest.mark.parametrize(
"path",
[
("button_map", "south"),
("extra_button_map", "c"),
("triggers", "left", "output"),
("shift", "extra_button_map", "c"),
],
)
def test_extra_controls_cannot_be_output_destinations(path: tuple[str, ...]) -> None:
obj = config_manager.ControllerProfile.default().to_json_object()
target = obj
@ -2075,10 +2209,20 @@ def test_extra_controls_cannot_be_output_destinations(path: tuple[str, ...]) ->
config_manager.ControllerProfile.from_json_object(obj)
@pytest.mark.parametrize("offset,value", [
(98, 18 << 2), (256, 18), (266, 24), (344, 1), (358, 1), (363, 1),
])
def test_schema6_rejects_schema7_controls_in_old_fields(offset: int, value: int) -> None:
@pytest.mark.parametrize(
"offset,value",
[
(98, 18 << 2),
(256, 18),
(266, 24),
(344, 1),
(358, 1),
(363, 1),
],
)
def test_schema6_rejects_schema7_controls_in_old_fields(
offset: int, value: int
) -> None:
payload = bytearray(config_manager.ControllerProfile.default().to_bytes())
struct.pack_into("<H", payload, 0, 6)
payload[344:] = bytes(40)
@ -2087,9 +2231,17 @@ def test_schema6_rejects_schema7_controls_in_old_fields(offset: int, value: int)
config_manager.ControllerProfile.from_bytes(payload)
@pytest.mark.parametrize("offset,value", [
(344, 18), (351, 16), (358, 128), (362, 128), (363, 128), (364, 1),
])
@pytest.mark.parametrize(
"offset,value",
[
(344, 18),
(351, 16),
(358, 128),
(362, 128),
(363, 128),
(364, 1),
],
)
def test_schema7_rejects_extra_map_and_mask_overflow(offset: int, value: int) -> None:
payload = bytearray(config_manager.ControllerProfile.default().to_bytes())
struct.pack_into("<H", payload, 0, 7)
@ -2129,27 +2281,40 @@ def test_legacy_control_profiles_preserve_custom_actions(version: int) -> None:
macro["steps"].append(config_manager.MacroStep.end().to_json_object())
obj["macro"] = macro
for key in (
"shortcuts", "shift", "turbo_settings", "extra_button_map", "swing",
"nunchuk_swing", "combined_swing", "combination_window_ms",
"shortcuts",
"shift",
"turbo_settings",
"extra_button_map",
"swing",
"nunchuk_swing",
"combined_swing",
"combination_window_ms",
):
del obj[key]
assert config_manager.ControllerProfile.from_json_object(obj) == profile
assert config_manager.ControllerProfile.from_bytes(profile.to_bytes()) == profile
@pytest.mark.parametrize(("button", "sensitivity", "modifier"), [
("south", "low", "right_sr"),
("dpad_right", "high", "south"),
("west", "medium", "right_trigger"),
(None, "high", None),
])
@pytest.mark.parametrize(
("button", "sensitivity", "modifier"),
[
("south", "low", "right_sr"),
("dpad_right", "high", "south"),
("west", "medium", "right_trigger"),
(None, "high", None),
],
)
def test_swing_profile_round_trip_preserves_other_settings(
button: str | None, sensitivity: str, modifier: str | None,
button: str | None,
sensitivity: str,
modifier: str | None,
) -> None:
before = custom_profile()
obj = before.to_json_object()
obj["swing"] = {
"button": button, "sensitivity": sensitivity, "modifier": modifier,
"button": button,
"sensitivity": sensitivity,
"modifier": modifier,
"macro": None,
}
profile = config_manager.ControllerProfile.from_json_object(obj)
@ -2162,17 +2327,23 @@ def test_swing_profile_round_trip_preserves_other_settings(
assert config_manager.read_profile(device, device.stable_identity, 2) == profile
@pytest.mark.parametrize(("field", "value"), [
("button", "left_trigger"),
("button", "right_sr"),
("button", 0),
("sensitivity", "extreme"),
("sensitivity", None),
("sensitivity", 1),
("modifier", "unknown"),
("modifier", False),
("macro", 0), ("macro", 5), ("macro", True), ("macro", "1"),
])
@pytest.mark.parametrize(
("field", "value"),
[
("button", "left_trigger"),
("button", "right_sr"),
("button", 0),
("sensitivity", "extreme"),
("sensitivity", None),
("sensitivity", 1),
("modifier", "unknown"),
("modifier", False),
("macro", 0),
("macro", 5),
("macro", True),
("macro", "1"),
],
)
def test_swing_rejects_invalid_json_settings(field: str, value: object) -> None:
obj = config_manager.ControllerProfile.default().to_json_object()
obj["swing"][field] = value
@ -2180,22 +2351,47 @@ def test_swing_rejects_invalid_json_settings(field: str, value: object) -> None:
config_manager.ControllerProfile.from_json_object(obj)
@pytest.mark.parametrize(("field", "value"), [
("button", -1), ("button", 16), ("button", True),
("sensitivity", -1), ("sensitivity", 3), ("sensitivity", False),
("modifier", -1), ("modifier", 25), ("modifier", 256),
("macro", -1), ("macro", 4), ("macro", False),
])
@pytest.mark.parametrize(
("field", "value"),
[
("button", -1),
("button", 16),
("button", True),
("sensitivity", -1),
("sensitivity", 3),
("sensitivity", False),
("modifier", -1),
("modifier", 25),
("modifier", 256),
("macro", -1),
("macro", 4),
("macro", False),
],
)
def test_swing_rejects_invalid_in_memory_settings(field: str, value: object) -> None:
with pytest.raises(config_manager.ConfigManagerError):
replace(config_manager.ProfileSwing(), **{field: value})
@pytest.mark.parametrize(("offset", "value"), [
(364, 16), (365, 3), (366, 25), (367, 4),
(368, 16), (369, 3), (370, 25), (371, 4),
(372, 16), (373, 4), (374, 25), (375, 29), (375, 201), (376, 1),
])
@pytest.mark.parametrize(
("offset", "value"),
[
(364, 16),
(365, 3),
(366, 25),
(367, 4),
(368, 16),
(369, 3),
(370, 25),
(371, 4),
(372, 16),
(373, 4),
(374, 25),
(375, 29),
(375, 201),
(376, 1),
],
)
def test_swing_rejects_corrupt_wire_settings(offset: int, value: int) -> None:
payload = bytearray(config_manager.ControllerProfile.default().to_bytes())
payload[offset] = value
@ -2203,11 +2399,18 @@ def test_swing_rejects_corrupt_wire_settings(offset: int, value: int) -> None:
config_manager.ControllerProfile.from_bytes(payload)
@pytest.mark.parametrize("mutation", [
"missing", "unknown", "missing_button", "missing_sensitivity", "missing_modifier",
"missing_macro",
"legacy_field",
])
@pytest.mark.parametrize(
"mutation",
[
"missing",
"unknown",
"missing_button",
"missing_sensitivity",
"missing_modifier",
"missing_macro",
"legacy_field",
],
)
def test_swing_json_fields_are_strict(mutation: str) -> None:
obj = config_manager.ControllerProfile.default().to_json_object()
if mutation == "missing":
@ -2246,7 +2449,9 @@ def test_schema8_swing_migration_preserves_remote_binding() -> None:
def test_gesture_macro_json_indices_and_wire_layout(macro_number: int) -> None:
obj = custom_profile().to_json_object()
obj["macros"][macro_number - 1] = {
**obj["macros"][0], "trigger": [], "playback": "toggle",
**obj["macros"][0],
"trigger": [],
"playback": "toggle",
}
obj["swing"].update(macro=macro_number, sensitivity="high", modifier="right_sr")
obj["nunchuk_swing"].update(button="dpad_right", sensitivity="low", modifier="c")
@ -2255,10 +2460,22 @@ def test_gesture_macro_json_indices_and_wire_layout(macro_number: int) -> None:
profile = config_manager.ControllerProfile.from_json_object(obj)
assert profile.swing.macro == profile.combined_swing.macro == macro_number - 1
encoded = profile.to_bytes()
assert encoded[364:376] == bytes((
255, 2, 24, macro_number - 1, 15, 0, 18, 255,
255, macro_number - 1, 16, 200,
))
assert encoded[364:376] == bytes(
(
255,
2,
24,
macro_number - 1,
15,
0,
18,
255,
255,
macro_number - 1,
16,
200,
)
)
assert encoded[376:] == bytes(8)
assert config_manager.ControllerProfile.from_bytes(encoded) == profile
assert profile.to_json_object() == obj
@ -2271,14 +2488,17 @@ def test_gesture_actions_are_exclusive(gesture: str) -> None:
with pytest.raises(config_manager.ConfigManagerError):
config_manager.ControllerProfile.from_json_object(obj)
action_type = (
config_manager.ProfileCombinedSwing if gesture == "combined_swing"
config_manager.ProfileCombinedSwing
if gesture == "combined_swing"
else config_manager.ProfileSwing
)
with pytest.raises(config_manager.ConfigManagerError):
action_type(button=0, macro=0)
payload = bytearray(custom_profile().to_bytes())
button_offset, macro_offset = {
"swing": (364, 367), "nunchuk_swing": (368, 371), "combined_swing": (372, 373),
"swing": (364, 367),
"nunchuk_swing": (368, 371),
"combined_swing": (372, 373),
}[gesture]
payload[button_offset] = payload[macro_offset] = 0
with pytest.raises(config_manager.ConfigManagerError):
@ -2288,7 +2508,8 @@ def test_gesture_actions_are_exclusive(gesture: str) -> None:
@pytest.mark.parametrize("gesture", ["swing", "nunchuk_swing", "combined_swing"])
@pytest.mark.parametrize("empty", [False, True])
def test_gesture_macro_requires_configured_positive_duration_target(
gesture: str, empty: bool,
gesture: str,
empty: bool,
) -> None:
obj = custom_profile().to_json_object()
target = obj["macros"][0]
@ -2315,17 +2536,26 @@ def test_combination_window_inclusive_bounds_round_trip(window: int) -> None:
obj = config_manager.ControllerProfile.default().to_json_object()
obj["combination_window_ms"] = window
profile = config_manager.ControllerProfile.from_json_object(obj)
assert config_manager.ControllerProfile.from_bytes(profile.to_bytes()).combination_window_ms == window
assert (
config_manager.ControllerProfile.from_bytes(
profile.to_bytes()
).combination_window_ms
== window
)
@pytest.mark.parametrize("window", [29, 201, True, 100.0, None])
def test_combination_window_rejects_out_of_range_or_noninteger_values(window: object) -> None:
def test_combination_window_rejects_out_of_range_or_noninteger_values(
window: object,
) -> None:
obj = config_manager.ControllerProfile.default().to_json_object()
obj["combination_window_ms"] = window
with pytest.raises(config_manager.ConfigManagerError):
config_manager.ControllerProfile.from_json_object(obj)
with pytest.raises(config_manager.ConfigManagerError):
replace(config_manager.ControllerProfile.default(), combination_window_ms=window)
replace(
config_manager.ControllerProfile.default(), combination_window_ms=window
)
@pytest.mark.parametrize(
@ -2610,20 +2840,31 @@ def test_profile_playtest_decodes_raw_controller_state() -> None:
def test_playtest_layout_and_extra_inputs_preserve_legacy_firmware() -> None:
device = FakeDevice()
device.stable_identity = replace(device.stable_identity, vendor_id=0x057E, product_id=0x2067)
device.stable_identity = replace(
device.stable_identity, vendor_id=0x057E, product_id=0x2067
)
device.playtest_extra_buttons = 0x55
device.playtest_layout = 3
payload, flags = device._profile_playtest_payload()
def parse(data: bytes, schema: int) -> config_manager.ProfilePlaytest:
return config_manager.parse_profile_playtest(config_manager.parse_response(
make_response(config_manager.OP_PROFILE_PLAYTEST, data, flags=flags, schema=schema),
config_manager.OP_PROFILE_PLAYTEST,
))
return config_manager.parse_profile_playtest(
config_manager.parse_response(
make_response(
config_manager.OP_PROFILE_PLAYTEST, data, flags=flags, schema=schema
),
config_manager.OP_PROFILE_PLAYTEST,
)
)
current = parse(payload, 4)
assert current.to_json_object()["layout"] == "joycon2-pair"
assert current.to_json_object()["extra_buttons"] == ["c", "gr", "left_sr", "right_sr"]
assert current.to_json_object()["extra_buttons"] == [
"c",
"gr",
"left_sr",
"right_sr",
]
assert current.to_json_object()["buttons"] == ["south", "dpad_up", "dpad_right"]
assert parse(payload[:55], 3) == replace(current, layout=None)
assert parse(payload[:54], 2) == replace(current, extra_buttons=0, layout=None)
@ -3085,31 +3326,80 @@ def test_mode_parser_accepts_all_implemented_modes() -> None:
assert args.mode == mode
def test_candidate_discovery_checks_all_usb_identities(
monkeypatch: pytest.MonkeyPatch,
) -> None:
lookups: list[tuple[int, int]] = []
assert config_manager.USB_IDENTITIES == (
@pytest.mark.parametrize(
"identity",
[
(0x057E, 0x2009),
(0x057E, 0x2068),
(0xCAFE, 0x4010),
(0xCAFE, 0x4020),
(0xCAFE, 0x4021),
)
],
)
def test_discovery_finds_one_adapter_without_its_native_children(
monkeypatch: pytest.MonkeyPatch,
identity: tuple[int, int],
) -> None:
device = FakeDevice()
if identity == (0x057E, 0x2068):
device.firmware_version = (0, 72, 0)
device.active_mode = 5
right = FakeDevice()
right.address = 8
left = FakeDevice()
left.address = 9
devices = {
identity: (device,),
(0x057E, 0x2066): (right,),
(0x057E, 0x2067): (left,),
}
def find(**arguments: object) -> tuple[object, ...]:
lookups.append(
(
int(arguments["idVendor"]),
int(arguments["idProduct"]),
)
def find(**arguments: object) -> tuple[FakeDevice, ...]:
return devices.get(
(int(arguments["idVendor"]), int(arguments["idProduct"])), ()
)
monkeypatch.setattr(config_manager.usb.core, "find", find)
assert config_manager.find_pico(None, None, timeout=0) is device
for child in (right, left):
with pytest.raises(config_manager.ConfigManagerError, match="no USB-connected"):
config_manager.find_pico(child.bus, child.address, timeout=0)
@pytest.mark.parametrize(
"response",
[
config_manager.usb.core.USBError("management request stalled", errno=32),
b"Nintendo",
make_response(config_manager.OP_INFO, b""),
make_response(config_manager.OP_INFO, bytes((0, 72, 0, 2, 5, 0x80, 0, 2))),
],
)
def test_native_discovery_requires_validated_management_info(
monkeypatch: pytest.MonkeyPatch,
response: bytes | Exception,
) -> None:
class NintendoDevice(FakeDevice):
address = 8
def ctrl_transfer(self, *args: object, **kwargs: object) -> bytes:
if isinstance(response, Exception):
raise response
return response
nintendo = NintendoDevice()
hub = FakeDevice()
hub.active_mode = 5
def find(**arguments: object) -> tuple[FakeDevice, ...]:
if (arguments["idVendor"], arguments["idProduct"]) == (0x057E, 0x2068):
return nintendo, hub
return ()
monkeypatch.setattr(config_manager.usb.core, "find", find)
assert list(config_manager._candidate_devices()) == []
assert tuple(lookups) == config_manager.USB_IDENTITIES
assert config_manager.find_pico(None, None, timeout=0) is hub
with pytest.raises(config_manager.ConfigManagerError, match="none accepted"):
config_manager.find_pico(nintendo.bus, nintendo.address, timeout=0)
def test_find_requires_selector_for_multiple_picos(
@ -4241,7 +4531,9 @@ def test_haptics_fixture_metadata_follows_reported_frame_count(
)
pattern = snapshot.to_json_object()["pattern"]
assert pattern["stereo_frames_per_packet"] == packet_frames
assert pattern["packet_interval_us"] == pytest.approx(packet_frames * 1000000 / 3000)
assert pattern["packet_interval_us"] == pytest.approx(
packet_frames * 1000000 / 3000
)
assert pattern["total_packets"] == total_packets
assert pattern["priming_silence_packets"] == silence_packets
assert pattern["trailing_silence_packets"] == silence_packets
@ -4261,7 +4553,9 @@ def test_haptics_fixture_metadata_follows_reported_frame_count(
@pytest.mark.parametrize("mode", [0, 1])
@pytest.mark.parametrize("packet_frames", [0, 48])
def test_haptics_rejects_unadvertised_frame_sizes(mode: int, packet_frames: int) -> None:
def test_haptics_rejects_unadvertised_frame_sizes(
mode: int, packet_frames: int
) -> None:
device = HapticsDevice(
[haptics_response(2, slot=0, mode=mode, packet_frames=packet_frames)]
)

59
tests/test_native_hub_management_native.py Normal file
View file

@ -0,0 +1,59 @@
import shutil
import subprocess
from pathlib import Path
def test_native_hub_management_native(tmp_path: Path) -> None:
root = Path(__file__).resolve().parents[1]
cc = shutil.which("cc") or shutil.which("gcc")
cxx = shutil.which("c++") or shutil.which("g++")
assert cc is not None and cxx is not None, "host C and C++ compilers are required"
includes = [
f"-I{root / 'tests' / 'native_hub_stubs'}",
f"-I{root / 'tests' / 'bluepad32_native_stubs'}",
f"-I{root / 'src' / 'firmware'}",
f"-I{root / 'tools' / 'pico_usb_address_probe'}",
f"-I{root / 'tools' / 'switch2_usb_probe'}",
]
flags = ["-Wall", "-Wextra", "-Werror", "-pedantic", "-DSWITCH2_PROBE_HUB=1"]
transport = tmp_path / "native_hub_transport.o"
executable = tmp_path / "native_hub_management_test"
subprocess.run(
[
cc,
"-std=c11",
*flags,
*includes,
"-c",
str(root / "tests" / "native_hub_transport_fixture.c"),
"-o",
str(transport),
],
check=True,
cwd=root,
)
sources = [
"tests/native_hub_management_test.cpp",
"src/firmware/usb/usb_configuration_management.cpp",
"tools/switch2_usb_probe/bootsel.cpp",
"src/firmware/configuration/adapter_configuration.cpp",
"src/firmware/core/controller_identity.cpp",
"src/firmware/profile/controller_profile.cpp",
"src/firmware/profile/profile_storage.cpp",
"src/firmware/profile/profile_service.cpp",
]
subprocess.run(
[
cxx,
"-std=c++17",
*flags,
*includes,
*(str(root / path) for path in sources),
str(transport),
"-o",
str(executable),
],
check=True,
cwd=root,
)
subprocess.run([str(executable)], check=True, cwd=root)

356
tests/test_profile_web.py
View file

@ -18,7 +18,7 @@ from tests.test_config_manager import FakeDevice, custom_profile, native_rumble_
@contextmanager
def running_server(
monkeypatch: pytest.MonkeyPatch, device: FakeDevice
monkeypatch: pytest.MonkeyPatch, device: FakeDevice | None
) -> Iterator[tuple[str, str]]:
server = profile_web.ProfileEditorServer(
("127.0.0.1", 0),
@ -26,7 +26,8 @@ def running_server(
device_address=None,
timeout=1.0,
)
monkeypatch.setattr(server, "find_device", lambda: device)
if device is not None:
monkeypatch.setattr(server, "find_device", lambda: device)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
@ -118,13 +119,80 @@ def test_editor_serves_assets_and_complete_schema(
)
@pytest.mark.parametrize(("product_id", "extras"), [
(0x2069, {"c", "gl", "gr"}),
(0x2067, {"left_sl", "left_sr"}),
(0x2066, {"c", "right_sl", "right_sr"}),
])
def test_native_hub_editor_discovers_root_and_preserves_saved_state(
monkeypatch: pytest.MonkeyPatch,
) -> None:
device = FakeDevice()
device.firmware_version = (0, 72, 0)
device.active_mode = 5
device.capabilities = 7
device.profiles[(device.global_identity.to_bytes(), 2)] = (
custom_profile().to_bytes()
)
configuration = config_manager.read_configuration(device)
pairings = config_manager.read_pairings(device)
profiles = device.profiles.copy()
active_profiles = device.active_profiles.copy()
def find(**arguments: object) -> tuple[FakeDevice, ...]:
if (arguments["idVendor"], arguments["idProduct"]) == (0x057E, 0x2068):
return (device,)
return ()
monkeypatch.setattr(config_manager.usb.core, "find", find)
with running_server(monkeypatch, None) as (base_url, token):
status, listing = request_json(f"{base_url}/api/profiles")
assert status == 200
assert [entry["key"] for entry in listing["identities"]] == [
identity.to_bytes().hex() for identity in device.profile_identities
]
status, selected = request_json(f"{base_url}/api/profiles/1/8")
assert status == 200
assert (
selected["profile"]
== config_manager.ControllerProfile.default().to_json_object()
)
draft = custom_profile().to_json_object()
status, _ = request_json(
f"{base_url}/api/profiles/1/8", method="PUT", value=draft, token=token
)
assert status == 200
status, _ = request_json(
f"{base_url}/api/profiles/1/8/activate", method="POST", token=token
)
assert status == 200
status, stored = request_json(f"{base_url}/api/profiles/1/8")
assert status == 200
assert stored["profile"] == draft
assert stored["active"] is True
profiles[(device.stable_identity.to_bytes(), 7)] = custom_profile().to_bytes()
active_profiles[device.stable_identity.to_bytes()] = 7
assert device.profiles == profiles
assert device.active_profiles == active_profiles
assert config_manager.read_configuration(device) == configuration
assert config_manager.read_pairings(device) == pairings
assert device.active_mode == 5
assert not {
config_manager.OP_MODE_SET,
config_manager.OP_REBOOT,
config_manager.OP_BOOTSEL_REBOOT,
}.intersection(device.requests)
@pytest.mark.parametrize(
("product_id", "extras"),
[
(0x2069, {"c", "gl", "gr"}),
(0x2067, {"left_sl", "left_sr"}),
(0x2066, {"c", "right_sl", "right_sr"}),
],
)
def test_switch2_input_choices_are_never_output_targets(
monkeypatch: pytest.MonkeyPatch, product_id: int, extras: set[str],
monkeypatch: pytest.MonkeyPatch,
product_id: int,
extras: set[str],
) -> None:
device = FakeDevice()
identity = replace(device.stable_identity, vendor_id=0x057E, product_id=product_id)
@ -145,7 +213,8 @@ def test_switch2_input_choices_are_never_output_targets(
@pytest.mark.parametrize("version", [6, 7, 8])
def test_editor_migrates_old_profiles_and_saves_swing_without_metadata_loss(
monkeypatch: pytest.MonkeyPatch, version: int,
monkeypatch: pytest.MonkeyPatch,
version: int,
) -> None:
device = FakeDevice()
key = (device.stable_identity.to_bytes(), 1)
@ -167,10 +236,15 @@ def test_editor_migrates_old_profiles_and_saves_swing_without_metadata_loss(
with running_server(monkeypatch, device) as (base_url, token):
status, migrated = request_json(f"{base_url}/api/profiles/1/2")
assert status == 200
assert config_manager.ControllerProfile.from_json_object(migrated["profile"]) == profile
assert (
config_manager.ControllerProfile.from_json_object(migrated["profile"])
== profile
)
assert migrated["profile"]["swing"] == profile.swing.to_json_object()
draft = migrated["profile"]
draft["extra_button_map"] = dict(zip(config_manager.EXTRA_BUTTONS, config_manager.LOGICAL_BUTTONS[:7]))
draft["extra_button_map"] = dict(
zip(config_manager.EXTRA_BUTTONS, config_manager.LOGICAL_BUTTONS[:7])
)
draft["shift"]["mode"] = "hold"
draft["shift"]["modifier"] = "c"
draft["shift"]["extra_button_map"]["right_sr"] = "dpad_right"
@ -179,26 +253,40 @@ def test_editor_migrates_old_profiles_and_saves_swing_without_metadata_loss(
draft["switching_chord"] = ["left_sl", "left_sr"]
draft["motion_toggle_chord"] = ["right_sl", "c"]
draft["swing"] = {
"button": "west", "sensitivity": "high", "modifier": "right_trigger",
"button": "west",
"sensitivity": "high",
"modifier": "right_trigger",
"macro": None,
}
draft["nunchuk_swing"].update(macro=1, sensitivity="low", modifier="c")
draft["combined_swing"].update(macro=2, modifier="left_trigger")
draft["combination_window_ms"] = 30
status, validated = request_json(
f"{base_url}/api/profiles/validate", method="POST", value=draft, token=token,
f"{base_url}/api/profiles/validate",
method="POST",
value=draft,
token=token,
)
assert status == 200
expected_profile = config_manager.ControllerProfile.from_json_object(draft)
assert config_manager.ControllerProfile.from_json_object(validated["profile"]) == expected_profile
assert (
config_manager.ControllerProfile.from_json_object(validated["profile"])
== expected_profile
)
assert device.profiles[key] == bytes(legacy_wire)
status, _ = request_json(
f"{base_url}/api/profiles/1/2", method="PUT", value=draft, token=token,
f"{base_url}/api/profiles/1/2",
method="PUT",
value=draft,
token=token,
)
assert status == 200
status, stored = request_json(f"{base_url}/api/profiles/1/2")
assert status == 200
assert config_manager.ControllerProfile.from_json_object(stored["profile"]) == expected_profile
assert (
config_manager.ControllerProfile.from_json_object(stored["profile"])
== expected_profile
)
assert stored["alias"] == "Living room"
assert stored["name"] == "Racing"
assert stored["active"] is True
@ -210,7 +298,8 @@ def test_editor_migrates_old_profiles_and_saves_swing_without_metadata_loss(
@pytest.mark.parametrize("gesture", ["swing", "nunchuk_swing", "combined_swing"])
def test_editor_rejects_unconfigured_gesture_macro_without_saving(
monkeypatch: pytest.MonkeyPatch, gesture: str,
monkeypatch: pytest.MonkeyPatch,
gesture: str,
) -> None:
device = FakeDevice()
key = (device.stable_identity.to_bytes(), 1)
@ -219,10 +308,14 @@ def test_editor_rejects_unconfigured_gesture_macro_without_saving(
draft[gesture]["macro"] = 4
with running_server(monkeypatch, device) as (base_url, token):
for endpoint, method in (
("/api/profiles/validate", "POST"), ("/api/profiles/1/2", "PUT"),
("/api/profiles/validate", "POST"),
("/api/profiles/1/2", "PUT"),
):
status, result = request_json(
f"{base_url}{endpoint}", method=method, value=draft, token=token,
f"{base_url}{endpoint}",
method=method,
value=draft,
token=token,
)
assert status == 400
assert "error" in result
@ -267,8 +360,10 @@ def test_editor_identifies_connected_controller_artwork(
(identity["controller"]["style"], identity["controller"]["layout"])
for identity in listing["identities"]
] == [
("generic", "generic"), ("switch", "switch-pro"),
("playstation", "dualsense"), ("xbox", "xbox"),
("generic", "generic"),
("switch", "switch-pro"),
("playstation", "dualsense"),
("xbox", "xbox"),
]
assert [identity["key"] for identity in listing["identities"]] == [
identity.to_bytes().hex() for identity in device.profile_identities
@ -285,7 +380,8 @@ def joycon_pair_device() -> FakeDevice:
device.profile_identities = [device.global_identity, left, right, pair]
device.stable_identity = pair
device.active_profiles = {
identity.to_bytes(): index for index, identity in enumerate(device.profile_identities)
identity.to_bytes(): index
for index, identity in enumerate(device.profile_identities)
}
default_profile = config_manager.ControllerProfile.default().to_bytes()
device.profiles = {
@ -299,7 +395,8 @@ def joycon_pair_device() -> FakeDevice:
def test_persistent_pair_owner_is_offline_capable_and_isolated_from_solo_banks(
monkeypatch: pytest.MonkeyPatch, joycon_pair_device: FakeDevice,
monkeypatch: pytest.MonkeyPatch,
joycon_pair_device: FakeDevice,
) -> None:
device = joycon_pair_device
_, left, right, pair = device.profile_identities
@ -312,7 +409,9 @@ def test_persistent_pair_owner_is_offline_capable_and_isolated_from_solo_banks(
identity.to_bytes().hex() for identity in device.profile_identities
]
assert [owner["controller"]["layout"] for owner in owners[1:]] == [
"joycon2-left", "joycon2-right", "joycon2-pair",
"joycon2-left",
"joycon2-right",
"joycon2-pair",
]
owner = owners[3]
assert owner["controller"]["model"] != owners[1]["controller"]["model"]
@ -322,7 +421,11 @@ def test_persistent_pair_owner_is_offline_capable_and_isolated_from_solo_banks(
assert owner["identity"]["members"]["right"]["address"] == right.address_text
assert owner["identity"]["members"]["right"]["address_type"] == 1
assert set(owner["source_controls"]) & set(config_manager.EXTRA_BUTTONS) == {
"c", "left_sl", "left_sr", "right_sl", "right_sr",
"c",
"left_sl",
"left_sr",
"right_sl",
"right_sr",
}
status, offline = request_json(f"{base_url}/api/profiles/3/8/playtest")
assert status == 200
@ -337,8 +440,10 @@ def test_persistent_pair_owner_is_offline_capable_and_isolated_from_solo_banks(
for bank, output in ((3, "north"), (1, "east"), (2, "west")):
expected[bank]["button_map"]["south"] = output
status, _ = request_json(
f"{base_url}/api/profiles/{bank}/8", method="PUT",
value=expected[bank], token=token,
f"{base_url}/api/profiles/{bank}/8",
method="PUT",
value=expected[bank],
token=token,
)
assert status == 200
for other_bank, expected_profile in expected.items():
@ -350,16 +455,26 @@ def test_persistent_pair_owner_is_offline_capable_and_isolated_from_solo_banks(
("identities/3/alias", "Couch pair"),
):
status, _ = request_json(
f"{base_url}/api/{path}", method="PUT", value={"value": value}, token=token,
f"{base_url}/api/{path}",
method="PUT",
value={"value": value},
token=token,
)
assert status == 200
status, _ = request_json(
f"{base_url}/api/profiles/3/8/activate", method="POST", token=token,
f"{base_url}/api/profiles/3/8/activate",
method="POST",
token=token,
)
assert status == 200
status, listing = request_json(f"{base_url}/api/profiles")
assert status == 200
assert [owner["active_profile"] for owner in listing["identities"]] == [1, 2, 3, 8]
assert [owner["active_profile"] for owner in listing["identities"]] == [
1,
2,
3,
8,
]
assert listing["identities"][3]["label"] == "Couch pair"
assert listing["identities"][3]["key"] == pair.to_bytes().hex()
for bank in (1, 2, 3):
@ -388,7 +503,8 @@ def test_persistent_pair_owner_is_offline_capable_and_isolated_from_solo_banks(
def test_capture_cannot_bind_pair_input_to_a_solo_owner(
monkeypatch: pytest.MonkeyPatch, joycon_pair_device: FakeDevice,
monkeypatch: pytest.MonkeyPatch,
joycon_pair_device: FakeDevice,
) -> None:
device = joycon_pair_device
left = device.profile_identities[1]
@ -396,7 +512,9 @@ def test_capture_cannot_bind_pair_input_to_a_solo_owner(
status, _ = request_json(f"{base_url}/api/profiles/1/1/playtest")
assert status == 200
status, _ = request_json(
f"{base_url}/api/profiles/1/1/capture/start", method="POST", token=token,
f"{base_url}/api/profiles/1/1/capture/start",
method="POST",
token=token,
value={
"owner_key": left.to_bytes().hex(),
"capture_id": "solo-bank-paired-input",
@ -416,37 +534,56 @@ def test_capture_cannot_bind_pair_input_to_a_solo_owner(
@pytest.mark.parametrize("owner_index", [0, 1])
def test_live_layout_transitions_do_not_infer_topology_from_profile_owner(
monkeypatch: pytest.MonkeyPatch, owner_index: int,
monkeypatch: pytest.MonkeyPatch,
owner_index: int,
) -> None:
device = FakeDevice()
left = replace(device.stable_identity, vendor_id=0x057E, product_id=0x2067)
device.stable_identity = left
device.profile_identities = [device.global_identity, left]
device.active_profiles[left.to_bytes()] = 0
device.profiles[(left.to_bytes(), 0)] = config_manager.ControllerProfile.default().to_bytes()
device.profiles[(left.to_bytes(), 0)] = (
config_manager.ControllerProfile.default().to_bytes()
)
device.playtest_motion = None
with running_server(monkeypatch, device) as (base_url, _):
for code, expected in ((1, "joycon2-left"), (3, "joycon2-pair"), (1, "joycon2-left")):
for code, expected in (
(1, "joycon2-left"),
(3, "joycon2-pair"),
(1, "joycon2-left"),
):
device.playtest_layout = code
status, sample = request_json(f"{base_url}/api/profiles/{owner_index}/1/playtest")
status, sample = request_json(
f"{base_url}/api/profiles/{owner_index}/1/playtest"
)
assert status == 200
assert sample["controller"]["layout"] == expected
assert sample["owner_key"] == device.profile_identities[owner_index].to_bytes().hex()
assert (
sample["owner_key"]
== device.profile_identities[owner_index].to_bytes().hex()
)
assert sample["identity_key"] == left.to_bytes().hex()
assert sample["identity"]["is_joycon_pair"] is False
expected_extras = (
{"c", "left_sl", "left_sr", "right_sl", "right_sr"}
if code == 3 else {"left_sl", "left_sr"}
if code == 3
else {"left_sl", "left_sr"}
)
assert (
set(sample["source_controls"]) & set(config_manager.EXTRA_BUTTONS)
== expected_extras
)
assert set(sample["source_controls"]) & set(config_manager.EXTRA_BUTTONS) == expected_extras
status, listing = request_json(f"{base_url}/api/profiles")
assert status == 200
assert [owner["key"] for owner in listing["identities"]] == [
device.global_identity.to_bytes().hex(), left.to_bytes().hex(),
device.global_identity.to_bytes().hex(),
left.to_bytes().hex(),
]
assert listing["identities"][1]["controller"]["layout"] == "joycon2-left"
device.playtest_connected = False
status, offline = request_json(f"{base_url}/api/profiles/{owner_index}/1/playtest")
status, offline = request_json(
f"{base_url}/api/profiles/{owner_index}/1/playtest"
)
assert status == 200
assert offline["connected"] is False
assert offline["layout"] is None
@ -468,7 +605,9 @@ def test_wii_pid_does_not_claim_a_remote_or_extension_without_live_metadata(
monkeypatch: pytest.MonkeyPatch,
) -> None:
device = FakeDevice()
device.stable_identity = replace(device.stable_identity, vendor_id=0x057E, product_id=0x0330)
device.stable_identity = replace(
device.stable_identity, vendor_id=0x057E, product_id=0x0330
)
device.profile_identities = [device.global_identity, device.stable_identity]
device.active_profiles[device.stable_identity.to_bytes()] = 0
with running_server(monkeypatch, device) as (base_url, _):
@ -488,9 +627,12 @@ def test_wii_pid_does_not_claim_a_remote_or_extension_without_live_metadata(
assert status == 200
assert sample["controller"]["layout"] == expected
if code:
assert set(sample["source_controls"]).isdisjoint(config_manager.EXTRA_BUTTONS)
assert set(sample["source_controls"]).isdisjoint(
config_manager.EXTRA_BUTTONS
)
assert ("left_shoulder" in sample["source_controls"]) == (code == 5)
def test_editor_reads_writes_and_activates_profiles_atomically(
monkeypatch: pytest.MonkeyPatch,
) -> None:
@ -651,13 +793,17 @@ def test_editor_rejects_invalid_or_unauthorized_mutations(
def test_joycon_mode_preserves_configuration_and_all_profile_banks(
monkeypatch: pytest.MonkeyPatch, joycon_pair_device: FakeDevice,
monkeypatch: pytest.MonkeyPatch,
joycon_pair_device: FakeDevice,
) -> None:
device = joycon_pair_device
config_manager.write_configuration(
device,
config_manager.AdapterConfiguration(
135, 0, 0, config_manager.REQUESTED_MODE_XINPUT,
135,
0,
0,
config_manager.REQUESTED_MODE_XINPUT,
(native_rumble_identity(),),
),
1.0,
@ -677,24 +823,35 @@ def test_joycon_mode_preserves_configuration_and_all_profile_banks(
status, initial = request_json(f"{base_url}/api/joycon-mode")
assert status == 200
assert initial == {
"mode": "paired", "generation": before.generation, "supported": True,
"mode": "paired",
"generation": before.generation,
"supported": True,
}
for mode in ("individual", "paired"):
status, committed = request_json(
f"{base_url}/api/joycon-mode", method="PUT",
value={"mode": mode}, token=token,
f"{base_url}/api/joycon-mode",
method="PUT",
value={"mode": mode},
token=token,
)
assert status == 200
stored = config_manager.read_configuration(device)
assert committed == {
"mode": mode, "generation": stored.generation, "supported": True,
"mode": mode,
"generation": stored.generation,
"supported": True,
}
assert stored.joycon_mode == config_manager.JOYCON_MODE_NAMES.index(mode)
assert stored.generation > before.generation
assert replace(
stored, joycon_mode=before.joycon_mode,
generation=before.generation, crc=before.crc,
) == before
assert (
replace(
stored,
joycon_mode=before.joycon_mode,
generation=before.generation,
crc=before.crc,
)
== before
)
assert request_json(f"{base_url}/api/joycon-mode") == (200, committed)
assert device.profiles == profiles
assert device.profile_names == names
@ -706,14 +863,19 @@ def test_joycon_mode_preserves_configuration_and_all_profile_banks(
@pytest.mark.parametrize("schema", [1, 2, 3])
def test_joycon_mode_legacy_reads_default_and_refuses_mutations(
monkeypatch: pytest.MonkeyPatch, schema: int,
monkeypatch: pytest.MonkeyPatch,
schema: int,
) -> None:
device = FakeDevice()
config_manager.write_configuration(
device,
config_manager.AdapterConfiguration(
95, 0, 0,
native_switch_controllers=(native_rumble_identity(),) if schema == 3 else (),
95,
0,
0,
native_switch_controllers=(native_rumble_identity(),)
if schema == 3
else (),
schema_version=schema,
),
1.0,
@ -723,12 +885,16 @@ def test_joycon_mode_legacy_reads_default_and_refuses_mutations(
status, legacy = request_json(f"{base_url}/api/joycon-mode")
assert status == 200
assert legacy == {
"mode": "paired", "generation": before.generation, "supported": False,
"mode": "paired",
"generation": before.generation,
"supported": False,
}
for mode in ("paired", "individual"):
status, _ = request_json(
f"{base_url}/api/joycon-mode", method="PUT",
value={"mode": mode}, token=token,
f"{base_url}/api/joycon-mode",
method="PUT",
value={"mode": mode},
token=token,
)
assert status == 400
assert config_manager.read_configuration(device) == before
@ -739,29 +905,48 @@ def test_joycon_mode_rejects_unauthorized_and_malformed_requests(
) -> None:
device = FakeDevice()
config_manager.write_configuration(
device, config_manager.AdapterConfiguration(110, 0, 0), 1.0,
device,
config_manager.AdapterConfiguration(110, 0, 0),
1.0,
)
before = config_manager.read_configuration(device)
with running_server(monkeypatch, device) as (base_url, token):
endpoint = f"{base_url}/api/joycon-mode"
for supplied_token in (None, "invalid"):
status, _ = request_json(
endpoint, method="PUT", value={"mode": "individual"},
endpoint,
method="PUT",
value={"mode": "individual"},
token=supplied_token,
)
assert status == 403
for body in (
{}, [], {"mode": 1}, {"mode": True}, {"mode": None},
{"mode": ["individual"]}, {"mode": "Individual"},
{},
[],
{"mode": 1},
{"mode": True},
{"mode": None},
{"mode": ["individual"]},
{"mode": "Individual"},
{"mode": "individual", "pairing_window_seconds": 10},
):
status, _ = request_json(
endpoint, method="PUT", value=body, token=token,
endpoint,
method="PUT",
value=body,
token=token,
)
assert status == 400
for raw in (b"", b"{", b"\xff", b" " * (profile_web._MAXIMUM_REQUEST_BYTES + 1)):
for raw in (
b"",
b"{",
b"\xff",
b" " * (profile_web._MAXIMUM_REQUEST_BYTES + 1),
):
request = urllib.request.Request(
endpoint, data=raw, method="PUT",
endpoint,
data=raw,
method="PUT",
headers={"X-Switch-Pico-Token": token},
)
with pytest.raises(urllib.error.HTTPError) as rejected:
@ -769,7 +954,8 @@ def test_joycon_mode_rejects_unauthorized_and_malformed_requests(
assert rejected.value.code == 400
for method in ("GET", "PUT"):
request = urllib.request.Request(
endpoint, method=method,
endpoint,
method=method,
data=b'{"mode":"individual"}' if method == "PUT" else None,
headers={"Host": "untrusted.example", "X-Switch-Pico-Token": token},
)
@ -784,12 +970,16 @@ def test_joycon_mode_commit_is_not_reported_as_confirmed_when_readback_fails(
) -> None:
device = FakeDevice()
config_manager.write_configuration(
device, config_manager.AdapterConfiguration(110, 0, 0), 1.0,
device,
config_manager.AdapterConfiguration(110, 0, 0),
1.0,
)
read_configuration = config_manager.read_configuration
before = read_configuration(device)
def disconnect_after_commit(device: config_manager.UsbDevice) -> config_manager.AdapterConfiguration:
def disconnect_after_commit(
device: config_manager.UsbDevice,
) -> config_manager.AdapterConfiguration:
configuration = read_configuration(device)
if configuration.generation != before.generation:
raise usb.core.USBError("device disconnected")
@ -798,18 +988,23 @@ def test_joycon_mode_commit_is_not_reported_as_confirmed_when_readback_fails(
monkeypatch.setattr(config_manager, "read_configuration", disconnect_after_commit)
with running_server(monkeypatch, device) as (base_url, token):
status, response = request_json(
f"{base_url}/api/joycon-mode", method="PUT",
value={"mode": "individual"}, token=token,
f"{base_url}/api/joycon-mode",
method="PUT",
value={"mode": "individual"},
token=token,
)
assert status == 503
assert "error" in response
assert "mode" not in response
assert read_configuration(device).joycon_mode == config_manager.JOYCON_MODE_INDIVIDUAL
assert (
read_configuration(device).joycon_mode == config_manager.JOYCON_MODE_INDIVIDUAL
)
@pytest.mark.parametrize("method", ["GET", "PUT"])
def test_joycon_mode_usb_failure_returns_service_unavailable(
monkeypatch: pytest.MonkeyPatch, method: str,
monkeypatch: pytest.MonkeyPatch,
method: str,
) -> None:
device = FakeDevice()
before = device.configuration
@ -820,8 +1015,10 @@ def test_joycon_mode_usb_failure_returns_service_unavailable(
monkeypatch.setattr(device, "ctrl_transfer", disconnected)
with running_server(monkeypatch, device) as (base_url, token):
status, response = request_json(
f"{base_url}/api/joycon-mode", method=method,
value={"mode": "individual"} if method == "PUT" else None, token=token,
f"{base_url}/api/joycon-mode",
method=method,
value={"mode": "individual"} if method == "PUT" else None,
token=token,
)
assert status == 503
assert "error" in response
@ -906,7 +1103,6 @@ def test_recorder_accepts_first_connection_generation_zero(
assert status == 200 and stopped["state_name"] == "stopped"
def test_wii_orientation_endpoint_validation_and_failures(
monkeypatch: pytest.MonkeyPatch,
) -> None:
@ -951,9 +1147,7 @@ def test_wii_orientation_endpoint_validation_and_failures(
def failing_config(*args: Any, **kwargs: Any) -> None:
raise config_manager.ConfigManagerError("firmware rejected orientation")
monkeypatch.setattr(
config_manager, "set_wii_orientation", failing_config
)
monkeypatch.setattr(config_manager, "set_wii_orientation", failing_config)
status, err = request_json(
f"{base_url}/api/identities/1/wii-orientation",
method="POST",
@ -967,9 +1161,7 @@ def test_wii_orientation_endpoint_validation_and_failures(
def usb_failing_config(*args: Any, **kwargs: Any) -> None:
raise usb.core.USBError("USB pipe error")
monkeypatch.setattr(
config_manager, "set_wii_orientation", usb_failing_config
)
monkeypatch.setattr(config_manager, "set_wii_orientation", usb_failing_config)
status, err = request_json(
f"{base_url}/api/identities/1/wii-orientation",
method="POST",