77 lines
3.1 KiB
Python
77 lines
3.1 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from servos import ControleServos, PCA
|
|
|
|
class Fake:
|
|
def __init__(self): self.calls = []
|
|
def pulse(self, channel, value): self.calls.append((channel, value))
|
|
def off(self, channel): self.calls.append((channel, None))
|
|
|
|
class ServoTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.hw = Fake()
|
|
self.c = ControleServos(Path(self.tmp.name) / 'servos.json', self.hw)
|
|
def tearDown(self):
|
|
self.c.close()
|
|
self.tmp.cleanup()
|
|
def test_open_does_not_move(self):
|
|
self.assertEqual(len(self.c.status()['servos']), 6)
|
|
self.assertEqual(self.hw.calls, [])
|
|
def test_limits_and_types(self):
|
|
for pulse in [1399, 1601, True, float('nan')]:
|
|
with self.assertRaises(ValueError): self.c.move(0, pulse)
|
|
self.assertEqual(self.hw.calls, [])
|
|
def test_valid_pulse_and_stop(self):
|
|
self.c.move(0, 1500)
|
|
self.assertEqual(self.hw.calls[-1], (0, 1500))
|
|
self.c.stop()
|
|
self.assertIn((0, None), self.hw.calls)
|
|
def test_duplicate_channel_and_atomic_config(self):
|
|
config = self.c.status()['servos']
|
|
config[1]['canal'] = 0
|
|
with self.assertRaises(ValueError): self.c.save(config)
|
|
self.assertEqual(self.c.status()['servos'][1]['canal'], 1)
|
|
def test_expiry_turns_off(self):
|
|
self.c.move(1, 1500)
|
|
self.c.expire(float('inf'))
|
|
self.assertEqual(self.hw.calls[-1], (1, None))
|
|
def test_config_persists_and_supports_sixteen(self):
|
|
rows = [dict(nome=f'Motor {i}',canal=i,minimo=1400,maximo=1600,centro=1500) for i in range(16)]
|
|
self.c.save(rows)
|
|
other = ControleServos(self.c.path, self.hw)
|
|
try: self.assertEqual(other.status()['servos'], rows)
|
|
finally: other.close()
|
|
def test_frequency_initialization_disables_all_old_outputs(self):
|
|
hw = PCA.__new__(PCA)
|
|
hw.ready = False
|
|
writes = []
|
|
hw.write = lambda reg, values: writes.append((reg, values))
|
|
hw.read = lambda reg: 0x11
|
|
hw.pulse(1, 1500)
|
|
release = next(i for i, value in enumerate(writes) if value[0] == 10)
|
|
for channel in range(16):
|
|
self.assertIn((9 + 4*channel, [0x10]), writes[:release])
|
|
self.assertIn((0xFE, [121]), writes)
|
|
self.assertIn((10, [0, 0, 51, 1]), writes)
|
|
def test_global_register_cannot_overwrite_pulse_or_enable_other_channels(self):
|
|
hw = PCA.__new__(PCA)
|
|
hw.ready = False
|
|
registers = {}
|
|
def write(reg, values):
|
|
for offset, value in enumerate(values):
|
|
address = reg + offset
|
|
registers[address] = value
|
|
if 0xFA <= address <= 0xFD:
|
|
for channel in range(16):
|
|
registers[6 + 4*channel + address-0xFA] = value
|
|
hw.write = write
|
|
hw.read = lambda reg: 0x21
|
|
hw.pulse(1, 1500)
|
|
self.assertEqual(registers[12] + ((registers[13] & 15) << 8), 307)
|
|
for channel in range(16):
|
|
if channel != 1: self.assertTrue(registers[9 + 4*channel] & 16)
|
|
|
|
if __name__ == '__main__': unittest.main()
|