From 229de6d686dc6f3eeaa11ae992023a3a7e4fa6d5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=94=D0=B0=D0=BD=D0=B8=D0=B8=D0=BB=20=D0=93=D1=80=D0=B0?= =?UTF-8?q?=D0=B1=D0=B0=D1=80=D1=8C?= Date: Sun, 13 Sep 2026 10:07:19 +0300 Subject: [PATCH] new control robot --- .../include/iiwa_controller/CommandGuard.hpp | 89 +++++++ .../include/iiwa_controller/StateGuard.hpp | 99 ++++++++ .../include/iiwa_controller/filters.hpp | 18 ++ src/iiwa_controller/test/test_guards.cpp | 120 +++++++++ .../test/test_safety_contract.py | 34 +++ src/iiwa_planning/test/conftest.py | 232 ++++++++++++++++++ .../test/test_motion_sequence_runner.py | 73 ++++++ .../test/test_move_to_pose_server.py | 206 ++++++++++++++++ src/iiwa_web/test/conftest.py | 198 +++++++++++++++ src/iiwa_web/test/test_ros_action_timeout.py | 65 +++++ .../test/test_runner_upload_security.py | 130 ++++++++++ .../test/test_trajectory_validation.py | 118 +++++++++ src/iiwa_web/test/test_web_auth.py | 137 +++++++++++ 13 files changed, 1519 insertions(+) create mode 100644 src/iiwa_controller/include/iiwa_controller/CommandGuard.hpp create mode 100644 src/iiwa_controller/include/iiwa_controller/StateGuard.hpp create mode 100644 src/iiwa_controller/include/iiwa_controller/filters.hpp create mode 100644 src/iiwa_controller/test/test_guards.cpp create mode 100644 src/iiwa_controller/test/test_safety_contract.py create mode 100644 src/iiwa_planning/test/conftest.py create mode 100644 src/iiwa_planning/test/test_motion_sequence_runner.py create mode 100644 src/iiwa_planning/test/test_move_to_pose_server.py create mode 100644 src/iiwa_web/test/conftest.py create mode 100644 src/iiwa_web/test/test_ros_action_timeout.py create mode 100644 src/iiwa_web/test/test_runner_upload_security.py create mode 100644 src/iiwa_web/test/test_trajectory_validation.py create mode 100644 src/iiwa_web/test/test_web_auth.py diff --git a/src/iiwa_controller/include/iiwa_controller/CommandGuard.hpp b/src/iiwa_controller/include/iiwa_controller/CommandGuard.hpp new file mode 100644 index 0000000..2d1bc97 --- /dev/null +++ b/src/iiwa_controller/include/iiwa_controller/CommandGuard.hpp @@ -0,0 +1,89 @@ +#pragma once + +#include +#include +#include +#include + +namespace iiwa_controller +{ + +constexpr std::size_t COMMAND_GUARD_JOINTS = 7; + +struct CommandGuardLimits +{ + std::array min_position{}; + std::array max_position{}; + std::array max_velocity{}; +}; + +enum class CommandGuardResult +{ + OK, + INVALID_PERIOD, + NON_FINITE, + POSITION_LIMIT, + VELOCITY_LIMIT, +}; + +inline const char * commandGuardResultName(const CommandGuardResult result) +{ + switch (result) { + case CommandGuardResult::OK: return "OK"; + case CommandGuardResult::INVALID_PERIOD: return "INVALID_PERIOD"; + case CommandGuardResult::NON_FINITE: return "NON_FINITE"; + case CommandGuardResult::POSITION_LIMIT: return "POSITION_LIMIT"; + case CommandGuardResult::VELOCITY_LIMIT: return "VELOCITY_LIMIT"; + default: return "UNKNOWN"; + } +} + +class CommandGuard +{ +public: + CommandGuard() + { + limits_.min_position.fill(-std::numeric_limits::infinity()); + limits_.max_position.fill(std::numeric_limits::infinity()); + limits_.max_velocity.fill(std::numeric_limits::infinity()); + } + + explicit CommandGuard(const CommandGuardLimits & limits) + : limits_(limits) {} + + CommandGuardResult validate( + const std::array & command, + const std::array & previous_command, + const double sample_time, + const bool previous_command_initialized) const + { + if (!std::isfinite(sample_time) || sample_time <= 0.0) { + return CommandGuardResult::INVALID_PERIOD; + } + + for (std::size_t i = 0; i < COMMAND_GUARD_JOINTS; ++i) { + if (!std::isfinite(command[i])) { + return CommandGuardResult::NON_FINITE; + } + if (command[i] < limits_.min_position[i] || command[i] > limits_.max_position[i]) { + return CommandGuardResult::POSITION_LIMIT; + } + if (previous_command_initialized) { + if (!std::isfinite(previous_command[i]) || !std::isfinite(limits_.max_velocity[i])) { + return CommandGuardResult::NON_FINITE; + } + const double max_delta = limits_.max_velocity[i] * sample_time; + if (std::abs(command[i] - previous_command[i]) > max_delta + 1e-9) { + return CommandGuardResult::VELOCITY_LIMIT; + } + } + } + + return CommandGuardResult::OK; + } + +private: + CommandGuardLimits limits_{}; +}; + +} // namespace iiwa_controller diff --git a/src/iiwa_controller/include/iiwa_controller/StateGuard.hpp b/src/iiwa_controller/include/iiwa_controller/StateGuard.hpp new file mode 100644 index 0000000..793d65a --- /dev/null +++ b/src/iiwa_controller/include/iiwa_controller/StateGuard.hpp @@ -0,0 +1,99 @@ +#pragma once + +#include +#include + +#include "iiwa_controller/FRIClient.h" + +namespace iiwa_controller +{ + +enum class StateGuardResult +{ + OK, + INVALID_SNAPSHOT, + SESSION_STATE, + CONNECTION_QUALITY, + SAFETY_STOP, + OPERATION_MODE, + DRIVE_STATE, + CLIENT_COMMAND_MODE, + CONTROL_MODE, + TRACKING_PERFORMANCE, +}; + +inline const char * stateGuardResultName(const StateGuardResult result) +{ + switch (result) { + case StateGuardResult::OK: return "OK"; + case StateGuardResult::INVALID_SNAPSHOT: return "INVALID_SNAPSHOT"; + case StateGuardResult::SESSION_STATE: return "SESSION_STATE"; + case StateGuardResult::CONNECTION_QUALITY: return "CONNECTION_QUALITY"; + case StateGuardResult::SAFETY_STOP: return "SAFETY_STOP"; + case StateGuardResult::OPERATION_MODE: return "OPERATION_MODE"; + case StateGuardResult::DRIVE_STATE: return "DRIVE_STATE"; + case StateGuardResult::CLIENT_COMMAND_MODE: return "CLIENT_COMMAND_MODE"; + case StateGuardResult::CONTROL_MODE: return "CONTROL_MODE"; + case StateGuardResult::TRACKING_PERFORMANCE: return "TRACKING_PERFORMANCE"; + default: return "UNKNOWN"; + } +} + +class StateGuard +{ +public: + StateGuardResult validateSnapshot(const IIWAStateSnapshot & snapshot) const + { + for (std::size_t i = 0; i < FRIClient::N_JOINTS; ++i) { + if (!std::isfinite(snapshot.measured_pos[i]) || + !std::isfinite(snapshot.measured_tau[i]) || + !std::isfinite(snapshot.external_tau[i])) { + return StateGuardResult::INVALID_SNAPSHOT; + } + } + if (!std::isfinite(snapshot.sample_time) || snapshot.sample_time <= 0.0 || + snapshot.time_stamp_nano_sec >= 1000000000U || + !std::isfinite(snapshot.tracking_performance)) { + return StateGuardResult::INVALID_SNAPSHOT; + } + return StateGuardResult::OK; + } + + StateGuardResult validatePositionState( + const IIWAStateSnapshot & snapshot, + const KUKA::FRI::ESessionState session_state) const + { + auto result = validateSnapshot(snapshot); + if (result != StateGuardResult::OK) { + return result; + } + if (session_state != KUKA::FRI::COMMANDING_ACTIVE) { + return StateGuardResult::SESSION_STATE; + } + if (snapshot.quality < KUKA::FRI::GOOD) { + return StateGuardResult::CONNECTION_QUALITY; + } + if (snapshot.safety_state != KUKA::FRI::NORMAL_OPERATION) { + return StateGuardResult::SAFETY_STOP; + } + if (snapshot.operation_mode != KUKA::FRI::AUTOMATIC_MODE) { + return StateGuardResult::OPERATION_MODE; + } + if (snapshot.drive_state != KUKA::FRI::ACTIVE) { + return StateGuardResult::DRIVE_STATE; + } + if (snapshot.client_command_mode != KUKA::FRI::POSITION) { + return StateGuardResult::CLIENT_COMMAND_MODE; + } + if (snapshot.control_mode != KUKA::FRI::POSITION_CONTROL_MODE && + snapshot.control_mode != KUKA::FRI::JOINT_IMP_CONTROL_MODE) { + return StateGuardResult::CONTROL_MODE; + } + if (snapshot.tracking_performance < 0.5) { + return StateGuardResult::TRACKING_PERFORMANCE; + } + return StateGuardResult::OK; + } +}; + +} // namespace iiwa_controller diff --git a/src/iiwa_controller/include/iiwa_controller/filters.hpp b/src/iiwa_controller/include/iiwa_controller/filters.hpp new file mode 100644 index 0000000..4325635 --- /dev/null +++ b/src/iiwa_controller/include/iiwa_controller/filters.hpp @@ -0,0 +1,18 @@ +#pragma once + +#include + +namespace iiwa_controller +{ + +// Continuous-time first-order low-pass filter discretization used by +// lbr_fri_ros2_stack. tau == 0 intentionally disables smoothing. +inline double exponentialFilterAlpha(const double tau, const double sample_time) +{ + if (tau <= 0.0) { + return 1.0; + } + return 1.0 - std::exp(-sample_time / tau); +} + +} // namespace iiwa_controller diff --git a/src/iiwa_controller/test/test_guards.cpp b/src/iiwa_controller/test/test_guards.cpp new file mode 100644 index 0000000..597bb31 --- /dev/null +++ b/src/iiwa_controller/test/test_guards.cpp @@ -0,0 +1,120 @@ +#include +#include +#include + +#include + +#include "iiwa_controller/CommandGuard.hpp" +#include "iiwa_controller/FRIClient.h" +#include "iiwa_controller/StateGuard.hpp" +#include "iiwa_controller/filters.hpp" + +namespace iiwa_controller::test +{ + +TEST(ExponentialFilter, UsesContinuousTimeAlpha) +{ + EXPECT_NEAR(exponentialFilterAlpha(0.04, 0.01), 1.0 - std::exp(-0.25), 1e-12); + EXPECT_DOUBLE_EQ(exponentialFilterAlpha(0.0, 0.01), 1.0); +} + +CommandGuard makeGuard() +{ + CommandGuardLimits limits; + limits.min_position.fill(-1.0); + limits.max_position.fill(1.0); + limits.max_velocity.fill(2.0); + return CommandGuard(limits); +} + +TEST(CommandGuard, AcceptsFiniteCommandWithinPositionAndVelocityLimits) +{ + const auto guard = makeGuard(); + std::array previous{}; + std::array command{}; + command[0] = 0.01; + + EXPECT_EQ( + guard.validate(command, previous, 0.01, true), + CommandGuardResult::OK); +} + +TEST(CommandGuard, RejectsPositionLimitViolation) +{ + const auto guard = makeGuard(); + std::array previous{}; + std::array command{}; + command[0] = 1.01; + + EXPECT_EQ( + guard.validate(command, previous, 0.01, true), + CommandGuardResult::POSITION_LIMIT); +} + +TEST(CommandGuard, RejectsVelocityLimitViolation) +{ + const auto guard = makeGuard(); + std::array previous{}; + std::array command{}; + command[0] = 0.1; + + EXPECT_EQ( + guard.validate(command, previous, 0.01, true), + CommandGuardResult::VELOCITY_LIMIT); +} + +TEST(CommandGuard, RejectsNonFiniteCommand) +{ + const auto guard = makeGuard(); + std::array previous{}; + std::array command{}; + command[0] = std::numeric_limits::quiet_NaN(); + + EXPECT_EQ( + guard.validate(command, previous, 0.01, true), + CommandGuardResult::NON_FINITE); +} + +IIWAStateSnapshot nominalSnapshot() +{ + IIWAStateSnapshot snapshot; + snapshot.sample_time = 0.01; + snapshot.quality = KUKA::FRI::EXCELLENT; + snapshot.safety_state = KUKA::FRI::NORMAL_OPERATION; + snapshot.operation_mode = KUKA::FRI::AUTOMATIC_MODE; + snapshot.drive_state = KUKA::FRI::ACTIVE; + snapshot.client_command_mode = KUKA::FRI::POSITION; + snapshot.control_mode = KUKA::FRI::POSITION_CONTROL_MODE; + snapshot.tracking_performance = 0.9; + return snapshot; +} + +TEST(StateGuard, AcceptsNominalPositionState) +{ + const StateGuard guard; + EXPECT_EQ( + guard.validatePositionState(nominalSnapshot(), KUKA::FRI::COMMANDING_ACTIVE), + StateGuardResult::OK); +} + +TEST(StateGuard, ReportsSafetyStopSeparately) +{ + auto snapshot = nominalSnapshot(); + snapshot.safety_state = KUKA::FRI::SAFETY_STOP_LEVEL_1; + + EXPECT_EQ( + StateGuard().validatePositionState(snapshot, KUKA::FRI::COMMANDING_ACTIVE), + StateGuardResult::SAFETY_STOP); +} + +TEST(StateGuard, ReportsWrongControlModeSeparately) +{ + auto snapshot = nominalSnapshot(); + snapshot.control_mode = KUKA::FRI::NO_CONTROL; + + EXPECT_EQ( + StateGuard().validatePositionState(snapshot, KUKA::FRI::COMMANDING_ACTIVE), + StateGuardResult::CONTROL_MODE); +} + +} // namespace iiwa_controller::test diff --git a/src/iiwa_controller/test/test_safety_contract.py b/src/iiwa_controller/test/test_safety_contract.py new file mode 100644 index 0000000..ad1ba62 --- /dev/null +++ b/src/iiwa_controller/test/test_safety_contract.py @@ -0,0 +1,34 @@ +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +HEADER = (ROOT / "include/iiwa_controller/FRIClient.h").read_text() +FRI_CLIENT = (ROOT / "src/FRIClient.cpp").read_text() +HARDWARE = (ROOT / "src/IIWAHardwareInterface.cpp").read_text() + + +def test_fri_data_exchange_does_not_use_a_mutex_in_the_rt_path(): + assert "#include " not in HEADER + assert "lock_guard" not in FRI_CLIENT + + +def test_cleanup_releases_application_before_its_dependencies(): + cleanup = HARDWARE[HARDWARE.index("on_cleanup"):] + assert cleanup.index("app_.reset()") < cleanup.index("connection_.reset()") + assert cleanup.index("connection_.reset()") < cleanup.index("fri_client_.reset()") + + +def test_fri_step_failure_stops_the_session(): + step_pos = HARDWARE.index("app_->step()") + after_step = HARDWARE[step_pos:step_pos + 700] + assert "fri_running_.store(false" in after_step + + +def test_activation_timeout_is_an_error(): + timeout_pos = HARDWARE.index("FRI не подключился") + after_timeout = HARDWARE[timeout_pos:timeout_pos + 500] + assert "CallbackReturn::ERROR" in after_timeout + + +def test_commanding_snapshot_keeps_measured_position_from_fri(): + assert "snapshot_.measured_pos = filtered_pos_" not in FRI_CLIENT diff --git a/src/iiwa_planning/test/conftest.py b/src/iiwa_planning/test/conftest.py new file mode 100644 index 0000000..b15840a --- /dev/null +++ b/src/iiwa_planning/test/conftest.py @@ -0,0 +1,232 @@ +import importlib.util +import sys +import types +import uuid +from pathlib import Path + +import pytest + + +class _DummyActionServer: + def __init__(self, *args, **kwargs): + self.args = args + self.kwargs = kwargs + + +class _DummyGoalResponse: + ACCEPT = "accept" + REJECT = "reject" + + +class _DummyCancelResponse: + ACCEPT = "accept" + REJECT = "reject" + + +class _DummyNode: + pass + + +class _DummyExecutor: + def add_node(self, _node): + return None + + def spin(self): + return None + + def spin_once(self, timeout_sec=None): + return None + + +class _DummyPoseStamped: + def __init__(self): + self.header = types.SimpleNamespace(frame_id="") + self.pose = types.SimpleNamespace( + position=types.SimpleNamespace(x=0.0, y=0.0, z=0.0), + orientation=types.SimpleNamespace(x=0.0, y=0.0, z=0.0, w=1.0), + ) + + +class _DummyPlanRequestParameters: + def __init__(self, _moveit, _group): + self.planning_pipeline = "" + self.planner_id = "" + self.planning_time = 0.0 + self.planning_attempts = 0 + self.max_velocity_scaling_factor = 0.0 + self.max_acceleration_scaling_factor = 0.0 + + +class _DummyRobotState: + def __init__(self, _robot_model): + self.group = None + self.joints = None + + def set_joint_group_positions(self, group, joints): + self.group = group + self.joints = list(joints) + + def update(self): + return None + + +class _DummyTrigger: + class Request: + pass + + class Response: + def __init__(self): + self.success = False + self.message = "" + + +class _DummyMoveToPose: + class Goal: + pass + + class Result: + def __init__(self): + self.success = False + self.message = "" + + class Feedback: + def __init__(self): + self.state = "" + + +class _DummyMoveToJoints: + class Goal: + pass + + class Result: + def __init__(self): + self.success = False + self.message = "" + + class Feedback: + def __init__(self): + self.state = "" + + +class _DummyMoveToNamedPose: + class Request: + pass + + class Response: + def __init__(self): + self.success = False + self.message = "" + + +class _DummyActionClient: + def __init__(self, *args, **kwargs): + self.args = args + self.kwargs = kwargs + + +class _DummySequentialWriter: + def __init__(self): + self.open_calls = [] + self.topics = [] + self.messages = [] + + def open(self, storage_opts, converter_opts): + self.open_calls.append((storage_opts, converter_opts)) + + def create_topic(self, topic_metadata): + self.topics.append(topic_metadata) + + def write(self, topic, payload, timestamp): + self.messages.append((topic, payload, timestamp)) + + +class _DummyStorageOptions: + def __init__(self, uri, storage_id): + self.uri = uri + self.storage_id = storage_id + + +class _DummyConverterOptions: + def __init__(self, input_serialization_format, output_serialization_format): + self.input_serialization_format = input_serialization_format + self.output_serialization_format = output_serialization_format + + +class _DummyTopicMetadata: + def __init__(self, id, name, type, serialization_format): + self.id = id + self.name = name + self.type = type + self.serialization_format = serialization_format + + +@pytest.fixture +def load_script_module(monkeypatch): + def _loader(relative_path: str): + module_defs = { + "rclpy": types.ModuleType("rclpy"), + "rclpy.node": types.ModuleType("rclpy.node"), + "rclpy.action": types.ModuleType("rclpy.action"), + "rclpy.callback_groups": types.ModuleType("rclpy.callback_groups"), + "rclpy.executors": types.ModuleType("rclpy.executors"), + "rclpy.serialization": types.ModuleType("rclpy.serialization"), + "geometry_msgs": types.ModuleType("geometry_msgs"), + "geometry_msgs.msg": types.ModuleType("geometry_msgs.msg"), + "moveit": types.ModuleType("moveit"), + "moveit.planning": types.ModuleType("moveit.planning"), + "moveit.core": types.ModuleType("moveit.core"), + "moveit.core.robot_state": types.ModuleType("moveit.core.robot_state"), + "std_srvs": types.ModuleType("std_srvs"), + "std_srvs.srv": types.ModuleType("std_srvs.srv"), + "iiwa_msgs": types.ModuleType("iiwa_msgs"), + "iiwa_msgs.action": types.ModuleType("iiwa_msgs.action"), + "iiwa_msgs.srv": types.ModuleType("iiwa_msgs.srv"), + "rosbag2_py": types.ModuleType("rosbag2_py"), + "rosidl_runtime_py": types.ModuleType("rosidl_runtime_py"), + "rosidl_runtime_py.utilities": types.ModuleType("rosidl_runtime_py.utilities"), + } + + module_defs["rclpy"].init = lambda *args, **kwargs: None + module_defs["rclpy"].shutdown = lambda *args, **kwargs: None + module_defs["rclpy.node"].Node = _DummyNode + module_defs["rclpy.action"].ActionServer = _DummyActionServer + module_defs["rclpy.action"].ActionClient = _DummyActionClient + module_defs["rclpy.action"].GoalResponse = _DummyGoalResponse + module_defs["rclpy.action"].CancelResponse = _DummyCancelResponse + module_defs["rclpy.callback_groups"].ReentrantCallbackGroup = object + module_defs["rclpy.executors"].MultiThreadedExecutor = _DummyExecutor + module_defs["rclpy.executors"].ExternalShutdownException = RuntimeError + module_defs["rclpy.serialization"].serialize_message = lambda msg: b"serialized" + + module_defs["geometry_msgs.msg"].PoseStamped = _DummyPoseStamped + + module_defs["moveit.planning"].MoveItPy = object + module_defs["moveit.planning"].PlanningComponent = object + module_defs["moveit.planning"].PlanRequestParameters = _DummyPlanRequestParameters + module_defs["moveit.core.robot_state"].RobotState = _DummyRobotState + + module_defs["std_srvs.srv"].Trigger = _DummyTrigger + + module_defs["iiwa_msgs.action"].MoveToPose = _DummyMoveToPose + module_defs["iiwa_msgs.action"].MoveToJoints = _DummyMoveToJoints + module_defs["iiwa_msgs.srv"].MoveToNamedPose = _DummyMoveToNamedPose + + module_defs["rosbag2_py"].SequentialWriter = _DummySequentialWriter + module_defs["rosbag2_py"].StorageOptions = _DummyStorageOptions + module_defs["rosbag2_py"].ConverterOptions = _DummyConverterOptions + module_defs["rosbag2_py"].TopicMetadata = _DummyTopicMetadata + + module_defs["rosidl_runtime_py.utilities"].get_message = lambda _type_name: object + + for name, module in module_defs.items(): + monkeypatch.setitem(sys.modules, name, module) + + script_path = Path("/home/daniel/dev/ros2_iiwa7") / relative_path + module_name = f"test_{script_path.stem}_{uuid.uuid4().hex}" + spec = importlib.util.spec_from_file_location(module_name, script_path) + loaded_module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(loaded_module) + return loaded_module + + return _loader diff --git a/src/iiwa_planning/test/test_motion_sequence_runner.py b/src/iiwa_planning/test/test_motion_sequence_runner.py new file mode 100644 index 0000000..01f825b --- /dev/null +++ b/src/iiwa_planning/test/test_motion_sequence_runner.py @@ -0,0 +1,73 @@ +import threading +from pathlib import Path +from types import SimpleNamespace + + +def _make_logger(): + return SimpleNamespace(info=lambda *args, **kwargs: None, warn=lambda *args, **kwargs: None) + + +def _build_runner(module): + runner = module.MotionSequenceRunner.__new__(module.MotionSequenceRunner) + runner.get_logger = lambda: _make_logger() + runner.close_bag = lambda: None + runner._n_iter = 1 + runner._delay = 0.0 + runner._home = {"joints": [0.0] * 7} + runner._waypoints = [{"x": 0.5}, {"x": 0.6}] + return runner + + +def test_run_stops_after_first_failed_waypoint(load_script_module): + module = load_script_module("src/iiwa_planning/scripts/motion_sequence_runner.py") + runner = _build_runner(module) + calls = [] + results = iter([True, False, True, True]) + + def fake_send_waypoint(waypoint, idx=None): + calls.append((waypoint, idx)) + return next(results) + + runner._send_waypoint = fake_send_waypoint + + done_event = threading.Event() + runner.run(done_event) + + assert done_event.is_set() + assert calls == [ + (runner._home, None), + (runner._waypoints[0], 0), + ] + + +def test_init_bag_refuses_existing_directory_outside_temp_root(load_script_module, monkeypatch, tmp_path): + module = load_script_module("src/iiwa_planning/scripts/motion_sequence_runner.py") + runner = module.MotionSequenceRunner.__new__(module.MotionSequenceRunner) + runner.get_logger = lambda: _make_logger() + + unsafe_bag_dir = Path("/home/daniel/dev/ros2_iiwa7") / "tmp-existing-bag" + unsafe_bag_dir.mkdir(exist_ok=True) + (unsafe_bag_dir / "keep.txt").write_text("keep") + runner._bag_path = str(unsafe_bag_dir) + + removed = [] + + def fake_rmtree(path): + removed.append(Path(path)) + + monkeypatch.setattr(module.shutil, "rmtree", fake_rmtree) + + try: + try: + runner._init_bag() + except ValueError as exc: + assert "temporary" in str(exc).lower() or "tmp" in str(exc).lower() + else: + raise AssertionError("Expected ValueError for unsafe bag path") + + assert removed == [] + assert unsafe_bag_dir.exists() + assert (unsafe_bag_dir / "keep.txt").exists() + finally: + (unsafe_bag_dir / "keep.txt").unlink(missing_ok=True) + unsafe_bag_dir.rmdir() diff --git a/src/iiwa_planning/test/test_move_to_pose_server.py b/src/iiwa_planning/test/test_move_to_pose_server.py new file mode 100644 index 0000000..e3b8850 --- /dev/null +++ b/src/iiwa_planning/test/test_move_to_pose_server.py @@ -0,0 +1,206 @@ +import math +import threading +import time +from types import SimpleNamespace + +import pytest + + +class _FakeLogger: + def __init__(self): + self.errors = [] + self.infos = [] + + def info(self, message): + self.infos.append(message) + + def error(self, message): + self.errors.append(message) + + +class _FakeGoalHandle: + def __init__(self, request): + self.request = request + self.is_cancel_requested = False + self.feedback_states = [] + self.aborted = False + self.succeeded = False + self.cancelled = False + + def publish_feedback(self, feedback): + self.feedback_states.append(feedback.state) + + def abort(self): + self.aborted = True + + def succeed(self): + self.succeeded = True + + def canceled(self): + self.cancelled = True + + +class _FakeArm: + def __init__(self, plan_result): + self.plan_result = plan_result + self.goal_state_calls = [] + + def set_start_state_to_current_state(self): + return None + + def set_goal_state(self, **kwargs): + self.goal_state_calls.append(kwargs) + + def plan(self, single_plan_parameters=None): + return self.plan_result + + +class _FakeTrajectoryExecutionManager: + def __init__(self): + self.stop_calls = 0 + + def stop_execution(self): + self.stop_calls += 1 + + +class _FakeMoveIt: + def __init__(self, execute_result): + self.execute_result = execute_result + self.execution_manager = _FakeTrajectoryExecutionManager() + + def execute(self, trajectory, controllers=None): + return self.execute_result(trajectory, controllers) + + def get_trajectory_execution_manager(self): + return self.execution_manager + + +class _FakeRobotModel: + def __init__(self, bounds): + self._bounds = bounds + + def get_joint_bounds(self, group_name): + return list(self._bounds) + + +def _build_server(module, *, plan_result=None, execute_result=None): + server = module.IiwaMotionServer.__new__(module.IiwaMotionServer) + server._planning_group = "iiwa_arm" + server._pose_link = "tcp" + server._default_frame = "base_link" + server._default_planner = "ompl" + server._planning_attempts = 1 + server._execution_timeout_sec = 0.05 + server._logger = _FakeLogger() + server.get_logger = lambda: server._logger + server._robot_model = _FakeRobotModel([(-1.0, 1.0)] * 7) + server._arm = _FakeArm(plan_result or SimpleNamespace(trajectory=object())) + server._moveit = _FakeMoveIt(execute_result or (lambda *_args, **_kwargs: True)) + server._motion_lock = threading.Lock() + server._active_motion_kind = None + return server + + +@pytest.mark.parametrize( + ("joints", "speed", "expected_fragment"), + [ + ([0.0, 0.0, 0.0], 0.2, "7"), + ([0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 2.0], 0.2, "limit"), + ([0.0, 0.0, 0.0, 0.0, 0.0, 0.0, math.nan], 0.2, "finite"), + ([0.0] * 7, "fast", "speed"), + ], +) +def test_execute_joints_rejects_invalid_joint_targets(load_script_module, joints, speed, expected_fragment): + module = load_script_module("src/iiwa_planning/scripts/move_to_pose_server.py") + server = _build_server(module) + goal_handle = _FakeGoalHandle(SimpleNamespace(joints=joints, speed=speed)) + + result = server._execute_joints(goal_handle) + + assert result.success is False + assert goal_handle.aborted is True + assert expected_fragment.lower() in result.message.lower() + + +def test_parallel_motion_rejects_second_goal_and_named_pose(load_script_module): + module = load_script_module("src/iiwa_planning/scripts/move_to_pose_server.py") + server = _build_server(module) + + assert server._handle_pose_goal(object()) == module.GoalResponse.ACCEPT + assert server._handle_joints_goal(object()) == module.GoalResponse.REJECT + + response = module.MoveToNamedPose.Response() + response = server._handle_named( + SimpleNamespace(name="home", speed=0.2, accel_scale=0.2), + response, + ) + + assert response.success is False + assert "busy" in response.message.lower() or "parallel" in response.message.lower() + + +def test_plan_and_execute_reports_moveit_execution_failure(load_script_module): + module = load_script_module("src/iiwa_planning/scripts/move_to_pose_server.py") + server = _build_server(module, execute_result=lambda *_args, **_kwargs: False) + goal_handle = _FakeGoalHandle(SimpleNamespace()) + + ok, message = server._plan_and_execute(SimpleNamespace(), goal_handle) + + assert ok is False + assert "ошиб" in message.lower() or "fail" in message.lower() + + +def test_plan_and_execute_times_out_without_hanging(load_script_module): + module = load_script_module("src/iiwa_planning/scripts/move_to_pose_server.py") + block_event = threading.Event() + + def hanging_execute(*_args, **_kwargs): + block_event.wait(timeout=1.0) + return True + + server = _build_server(module, execute_result=hanging_execute) + goal_handle = _FakeGoalHandle(SimpleNamespace()) + + outcome = {} + + def run_plan(): + outcome["result"] = server._plan_and_execute(SimpleNamespace(), goal_handle) + + worker = threading.Thread(target=run_plan, daemon=True) + worker.start() + worker.join(timeout=0.3) + + assert not worker.is_alive(), "Expected timeout handling to finish promptly" + assert outcome["result"][0] is False + assert "timeout" in outcome["result"][1].lower() or "time" in outcome["result"][1].lower() + + +def test_plan_and_execute_cancel_finishes_without_waiting_forever(load_script_module): + module = load_script_module("src/iiwa_planning/scripts/move_to_pose_server.py") + block_event = threading.Event() + + def hanging_execute(*_args, **_kwargs): + block_event.wait(timeout=1.0) + return True + + server = _build_server(module, execute_result=hanging_execute) + goal_handle = _FakeGoalHandle(SimpleNamespace()) + goal_handle.is_cancel_requested = False + + outcome = {} + + def request_cancel(): + time.sleep(0.05) + goal_handle.is_cancel_requested = True + + def run_plan(): + outcome["result"] = server._plan_and_execute(SimpleNamespace(), goal_handle) + + threading.Thread(target=request_cancel, daemon=True).start() + worker = threading.Thread(target=run_plan, daemon=True) + worker.start() + worker.join(timeout=0.3) + + assert not worker.is_alive(), "Expected cancel handling to finish promptly" + assert outcome["result"][0] is None + assert "cancel" in outcome["result"][1].lower() diff --git a/src/iiwa_web/test/conftest.py b/src/iiwa_web/test/conftest.py new file mode 100644 index 0000000..2ff2193 --- /dev/null +++ b/src/iiwa_web/test/conftest.py @@ -0,0 +1,198 @@ +import sys +import types +from contextlib import asynccontextmanager +from pathlib import Path + + +PACKAGE_ROOT = Path(__file__).resolve().parents[1] +if str(PACKAGE_ROOT) not in sys.path: + sys.path.insert(0, str(PACKAGE_ROOT)) + + +def _install_ros_stubs() -> None: + ament_index_python = types.ModuleType("ament_index_python") + ament_index_packages = types.ModuleType("ament_index_python.packages") + + def get_package_share_directory(package_name: str) -> str: + return str(PACKAGE_ROOT.parent / package_name) + + ament_index_packages.get_package_share_directory = get_package_share_directory + ament_index_python.packages = ament_index_packages + sys.modules.setdefault("ament_index_python", ament_index_python) + sys.modules.setdefault("ament_index_python.packages", ament_index_packages) + + builtin_interfaces = types.ModuleType("builtin_interfaces") + builtin_interfaces_msg = types.ModuleType("builtin_interfaces.msg") + + class Duration: + def __init__(self, sec: int = 0, nanosec: int = 0): + self.sec = sec + self.nanosec = nanosec + + builtin_interfaces_msg.Duration = Duration + builtin_interfaces.msg = builtin_interfaces_msg + sys.modules.setdefault("builtin_interfaces", builtin_interfaces) + sys.modules.setdefault("builtin_interfaces.msg", builtin_interfaces_msg) + + trajectory_msgs = types.ModuleType("trajectory_msgs") + trajectory_msgs_msg = types.ModuleType("trajectory_msgs.msg") + + class JointTrajectoryPoint: + def __init__(self): + self.positions = [] + self.time_from_start = None + + class JointTrajectory: + def __init__(self): + self.joint_names = [] + self.points = [] + + trajectory_msgs_msg.JointTrajectory = JointTrajectory + trajectory_msgs_msg.JointTrajectoryPoint = JointTrajectoryPoint + trajectory_msgs.msg = trajectory_msgs_msg + sys.modules.setdefault("trajectory_msgs", trajectory_msgs) + sys.modules.setdefault("trajectory_msgs.msg", trajectory_msgs_msg) + + sensor_msgs = types.ModuleType("sensor_msgs") + sensor_msgs_msg = types.ModuleType("sensor_msgs.msg") + + class JointState: + def __init__(self, position=()): + self.position = position + + sensor_msgs_msg.JointState = JointState + sensor_msgs.msg = sensor_msgs_msg + sys.modules.setdefault("sensor_msgs", sensor_msgs) + sys.modules.setdefault("sensor_msgs.msg", sensor_msgs_msg) + + std_srvs = types.ModuleType("std_srvs") + std_srvs_srv = types.ModuleType("std_srvs.srv") + + class Trigger: + class Request: + pass + + std_srvs_srv.Trigger = Trigger + std_srvs.srv = std_srvs_srv + sys.modules.setdefault("std_srvs", std_srvs) + sys.modules.setdefault("std_srvs.srv", std_srvs_srv) + + fastmcp = types.ModuleType("fastmcp") + + class _Router: + @asynccontextmanager + async def lifespan_context(self, _): + yield + + class _HttpApp: + def __init__(self): + self.router = _Router() + + class FastMCP: + @classmethod + def from_fastapi(cls, app): + return cls() + + def http_app(self, path="/mcp"): + return _HttpApp() + + fastmcp.FastMCP = FastMCP + sys.modules.setdefault("fastmcp", fastmcp) + + rclpy = types.ModuleType("rclpy") + rclpy_time = types.ModuleType("rclpy.time") + rclpy_duration = types.ModuleType("rclpy.duration") + rclpy_node = types.ModuleType("rclpy.node") + rclpy_action = types.ModuleType("rclpy.action") + + class Time: + pass + + class DurationValue: + def __init__(self, seconds: float = 0.0): + self.seconds = seconds + + class _Logger: + def info(self, *_args, **_kwargs): + return None + + class _Parameter: + def __init__(self, value): + self.value = value + + class _Publisher: + def __init__(self): + self.published = [] + + def publish(self, msg): + self.published.append(msg) + + class _Client: + def wait_for_service(self, timeout_sec): + return True + + def call_async(self, request): + future = types.SimpleNamespace() + future.done = lambda: True + future.result = lambda: types.SimpleNamespace(success=True, message="") + return future + + class Node: + def __init__(self, name): + self.name = name + self._parameters = {} + self._logger = _Logger() + + def declare_parameter(self, name, default): + self._parameters[name] = default + + def get_parameter(self, name): + return _Parameter(self._parameters[name]) + + def create_subscription(self, *_args, **_kwargs): + return object() + + def create_publisher(self, *_args, **_kwargs): + return _Publisher() + + def create_client(self, *_args, **_kwargs): + return _Client() + + def get_logger(self): + return self._logger + + class ActionClient: + def __init__(self, *_args, **_kwargs): + pass + + rclpy.init = lambda: None + rclpy.spin = lambda node: None + rclpy.time = rclpy_time + rclpy.duration = rclpy_duration + rclpy.Time = Time + rclpy_time.Time = Time + rclpy_duration.Duration = DurationValue + rclpy_node.Node = Node + rclpy_action.ActionClient = ActionClient + sys.modules.setdefault("rclpy", rclpy) + sys.modules.setdefault("rclpy.time", rclpy_time) + sys.modules.setdefault("rclpy.duration", rclpy_duration) + sys.modules.setdefault("rclpy.node", rclpy_node) + sys.modules.setdefault("rclpy.action", rclpy_action) + + tf2_ros = types.ModuleType("tf2_ros") + + class Buffer: + def lookup_transform(self, *_args, **_kwargs): + raise RuntimeError("tf unavailable in tests") + + class TransformListener: + def __init__(self, *_args, **_kwargs): + pass + + tf2_ros.Buffer = Buffer + tf2_ros.TransformListener = TransformListener + sys.modules.setdefault("tf2_ros", tf2_ros) + + +_install_ros_stubs() diff --git a/src/iiwa_web/test/test_ros_action_timeout.py b/src/iiwa_web/test/test_ros_action_timeout.py new file mode 100644 index 0000000..8134da7 --- /dev/null +++ b/src/iiwa_web/test/test_ros_action_timeout.py @@ -0,0 +1,65 @@ +import itertools + +import pytest + +from iiwa_web.ros_node import CobotWebNode +from iiwa_web import ros_node + + +class DoneFuture: + def __init__(self, result): + self._result = result + + def done(self): + return True + + def result(self): + return self._result + + +class PendingFuture: + def done(self): + return False + + def result(self): + return None + + +class GoalHandle: + accepted = True + + def __init__(self): + self.cancel_calls = 0 + + def get_result_async(self): + return PendingFuture() + + def cancel_goal_async(self): + self.cancel_calls += 1 + return DoneFuture(None) + + +class ActionClientStub: + def __init__(self, goal_handle): + self.goal_handle = goal_handle + + def wait_for_server(self, timeout_sec): + return True + + def send_goal_async(self, goal): + return DoneFuture(self.goal_handle) + + +def test_send_action_cancels_goal_when_result_times_out(monkeypatch): + node = object.__new__(CobotWebNode) + goal_handle = GoalHandle() + node._action_clients = {"cobot/move_to_pose": ActionClientStub(goal_handle)} + + timestamps = itertools.chain([0.0, 0.3, 0.6], itertools.repeat(0.9)) + monkeypatch.setattr(ros_node.time, "monotonic", lambda: next(timestamps)) + monkeypatch.setattr(ros_node.time, "sleep", lambda _seconds: None) + + with pytest.raises(TimeoutError): + node.send_action(object, "cobot/move_to_pose", object(), timeout=0.5) + + assert goal_handle.cancel_calls == 1 diff --git a/src/iiwa_web/test/test_runner_upload_security.py b/src/iiwa_web/test/test_runner_upload_security.py new file mode 100644 index 0000000..505f67b --- /dev/null +++ b/src/iiwa_web/test/test_runner_upload_security.py @@ -0,0 +1,130 @@ +import asyncio +from pathlib import Path + +import pytest + +from iiwa_web import runner + + +class DummyProcess: + def __init__(self, cmd): + self.cmd = cmd + self.pid = 4321 + self.stdout = iter(()) + self.returncode = None + + def poll(self): + return None + + +class DummyThread: + def __init__(self, target=None, args=(), daemon=False): + self.target = target + self.args = args + self.daemon = daemon + + def start(self): + return None + + +class UploadStub: + def __init__(self, filename: str, content: bytes): + self.filename = filename + self._content = content + + async def read(self) -> bytes: + return self._content + + +@pytest.fixture +def runner_module(monkeypatch, tmp_path): + upload_dir = tmp_path / "uploads" + upload_dir.mkdir() + bag_root = tmp_path / "bags" + bag_root.mkdir() + + monkeypatch.setattr(runner, "_UPLOAD_DIR", upload_dir) + monkeypatch.setattr(runner, "_BAG_ROOT", bag_root, raising=False) + monkeypatch.setattr(runner, "MAX_UPLOAD_BYTES", 8, raising=False) + monkeypatch.setattr(runner, "_process", None) + + started = {} + + def fake_popen(cmd, **kwargs): + started["cmd"] = cmd + proc = DummyProcess(cmd) + started["proc"] = proc + return proc + + monkeypatch.setattr(runner.subprocess, "Popen", fake_popen) + monkeypatch.setattr(runner.threading, "Thread", DummyThread) + + return runner, upload_dir, bag_root, started + + +def _upload_file(name: str, content: bytes) -> UploadStub: + return UploadStub(filename=name, content=content) + + +def test_start_runner_sanitizes_uploaded_filename(runner_module): + runner, upload_dir, _bag_root, started = runner_module + + response = asyncio.run( + runner.start_runner( + config=_upload_file("../escape.json", b"{}"), + n_iterations=3, + delay_between_iterations=5.0, + bag_path="", + topics="", + joints_action="cobot/move_to_joints", + pose_action="cobot/move_to_pose", + ) + ) + + assert response["config"] == "escape.json" + assert (upload_dir / "escape.json").exists() + assert not (upload_dir.parent / "escape.json").exists() + config_arg = next(part for part in started["cmd"] if part.startswith("config_path:=")) + assert Path(config_arg.split(":=", 1)[1]).parent == upload_dir + + +def test_start_runner_rejects_oversized_upload(runner_module): + runner, _upload_dir, _bag_root, started = runner_module + + with pytest.raises(runner.HTTPException) as excinfo: + asyncio.run( + runner.start_runner( + config=_upload_file("config.json", b"123456789"), + n_iterations=3, + delay_between_iterations=5.0, + bag_path="", + topics="", + joints_action="cobot/move_to_joints", + pose_action="cobot/move_to_pose", + ) + ) + + assert excinfo.value.status_code == 413 + assert "слишком" in excinfo.value.detail.lower() + assert "cmd" not in started + + +def test_start_runner_rejects_bag_path_outside_allowed_root(runner_module): + runner, _upload_dir, _bag_root, started = runner_module + + with pytest.raises(runner.HTTPException) as excinfo: + asyncio.run( + runner.start_runner( + config=_upload_file("config.json", b"{}"), + n_iterations=3, + delay_between_iterations=5.0, + bag_path="/tmp/escape", + topics="", + joints_action="cobot/move_to_joints", + pose_action="cobot/move_to_pose", + ) + ) + + assert excinfo.value.status_code == 422 + assert "bag_path" in excinfo.value.detail + assert "cmd" not in started diff --git a/src/iiwa_web/test/test_trajectory_validation.py b/src/iiwa_web/test/test_trajectory_validation.py new file mode 100644 index 0000000..b901301 --- /dev/null +++ b/src/iiwa_web/test/test_trajectory_validation.py @@ -0,0 +1,118 @@ +import importlib +import pytest +from fastapi import HTTPException +from pydantic import ValidationError + +from iiwa_web import config_loader + + +class RecordingBridge: + def __init__(self): + self.messages = [] + + def publish(self, topic_name, message_type, msg): + self.messages.append((topic_name, message_type, msg)) + + def get_latest(self, topic_name): + return None + + +@pytest.fixture +def trajectory_module(monkeypatch): + monkeypatch.setattr( + config_loader, + "load_joint_names", + lambda *args, **kwargs: [f"joint{i}" for i in range(1, 8)], + ) + monkeypatch.setattr( + config_loader, + "load_joint_limits", + lambda *args, **kwargs: [(-1.0, 1.0)] * 7, + ) + + trajectory = importlib.import_module("iiwa_web.trajectory") + trajectory = importlib.reload(trajectory) + bridge = RecordingBridge() + monkeypatch.setattr(trajectory, "get_bridge", lambda: bridge) + return trajectory, bridge + + +@pytest.mark.parametrize( + ("payload", "detail"), + [ + ( + { + "points": [ + { + "positions": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0], + "time_from_start": 0.0, + } + ] + }, + "positions", + ), + ( + { + "points": [ + { + "positions": [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, float("nan")], + "time_from_start": 0.0, + } + ] + }, + "конеч", + ), + ], +) +def test_send_request_rejects_invalid_payloads(trajectory_module, payload, detail): + trajectory, bridge = trajectory_module + + with pytest.raises((ValidationError, HTTPException)) as excinfo: + request = trajectory.SendRequest.model_validate(payload) + trajectory.send_trajectory(request) + + assert detail.lower() in str(excinfo.value).lower() + assert bridge.messages == [] + + +def test_send_trajectory_rejects_non_monotonic_time(trajectory_module): + trajectory, bridge = trajectory_module + payload = { + "points": [ + { + "positions": [0.0] * trajectory.N_JOINTS, + "time_from_start": 0.5, + }, + { + "positions": [0.1] * trajectory.N_JOINTS, + "time_from_start": 0.4, + }, + ] + } + + with pytest.raises(ValidationError) as excinfo: + request = trajectory.SendRequest.model_validate(payload) + trajectory.send_trajectory(request) + + assert "монотон" in str(excinfo.value).lower() + assert bridge.messages == [] + + +def test_send_trajectory_enforces_joint_limits_even_when_bypass_requested(trajectory_module): + trajectory, bridge = trajectory_module + payload = { + "points": [ + { + "positions": [100.0] * trajectory.N_JOINTS, + "time_from_start": 0.0, + } + ], + "validate_limits": False, + } + + with pytest.raises(HTTPException) as excinfo: + request = trajectory.SendRequest.model_validate(payload) + trajectory.send_trajectory(request) + + assert "вне диапазона" in excinfo.value.detail.lower() + assert bridge.messages == [] diff --git a/src/iiwa_web/test/test_web_auth.py b/src/iiwa_web/test/test_web_auth.py new file mode 100644 index 0000000..1273b02 --- /dev/null +++ b/src/iiwa_web/test/test_web_auth.py @@ -0,0 +1,137 @@ +import pytest + +from iiwa_web import config_loader, ros_node + + +def test_web_node_defaults_to_loopback_host(): + node = ros_node.CobotWebNode() + + assert node.get_parameter("host").value == "127.0.0.1" + + +def test_external_bind_requires_token(): + import importlib + + monkeypatchable_main = _import_main_with_stubbed_config() + + with pytest.raises(RuntimeError): + monkeypatchable_main.ensure_safe_bind("0.0.0.0", token=None) + + monkeypatchable_main.ensure_safe_bind("0.0.0.0", token="secret-token") + + +def test_web_settings_load_token(): + import sys + from pathlib import Path + + utils_root = Path(__file__).resolve().parents[2] / "iiwa_utils" + if str(utils_root) not in sys.path: + sys.path.insert(0, str(utils_root)) + + from iiwa_utils.setting_loader import _parse_web + + settings = _parse_web( + {"enabled": True, "host": "0.0.0.0", "token": "configured-secret"}, + "/tmp", + ) + + assert settings.token == "configured-secret" + + +def test_web_server_passes_configured_token_to_process_environment(monkeypatch): + import sys + import types + from pathlib import Path + from types import SimpleNamespace + + class FakeNode: + def __init__(self, *args, **kwargs): + self.additional_env = kwargs.get("additional_env") + + launch_ros = types.ModuleType("launch_ros") + launch_ros_actions = types.ModuleType("launch_ros.actions") + launch_ros_actions.Node = FakeNode + launch_ros.actions = launch_ros_actions + monkeypatch.setitem(sys.modules, "launch_ros", launch_ros) + monkeypatch.setitem(sys.modules, "launch_ros.actions", launch_ros_actions) + + launch_root = str(Path(__file__).resolve().parents[2] / "iiwa_bringup" / "launch") + if launch_root not in sys.path: + sys.path.insert(0, launch_root) + + from supported.optional_nodes import make_web_server_node + + settings = SimpleNamespace( + web=SimpleNamespace( + host="0.0.0.0", + port=8007, + endpoints="/tmp/endpoints.yaml", + joint_limits="/tmp/joint_limits.yaml", + token="configured-secret", + ) + ) + + node = make_web_server_node(settings, use_sim_time=False) + + assert node.additional_env == {"IIWA_WEB_TOKEN": "configured-secret"} + + +def test_bearer_token_is_required_and_compared_in_constant_time(monkeypatch): + main = _import_main_with_stubbed_config() + compared = [] + + def fake_compare(left, right): + compared.append((left, right)) + return left == right + + monkeypatch.setattr(main.hmac, "compare_digest", fake_compare) + + assert main.has_valid_bearer_token(None, "secret-token") is False + assert main.has_valid_bearer_token("Bearer wrong", "secret-token") is False + assert main.has_valid_bearer_token("Bearer secret-token", "secret-token") is True + assert compared == [("wrong", "secret-token"), ("secret-token", "secret-token")] + + +def test_docs_paths_are_public_but_api_routes_are_not(): + main = _import_main_with_stubbed_config() + + assert main.is_public_auth_path("/docs") + assert main.is_public_auth_path("/docs/oauth2-redirect") + assert main.is_public_auth_path("/openapi.json") + assert main.is_public_auth_path("/redoc") + assert not main.is_public_auth_path("/protected") + + +def test_openapi_declares_bearer_security_for_swagger_authorize_button(): + from fastapi import FastAPI + + main = _import_main_with_stubbed_config() + app = FastAPI() + + @app.get("/protected") + def protected(): + return {"ok": True} + + main.install_bearer_openapi(app, "secret-token") + schema = app.openapi() + + assert schema["components"]["securitySchemes"]["BearerAuth"] == { + "type": "http", + "scheme": "bearer", + "bearerFormat": "opaque-token", + } + assert schema["paths"]["/protected"]["get"]["security"] == [ + {"BearerAuth": []} + ] + + +def _import_main_with_stubbed_config(): + monkeypatch_joint_names = [f"joint{i}" for i in range(1, 8)] + monkeypatch_joint_limits = [(-1.0, 1.0)] * 7 + config_loader.load_joint_names = lambda *args, **kwargs: list(monkeypatch_joint_names) + config_loader.load_joint_limits = lambda *args, **kwargs: list(monkeypatch_joint_limits) + + import importlib + + main = importlib.import_module("iiwa_web.main") + return importlib.reload(main)