KiCad - the KiCad Bridge

"""Exercise public handlers with real kipy copper types and a simulated editor.

IPC transport and DRC execution are mocked here; Windows live tests remain required.
"""
from contextlib import contextmanager
from pathlib import Path
from types import SimpleNamespace
import tempfile
import unittest
from unittest.mock import patch
import sys
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from handlers import live_routing as live, route
from demo.routing.make_fixture import make
try:
    from kipy.board_types import Net
except ImportError:
    Net = None


@unittest.skipIf(Net is None, 'Install requirements-routing.txt for protocol tests')
class IPCHandlers(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.path = make(self.tmp.name)
        text = self.path.read_text()
        self.args = {'filePath': str(self.path), 'net': 'NET_1', 'points': ['J1.1', 'J2.1'], 'validate': False}
        class Board:
            def __init__(b):
                b.text = text; b.items = []; b.ops = []; b.partial = False; b.fail_read = False; b.fail_save = False; b.keep_removed = False
            def get_as_string(b):
                if b.fail_read: raise RuntimeError('read failed after commit')
                return b.text
            def get_nets(b):
                n = Net(); n.name = 'NET_1'; return [n]
            def begin_commit(b): b.ops.append('begin'); return 'commit'
            def create_items(b, items):
                for i, item in enumerate(items): item.id.value = 'created-'+str(i)
                b.items += items
                return [] if b.partial else items
            def push_commit(b, *_): b.ops.append('push'); b.text += '\n'
            def drop_commit(b, *_): b.ops.append('drop'); b.items = []
            def save(b):
                if b.fail_save: raise RuntimeError('save failed')
                b.ops.append('save')
            def get_tracks(b): return b.items
            def get_vias(b): return []
            def remove_items(b, items):
                if not b.keep_removed: b.items = []
        self.board = Board()
        @contextmanager
        def connect(args): yield self.board
        self.patch = patch.object(live, '_connect', connect); self.patch.start()
        self.args['expectedRevision'] = live._snapshot(self.board, self.args)[2]

    def tearDown(self): self.patch.stop(); self.tmp.cleanup()

    def test_copper_protocol_units_net_and_layer(self):
        result = live.handle_route_net({}, self.args)
        self.assertTrue(result['success'], result)
        self.assertEqual(self.board.items[0].width, 250000)
        self.assertEqual(self.board.items[0].start.x, 105000000)
        self.assertEqual(self.board.items[0].net.name, 'NET_1')
        self.assertEqual(self.board.ops, ['begin', 'push'])
        self.assertNotEqual(result['revision'], self.args['expectedRevision'])
        self.assertFalse(result['saved'])

    def test_dry_run_does_not_start_transaction(self):
        result = live.handle_route_net({}, {**self.args, 'dryRun': True})
        self.assertTrue(result['success'], result)
        self.assertEqual(self.board.ops, [])

    def test_drc_rejection_and_edit_during_preflight_do_not_mutate(self):
        ok = {'violations': []}
        bad = {'violations': [{'severity': 'error', 'type': 'clearance'}]}
        with patch.object(live, '_drc', side_effect=[ok, bad]):
            r = live.handle_route_net({}, {**self.args, 'validate': True})
        self.assertEqual(r['errorCode'], 'drc_rejected')
        def drc(*a): self.board.text += '\n'; return ok
        with patch.object(live, '_drc', side_effect=drc):
            r = live.handle_route_net({}, {**self.args, 'validate': True})
        self.assertEqual(r['errorCode'], 'stale_board')
        self.assertEqual(self.board.ops, [])

    def test_partial_create_is_rolled_back(self):
        self.board.partial = True
        r = live.handle_route_net({}, self.args)
        self.assertEqual(r['errorCode'], 'mutation_rolled_back')
        self.assertEqual(self.board.ops, ['begin', 'drop'])
        self.assertFalse(self.board.items)

    def test_post_commit_save_failure_is_not_failed_mutation(self):
        self.board.fail_save = True
        r = live.handle_route_net({}, {**self.args, 'save': True})
        self.assertTrue(r['success'], r); self.assertTrue(r['mutated'])
        self.assertFalse(r['saved']); self.assertIn('save failed', r['postCommitError'])
        self.assertEqual(self.board.ops, ['begin', 'push'])

    def test_replay_rejected(self):
        self.assertTrue(live.handle_route_net({}, self.args)['success'])
        self.assertEqual(live.handle_route_net({}, self.args)['errorCode'], 'stale_board')
        self.assertEqual(self.board.ops, ['begin', 'push'])

    def test_remove_readback_detects_ignored_deletion(self):
        r = live.handle_route_net({}, self.args)
        self.board.keep_removed = True
        with patch.object(live, '_delete_items', side_effect=RuntimeError('deletion rejected')):
            r = live.handle_remove_route({}, {**self.args, 'expectedRevision': r['revision'], 'itemIds': r['itemIds']})
        self.assertEqual(r['errorCode'], 'mutation_rolled_back')
        self.assertEqual(self.board.ops[-2:], ['begin', 'drop'])

    def test_deletion_protocol_requires_each_id_accepted(self):
        from kipy.proto.common.commands.editor_commands_pb2 import DeleteItemsResponse, IDS_OK, IDS_IMMUTABLE
        from kipy.proto.common.types.base_types_pb2 import DocumentSpecifier, IRS_OK
        live.handle_route_net({}, self.args)
        response = DeleteItemsResponse(status=IRS_OK)
        row = response.deleted_items.add(); row.id.CopyFrom(self.board.items[0].id); row.status = IDS_OK
        self.board._doc = DocumentSpecifier()
        self.board._kicad = SimpleNamespace(send=lambda *a: response)
        live._delete_items(self.board, self.board.items)
        row.status = IDS_IMMUTABLE
        with self.assertRaises(RuntimeError): live._delete_items(self.board, self.board.items)
        del response.deleted_items[:]
        live._delete_items(self.board, self.board.items)  # KiCad 10.0.5 compatibility
        response.status = 0
        with self.assertRaises(RuntimeError): live._delete_items(self.board, self.board.items)

    def test_wrong_or_locked_item_never_deleted(self):
        r = live.handle_remove_route({}, {**self.args, 'itemIds': ['not-copper']})
        self.assertEqual(r['errorCode'], 'invalid_item_ids')
        self.assertEqual(self.board.ops, [])


class OfflineTests(unittest.TestCase):
    def test_numeric_net_name_and_fractional_code(self):
        b = {'nets': [{'number': 1, 'name': '123'}], 'net_format': 'number'}
        self.assertEqual(route._resolve_net(b, '123')[0]['number'], 1)
        for spec in [1.5, float('nan'), True, '123 '.strip()+'x']:
            self.assertIsNotNone(route._resolve_net(b, spec)[1])

    def test_atomic_file_edit_and_open_board_refusal(self):
        with tempfile.TemporaryDirectory() as d:
            # A user can have the demonstration's default "live-routing" board
            # open on the Windows test host. Give this disposable fixture a
            # unique title so the real open-editor guard does not match it.
            fixture = make(d)
            path = fixture.with_name(Path(d).name + '-offline-test.kicad_pcb')
            fixture.rename(path)
            before = path.read_bytes()
            r = route.handle_route({}, {'filePath':str(path),'net':'NET_1','points':['J1.1','J2.1']})
            self.assertTrue(r['success'], r)
            self.assertEqual(Path(str(path)+'.adom-bak').read_bytes(), before)
            path.with_name('~'+path.name+'.lck').write_text('open')
            after = path.read_bytes()
            r = route.handle_add_track({}, {'filePath':str(path),'net':'NET_1','start':[1,1],'end':[2,2]})
            self.assertEqual(r['errorCode'], 'board_open')
            self.assertEqual(path.read_bytes(), after)