KiCad - the KiCad Bridge

import copy
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from handlers import live_routing as live, route


def board():
    return {'net_format':'name', 'nets':[{'name':'GND'}, {'name':'VCC'}, {'name':'123'}],
        'layers':[{'name':s} for s in ['F.Cu','In1.Cu','B.Cu']],
        'footprints':[{'reference':'J1','layer':'F.Cu','position':{'x':10,'y':20,'rotation':90},
            'pads':[{'number':'1','x':2,'y':0,'net_name':'GND','layers':['*.Cu'],'type':'thru_hole'},
                    {'number':'2','x':0,'y':0,'net_name':'VCC','layers':['F.Cu'],'type':'smd'}]},
            {'reference':'J2','layer':'B.Cu','position':{'x':30,'y':20},
             'pads':[{'number':'1','x':0,'y':0,'net_name':'GND','layers':['B.Cu'],'type':'smd'}]}]}


class DrcSnapshot(unittest.TestCase):
    def test_capped_opens_are_lower_bound_without_blocking_error_comparison(self):
        limits = live._report_limits([], [{}] * 499)
        self.assertTrue(limits['unconnectedCountIsLowerBound'])
        self.assertFalse(limits['violationReportMayBeTruncated'])
        self.assertEqual(live._new_errors({'violations': [], **limits}, {'violations': [], **limits}), [])
        self.assertFalse(live._report_limits([], [{}] * 498)['unconnectedCountIsLowerBound'])

    def test_per_type_caps_block_error_delta_even_when_reports_match(self):
        for kind, cap in [('clearance', 499), ('shorting_items', 199)]:
            v = {'type': kind, 'severity': 'error', 'items': []}
            self.assertFalse(live._report_limits([v] * (cap - 1), [])['violationReportMayBeTruncated'])
            report = {'violations': [v] * cap, **live._report_limits([v] * cap, [])}
            with self.assertRaises(live.RoutingError) as cm:
                live._new_errors(report, report)
            self.assertEqual(cm.exception.code, 'drc_incomplete')
            self.assertFalse(cm.exception.detail['mutated'])

    def test_refills_snapshot_with_project_rules_without_changing_source(self):
        import json
        from types import SimpleNamespace
        with tempfile.TemporaryDirectory() as tmp:
            source = Path(tmp) / 'board.kicad_pcb'
            source.write_text('original board')
            source.with_suffix('.kicad_pro').write_text('project rules')
            source.with_suffix('.kicad_dru').write_text('custom rules')
            exe = Path(tmp) / 'kicad-cli'
            exe.touch()
            def run(command, **kwargs):
                candidate = Path(command[-1])
                self.assertNotEqual(candidate, source)
                self.assertEqual(candidate.read_text(), 'live snapshot')
                self.assertEqual(candidate.with_suffix('.kicad_pro').read_text(), 'project rules')
                self.assertEqual(candidate.with_suffix('.kicad_dru').read_text(), 'custom rules')
                self.assertIn('--refill-zones', command)
                Path(command[command.index('--output') + 1]).write_text(json.dumps({
                    'violations': [], 'unconnected_items': [{'items': []}]}))
                return SimpleNamespace(returncode=0, stderr='', stdout='')
            with patch.object(live.subprocess, 'run', side_effect=run):
                result = live._drc({'kicad_cli_exe': str(exe)}, {'filePath': str(source)}, 'live snapshot')
            self.assertTrue(result['zonesRefilled'])
            self.assertEqual(result['unconnected'], 1)
            self.assertFalse(result['clean'])
            self.assertEqual(source.read_text(), 'original board')


