From b8d57f56fe9b3cb77af5e62b864c5db1fdbf0d25 Mon Sep 17 00:00:00 2001
From: Josh Pinto <joshpinto6@outlook.com>
Date: Tue, 5 Aug 2025 22:37:32 -0400
Subject: [PATCH] :white_check_mark: Add unit test for LtAmp class

---
 README.md           |   9 ++++
 tests/test_ltamp.py | 109 ++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 118 insertions(+)
 create mode 100644 tests/test_ltamp.py

diff --git a/README.md b/README.md
index 8840600..b782a7c 100644
--- a/README.md
+++ b/README.md
@@ -63,6 +63,7 @@ Compatibility with LtAmp protocol:
 - [ ] Audio Status
 - [x] Audition Preset (status, start, exit)
 - [ ] Clear/rename/shift/swap/save (to/as) preset
+- [x] Retrieve preset
 - [x] Connection Status
 - [x] Current Preset (request/load)
 - [ ] DSP Unit Parameter
@@ -105,6 +106,14 @@ Fully-tested LT-series amps:
 
 LtAmp.py is licensed under the permissive MIT license, which means that you may fork, modify, adapt, and redistribute with few restrictions. If you wish to contribute your changes back to the base module, please open a [pull request](/pulls). To report bugs, request features, or discuss the project, open an [issue](/issues) or [discussion](/discussions).
 
+### 🧪 Testing
+
+To run unit tests, use the command:
+
+```bash
+python -m unittest discover tests
+```
+
 ## 🙌 Acknowledgements
 
 - Brent Maxwell ([@brentmaxwell](https://github.com/brentmaxwell)) and his LtAmp .NET libray: his published reverse engineering docs, schemas, proto files, etc. were instrumental in the creation of this Python module.
diff --git a/tests/test_ltamp.py b/tests/test_ltamp.py
new file mode 100644
index 0000000..0d96518
--- /dev/null
+++ b/tests/test_ltamp.py
@@ -0,0 +1,109 @@
+import unittest
+import time
+from ltamp import LtAmp
+import threading
+
+class TestLtAmp(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.amp = LtAmp()
+        try:
+            cls.amp.connect()
+            cls.amp.send_sync_begin()
+            time.sleep(1)
+            cls.amp.send_heartbeat()
+            cls.amp.send_sync_end()
+            cls._heartbeat_running = True
+            def heartbeat():
+                while cls._heartbeat_running:
+                    cls.amp.send_heartbeat()
+                    time.sleep(0.1)
+            cls._heartbeat_thread = threading.Thread(target=heartbeat, daemon=True)
+            cls._heartbeat_thread.start()
+        except Exception as e:
+            cls.amp = None
+            print(f"SKIPPING tests: Could not connect to amp: {e}")
+
+    @classmethod
+    def tearDownClass(cls):
+        if getattr(cls, "amp", None):
+            cls._heartbeat_running = False
+            cls._heartbeat_thread.join(timeout=2)
+            cls.amp.disconnect()
+
+    def setUp(self):
+        if self.amp is None:
+            self.skipTest("Amp not connected, skipping hardware integration tests.")
+
+    def test_firmware_version(self):
+        version = self.amp.request_firmware_version()
+        self.assertIsInstance(version, str)
+        self.assertRegex(version, r"^\d+\.\d+\.\d+$")
+
+    def test_product_id(self):
+        pid = self.amp.request_product_id()
+        self.assertIsInstance(pid, str)
+
+    def test_qa_slots(self):
+        slots = self.amp.request_qa_slots()
+        self.assertIsInstance(slots, list)
+        self.assertEqual(len(slots), 2)
+        self.assertTrue(all(isinstance(x, int) for x in slots))
+
+    def test_set_and_get_preset(self):
+        orig_preset = self.amp.request_current_preset()
+        idx = orig_preset["index"]
+        self.amp.set_preset(idx)
+        time.sleep(1)
+        curr = self.amp.request_current_preset()
+        self.assertEqual(curr["index"], idx)
+
+    def test_usb_gain(self):
+        orig_gain = self.amp.request_usb_gain()
+        self.amp.set_usb_gain(orig_gain)
+        time.sleep(0.5)
+        gain = self.amp.request_usb_gain()
+        self.assertEqual(gain, orig_gain)
+
+    
+    def test_memory_usage(self):
+        mem = self.amp.request_memory_usage()
+        self.assertIsInstance(mem, dict)
+        self.assertIn("stack", mem)
+        self.assertIn("heap", mem)
+        self.assertIsInstance(mem["stack"], int)
+        self.assertIsInstance(mem["heap"], int)
+
+    def test_processor_utilization(self):
+        util = self.amp.request_processor_utilization()
+        self.assertIsInstance(util, dict)
+        for key in ("percent", "minPercent", "maxPercent"):
+            self.assertIn(key, util)
+            self.assertIsInstance(util[key], float)
+            self.assertGreaterEqual(util[key], 0)
+
+    def test_send_sync_begin_end(self):
+        # Just ensure both calls do not raise
+        try:
+            self.amp.send_sync_begin()
+            time.sleep(1)
+            self.amp.send_sync_end()
+        except Exception as e:
+           self.fail(f"send_sync_begin() or send_sync_end() raised an exception: {e}")
+
+    def test_disconnect(self):
+        try:
+            self.amp.disconnect()
+        except Exception as e:
+            self.fail(f"disconnect() raised an exception: {e}")
+        # Reconnect for further tests (if needed)
+        try:
+            self.amp.connect()
+            self.amp.send_sync_begin()
+            time.sleep(1)
+            self.amp.send_sync_end()
+        except Exception as e:
+            self.skipTest(f"Could not reconnect to amp after disconnect: {e}")
+
+if __name__ == "__main__":
+    unittest.main()
-- 
2.51.2