Объединить разработки CAN и STM32 в master

python/tests/test_firmware_publish.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from setprotocol.firmware_publish import (
+    FirmwarePublication,
+    firmware_release_entry,
+    firmware_release_tag,
+    update_firmware_manifest,
+)
+
+
+class FirmwarePublishTests(unittest.TestCase):
+    def publication(self, path: Path, **overrides) -> FirmwarePublication:
+        fields = {
+            "path": path,
+            "product": "F103DS18",
+            "version_name": "1.1.0",
+            "version_code": 0x00010100,
+            "transport": "can",
+            "base_address": 0x08003000,
+            "notes": "Verified release",
+        }
+        fields.update(overrides)
+        return FirmwarePublication(**fields)
+
+    def test_publication_validates_file_and_metadata(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            image = Path(temporary) / "image.hex"
+            image.write_text(":00000001FF\n", encoding="ascii")
+            self.publication(image).validate()
+            with self.assertRaisesRegex(ValueError, "Unsupported"):
+                self.publication(image, transport="unknown").validate()
+
+    def test_entry_and_tag_are_deterministic(self) -> None:
+        publication = self.publication(Path("image.hex"))
+        entry = firmware_release_entry(
+            publication, "https://example.test/image.hex", "ab" * 32
+        )
+        self.assertEqual(
+            firmware_release_tag(publication), "firmware-F103DS18-v1.1.0"
+        )
+        self.assertEqual(entry["baseAddress"], "0x08003000")
+
+    def test_update_preserves_sections_and_replaces_same_release(self) -> None:
+        first = {
+            "product": "Device",
+            "versionCode": 7,
+            "versionName": "1.2.3",
+            "imageUrl": "https://example.test/old.bin",
+            "fileName": "old.bin",
+            "sha256": "11" * 32,
+            "transport": "rs485",
+        }
+        manifest = update_firmware_manifest(
+            {"windows": {"versionCode": 8}}, first
+        )
+        replacement = {
+            **first,
+            "imageUrl": "https://example.test/new.bin",
+            "fileName": "new.bin",
+            "sha256": "22" * 32,
+        }
+        updated = update_firmware_manifest(manifest, replacement)
+        self.assertEqual(updated["windows"], {"versionCode": 8})
+        self.assertEqual(len(updated["firmware"]["releases"]), 1)
+        self.assertEqual(
+            updated["firmware"]["releases"][0]["sha256"], "22" * 32
+        )
+        self.assertEqual(updated["firmware"]["catalogVersion"], 2)
+
+    def test_legacy_array_is_migrated_without_data_loss(self) -> None:
+        legacy = {
+            "product": "Legacy",
+            "versionCode": 1,
+            "versionName": "1.0",
+            "imageUrl": "https://example.test/legacy.bin",
+            "fileName": "legacy.bin",
+            "sha256": "33" * 32,
+            "transport": "rs485",
+        }
+        current = {**legacy, "product": "Current", "versionCode": 2}
+        updated = update_firmware_manifest({"firmware": [legacy]}, current)
+        self.assertCountEqual(
+            [row["product"] for row in updated["firmware"]["releases"]],
+            ["Legacy", "Current"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

python/tests/test_plot.py

@@ -56,6 +56,9 @@ class PlotTests(unittest.TestCase):
         self.assertAlmostEqual(20, self.core.db_delta(1, 10))
         self.assertAlmostEqual(-20, self.core.db_delta(10, 1))
         self.assertIsNone(self.core.db_delta(0, 1))
+        self.assertEqual(Bounds(0, 500, -2, 2), self.core.limits(0, 500, -2, 2))
+        with self.assertRaises(ValueError):
+            self.core.limits(1, 1, -2, 2)
 
 
 if __name__ == "__main__":

python/tests/test_setprotocol_can.py

@@ -0,0 +1,42 @@
+from setprotocol import Frame, FrameFlag, MessageType, build_frame, decode_datagram
+from setprotocol.can import CanAddress, CanFrame, CanReassembler, segment
+
+
+def test_can_id_roundtrip():
+    value = CanAddress(destination=13, source=0, priority=1, channel=7)
+    assert CanAddress.unpack(value.pack()) == value
+
+
+def test_segmented_setp_frame_roundtrip():
+    packet = build_frame(
+        Frame(
+            message_type=MessageType.FW_DATA,
+            sequence=0x1234,
+            payload=bytes(range(76)),
+            flags=FrameFlag.ACK_REQUIRED | FrameFlag.PRIORITY,
+            source=0,
+            destination=13,
+        )
+    )
+    frames = segment(packet, CanAddress(13, 0, priority=1, channel=1))
+    assert len(frames) == 14
+    reassembler = CanReassembler()
+    restored = None
+    for now_ms, frame in enumerate(frames):
+        restored = reassembler.feed(frame, now_ms)
+    assert restored == packet
+    assert decode_datagram(restored).payload == bytes(range(76))
+
+
+def test_reassembler_rejects_wrong_sequence():
+    packet = build_frame(Frame(MessageType.PING, 1, source=0, destination=13))
+    frames = segment(packet, CanAddress(13, 0))
+    reassembler = CanReassembler()
+    assert reassembler.feed(frames[0], 0) is None
+    damaged = CanFrame(frames[1].can_id, bytes((0x22,)) + frames[1].data[1:])
+    try:
+        reassembler.feed(damaged, 1)
+    except ValueError as error:
+        assert "последовательность" in str(error)
+    else:
+        raise AssertionError("wrong sequence must fail")

python/tests/test_spectrum.py

@@ -35,6 +35,9 @@ class SpectrumTests(unittest.TestCase):
                 peak = max(range(len(result.amplitudes)), key=result.amplitudes.__getitem__)
                 self.assertEqual(64, result.frequencies[peak])
                 self.assertAlmostEqual(3.25, result.amplitudes[peak], places=9)
+                detected = self.core.dominant_peak(result)
+                self.assertAlmostEqual(64, detected.frequency_hz, places=9)
+                self.assertAlmostEqual(3.25, detected.amplitude, places=9)
 
     def test_dc_and_nyquist_are_not_doubled(self):
         times, _ = self.sample()

python/tests/test_trends.py

@@ -14,6 +14,16 @@ FIXTURE = Path(__file__).resolve().parents[2] / "c/set-protocol/tests/fixtures/t
 
 
 class TrendTests(unittest.TestCase):
+    def test_multiplier_and_iq_scale_display_and_old_json_defaults(self):
+        signal = TrendSignal("scaled", multiplier=2.5, iq=3)
+        self.assertEqual(10.0, signal.display_value(32.0))
+        signal.validate("TMS2812")
+        payload = json.loads(encode_settings({"TMS2812": [signal]}))
+        payload["profiles"]["TMS2812"][0].pop("multiplier")
+        payload["profiles"]["TMS2812"][0].pop("iq")
+        restored = decode_settings(json.dumps(payload))["TMS2812"][0]
+        self.assertEqual((1.0, 0), (restored.multiplier, restored.iq))
+
     def test_shared_kotlin_fixture_and_round_trip(self):
         settings = decode_settings(FIXTURE.read_text(encoding="utf-8"))
         self.assertEqual(5, sum(map(len, settings.values())))
@@ -57,7 +67,6 @@ class TrendTests(unittest.TestCase):
         for value in ("-1", "+1", "FF", "0x", "1.0", "256"):
             with self.assertRaises(ValueError):
                 parse_address(value, 255)
-        self.assertEqual(-2.0, TrendSignal("s", valueType="INT16").word_value(65534))
 
     def test_bounded_history(self):
         history = TrendHistory()
@@ -81,6 +90,13 @@ class TrendTests(unittest.TestCase):
         self.assertIsNone(core.can_value(signal, frame_id ^ 0x08000000, 1, data))
         raw = replace(signal, source="CAN_RAW", address="0x321", extended=False, byteOffset=2)
         self.assertEqual(-2.0, core.can_value(raw, 0x321, 0, data))
+        self.assertEqual(-2.0, core.word_value(TrendSignal("s", valueType="INT16"), 65534))
+        self.assertEqual(bytes([232, 3, 2, 0, 52, 18, 255, 255]),
+                         core.watch_request(1000, [0x1234, 0xFFFF]))
+        core.validate_watch_ack(bytes([232, 3, 2, 0]), 1000, 2)
+        timestamp, words = core.watch_values(bytes([1, 2, 3, 4, 2, 0, 52, 18, 255, 255]), 2)
+        self.assertEqual(0x04030201, timestamp)
+        self.assertEqual([0x1234, 0xFFFF], words)
 
 
 if __name__ == "__main__":