class Planning(unittest.TestCase):
    def test_rotated_pad_coordinates_and_final_layer_transition(self):
        p=live._plan(board(), {'net':'GND','points':['J1.1',{'pad':'J2.1','layer':'B.Cu'}]})
        self.assertEqual(p['segments'][0]['start'],[10,18])
        self.assertEqual(p['vias'][0]['at'],[30,20])
        self.assertEqual(p['vias'][0]['layers'],['F.Cu','B.Cu'])

    def test_branches_on_one_net(self):
        p=live._plan(board(), {'net':'GND','paths':[['J1.1',[20,18]],[[20,18],[20,25]]]})
        self.assertEqual(len(p['segments']),2)

    def test_reject_wrong_net_and_layer(self):
        for args,code in [({'net':'GND','points':['J1.1','J1.2']},'net_mismatch'),
                          ({'net':'GND','points':['J1.1','J2.1']},'pad_layer_mismatch'),
                          ({'net':'GND','points':['J1.1',{'x':20,'y':20,'layer':'F.SilkS'}]},'unknown_layer')]:
            with self.subTest(code=code),self.assertRaises(live.RoutingError) as cm: live._plan(board(),args)
            self.assertEqual(cm.exception.code,code)

    def test_invalid_geometry_is_rejected_before_mutation(self):
        for args in [{'width':0},{'width':-1},{'width':float('nan')},{'width':True},{'points':[[True,0],[1,1]]},{'points':[[0,0,9],[1,1]]},
                     {'viaDrill':0.8,'viaSize':0.6},{'points':[[0,0],[float('inf'),0]]},
                     {'points':[[0,0],[0,0]]}]:
            with self.subTest(args=args),self.assertRaises(live.RoutingError):
                live._plan(board(),{'net':'GND','points':[[0,0],[1,1]],**args})

    def test_numeric_net_name_in_kicad_10_file(self):
        from parsers.pcb import parse_pcb_text
        data=parse_pcb_text('(kicad_pcb (version 20260206) (footprint "X" (layer "F.Cu") (pad "1" thru_hole circle (net "123"))))')
        self.assertEqual(data["net_format"],"name")
        self.assertEqual(data["nets"],[{"number":None,"name":"123"}])

    def test_exact_numeric_net_name(self):
        self.assertEqual(live._plan(board(),{'net':'123','points':[[0,0],[1,1]]})['netName'],'123')

    def test_quoted_net_serialization(self):
        self.assertIn('"a\\"b\\\\c"',route._net_sexpr({'format':'name','name':'a"b\\c'}))

    def test_stale_revision(self):
        for rev in [None,'old']:
            with self.assertRaises(live.RoutingError): live._revision({'expectedRevision':rev},'new')

    def test_existing_error_does_not_mask_additional_same_error(self):
        v={'type':'clearance','severity':'error','description':'collision','items':[]}
        self.assertEqual(live._new_errors({'violations':[v]},{'violations':[v,v]}),[v])

    def test_open_file_refused_even_with_private_arg(self):
        with patch.object(route,'_lock_file',return_value=Path('lock')):
            self.assertEqual(route._load({'filePath':'test.kicad_pcb','_readOnly':True})[-1]['errorCode'],'board_open')


class Transactions(unittest.TestCase):
    class Board:
        def __init__(self): self.ops=[]
        def begin_commit(self): self.ops.append('begin'); return 'commit'
        def push_commit(self,c,m): self.ops.append('push')
        def drop_commit(self,c): self.ops.append('drop')

    def test_success_is_one_undo_step(self):
        b=self.Board()
        self.assertEqual(live._transaction(b,lambda:42,'route'),42)
        self.assertEqual(b.ops,['begin','push'])

    def test_partial_create_failure_rolls_back(self):
        b=self.Board()
        def fail(): raise RuntimeError('partial create')
        with self.assertRaises(live.RoutingError) as cm: live._transaction(b,fail,'route')
        self.assertEqual(cm.exception.code,'mutation_rolled_back')
        self.assertEqual(b.ops,['begin','drop'])

    def test_push_timeout_is_unknown_even_if_drop_acknowledged(self):
        b = self.Board()
        def push(*args): raise TimeoutError('commit response lost')
        b.push_commit = push
        with self.assertRaises(live.RoutingError) as cm:
            live._transaction(b, lambda: 42, 'route')
        self.assertEqual(cm.exception.code, 'mutation_outcome_unknown')
        self.assertIsNone(cm.exception.detail['mutated'])

    def test_failed_rollback_is_unknown_not_safe_to_retry(self):
        b=self.Board()
        def fail(*a): raise RuntimeError('lost IPC')
        b.drop_commit=fail
        with self.assertRaises(live.RoutingError) as cm: live._transaction(b,fail,'route')
        self.assertEqual(cm.exception.code,'mutation_outcome_unknown')
        self.assertIsNone(cm.exception.detail['mutated'])


if __name__=='__main__': unittest.main()