AI Flow

"""Unit tests for tools/fusion-project.py helpers (no Fusion, no bridge): python3 -m unittest tools/test_fusion_project.py"""
import importlib.util, os, re, sys, tempfile, unittest, unittest.mock
import xml.etree.ElementTree as ET
from pathlib import Path

os.environ.setdefault('AIFLOW_TARGET', 'test-box')  # no bridge lookup at import
os.environ.setdefault('AIFLOW_RUN', tempfile.mkdtemp(prefix='aiflow-fusion-test-'))
os.environ.setdefault('AIFLOW_DASHBOARD', 'off')
_spec = importlib.util.spec_from_file_location('fusion_project', Path(__file__).with_name('fusion-project.py'))
fp = importlib.util.module_from_spec(_spec); _spec.loader.exec_module(fp)

# a smashed part whose NAME was deleted is written self-closing; the next part is an ordinary smashed one
BOARD = '''<?xml version="1.0" encoding="utf-8"?>
<!DOCTYPE eagle SYSTEM "eagle.dtd">
<eagle version="9.7"><drawing><board>
<plain><wire x1="0" y1="0" x2="20" y2="0" width="0" layer="20"/><text x="3" y="4" size="0.8" layer="21">R1</text></plain>
<libraries><library name="L"><packages><package name="P"><smd name="1" x="-0.5" y="0" dx="0.5" dy="0.6" layer="1"/><smd name="2" x="0.5" y="0" dx="0.5" dy="0.6" layer="1"/><text x="0" y="1" size="0.8" layer="25">&gt;NAME</text></package></packages></library></libraries>
<elements>
<element name="R1" library="L" package="P" value="10k" x="5" y="5" smashed="yes"/>
<element name="R2" library="L" package="P" value="1k" x="10" y="5" smashed="yes">
<attribute name="NAME" x="10" y="6.5" size="0.8" layer="25"/>
</element>
</elements><signals/></board></drawing></eagle>'''


class Nameless(unittest.TestCase):
    def test_self_closing_smashed_part_is_nameless(self):
        self.assertEqual(fp.nameless_parts(BOARD), {'R1'})
        # the regex it replaces ran from R1's '/>' on into R2 and read R2's NAME as R1's
        old = {m.group(1) for m in re.finditer(r'<element name="([^"]+)"[^>]*smashed="yes"[^>]*>(.*?)</element>', BOARD, re.S) if 'name="NAME"' not in m.group(2)}
        self.assertNotIn('R1', old)

    def test_name_positions_read_back_per_element(self):
        e = fp.board_elements(BOARD)
        self.assertEqual(e['R1'], {'smashed': True, 'name': None, 'hasName': False})
        self.assertEqual(e['R2']['name'], (10.0, 6.5))
        self.assertEqual(fp.plain_texts(BOARD), [('R1', 3.0, 4.0)])
        self.assertEqual(fp.board_elements('not xml'), {})

    def test_board_model_keeps_both_parts_apart(self):
        els = fp.board_model(BOARD)[0]
        self.assertEqual([e['name'] for e in els], ['R1', 'R2'])
        self.assertIsNone(els[0]['nameAttr'])
        self.assertIn('y="6.5"', els[1]['nameAttr'])


class WiredLayoutPins(unittest.TestCase):
    """EAGLE's rotation string: rotate k*90 counterclockwise, then mirror x (MR90 turns clockwise as seen
    from the front). Checked against a real Fusion schematic: the wires on an MR270 MOSFET's three pins end
    where this order puts them, and the gate pin only there."""
    LIB = ET.fromstring('<library><symbols><symbol name="S"><pin name="A" x="-2.54" y="1.27"/><pin name="B" x="2.54" y="0"/></symbol></symbols>'
                        '<devicesets><deviceset name="DS"><gates><gate name="G" symbol="S"/></gates><devices><device name=""><connects>'
                        '<connect gate="G" pin="A" pad="1"/><connect gate="G" pin="B" pad="2"/></connects></device></devices></deviceset></devicesets></library>')

    def pin1(self, rot):
        layout = {'groups': [{'name': 'g', 'parts': {'U1': [10, 10, rot]}, 'nets': {'N': [['U1.1', [0, 0]]]}}]}
        cmds = fp.wired_groups(layout, self.LIB, {'U1': 'DS'}, {}, 'lib', set())[0]['commands']
        net = next(c for c in cmds if c.startswith("NET 'N'"))
        x, y = re.findall(r'\(([-\d.]+)mm ([-\d.]+)mm\)', net)[0]
        return float(x), float(y)

    def test_pin_positions(self):
        want = {'R0': (7.46, 11.27), 'R90': (8.73, 7.46), 'MR0': (12.54, 11.27), 'MR90': (11.27, 7.46), 'MR270': (8.73, 12.54)}
        for rot, (x, y) in want.items():
            gx, gy = self.pin1(rot)
            self.assertAlmostEqual(gx, x, places=3, msg=rot); self.assertAlmostEqual(gy, y, places=3, msg=rot)
            # the board model's transform is the same order
            ang, mir = fp._rot_of(rot)
            bx, by = fp._xf(-2.54, 1.27, 10, 10, ang, mir)
            self.assertAlmostEqual(bx, x, places=6, msg=rot); self.assertAlmostEqual(by, y, places=6, msg=rot)


class CloudSave(unittest.TestCase):
    def test_save_as_is_refused_on_a_saved_design(self):
        fresh = {'data': {'activeDocument': 'Untitled', 'documents': [{'name': 'Untitled', 'isActive': True}, {'name': 'Other v2', 'dataFile': {'name': 'Other'}}]}}
        self.assertIsNone(fp.save_refusal(fresh, 'Buck'))
        board_in_front = {'data': {'activeDocument': 'Buck v3', 'documents': [{'name': 'Buck v3', 'isActive': True, 'dataFile': {'name': 'Buck'}}]}}
        self.assertIn('fork', fp.save_refusal(board_in_front, 'Buck'))
        open_elsewhere = {'data': {'activeDocument': 'Untitled', 'documents': [{'name': 'Untitled', 'isActive': True}, {'name': 'Buck v3', 'dataFile': {'name': 'Buck'}}]}}
        self.assertIn('second', fp.save_refusal(open_elsewhere, 'Buck'))

    def test_copies_by_exact_name(self):
        listing = {'success': True, 'data': {'files': [{'name': 'Buck', 'fileExtension': 'fprj'}, {'name': 'Buck', 'fileExtension': 'fbrd'}, {'name': 'Buck2', 'fileExtension': 'fbrd'}]}}
        self.assertEqual([f['fileExtension'] for f in fp.cloud_copies(listing, 'Buck')], ['fprj', 'fbrd'])

    def test_cloud_home(self):
        class A: project = None; folder = None
        real = fp.ab; fp._PROJECTS.clear()
        fp.ab = lambda verb, args, timeout=600: {'hubs': [{'projects': [{'name': n} for n in ('Main', 'Claude AI', 'P', 'Q')]}]}
        try:
            self.assertEqual(fp.cloud_place(A(), {}), ('Main', 'Adom AI Flow'))
            self.assertEqual(fp.cloud_place(A(), {'cloud': {'project': 'Claude AI', 'folder': 'Adom AI Flow'}}), ('Claude AI', 'Adom AI Flow'))
            a = A(); a.project = 'P'
            self.assertEqual(fp.cloud_place(a, {'cloud': {'project': 'Q', 'folder': 'F'}}), ('P', 'F'))
        finally:
            fp.ab = real; fp._PROJECTS.clear()


class LibraryLineBreaks(unittest.TestCase):
    """adom-aiflow#114 / fusion-bridge#135: Fusion 2706 cannot ADD from a .lbr whose elements share one line."""
    COMPACT = ('<?xml version="1.0" encoding="utf-8"?>\n<!DOCTYPE eagle SYSTEM "eagle.dtd">\n<eagle version="9.7"><drawing><library name="P">'
               '<packages><package name="R0402"><description>a &lt;b&gt;0402&lt;/b&gt; resistor</description>'
               '<smd name="1" x="-0.5" y="0" dx="0.5" dy="0.6" layer="1"/><text x="0" y="1" size="0.8" layer="25">&gt;NAME</text>'
               '<text x="0" y="-1" size="0.8" layer="27"></text></package></packages>'
               '<packages3d><package3d name="R0402" urn="" wip_urn="u" locally_modified="yes" type="model"><packageinstances><packageinstance name="R0402"/></packageinstances></package3d></packages3d>'
               '<devicesets><deviceset name="R" prefix="R"><gates><gate name="G$1" symbol="R" x="0" y="0"/></gates>'
               '<devices><device name="" package="R0402"><connects><connect gate="G$1" pin="1" pad="1"/></connects>'
               '<technologies><technology name=""><attribute name="NOTE" value="a > b"/></technology></technologies></device></devices></deviceset></devicesets>'
               '</library></drawing></eagle>')

    @staticmethod
    def canon(text):
        def walk(e):
            return (e.tag, sorted(e.attrib.items()), (e.text or '').strip() if len(e) else (e.text or ''), [walk(c) for c in e])
        return walk(ET.fromstring(text.split('?>', 1)[1].split('>', 1)[1] if text.startswith('<?xml') else text))

    def test_every_element_on_its_own_line_and_nothing_else_changes(self):
        out = fp.break_lines(self.COMPACT)
        self.assertNotIn('><', out.replace('></text>', ''))  # the only adjacent pair left is an empty element's own open and close
        self.assertIn('<text x="0" y="-1" size="0.8" layer="27"></text>', out)  # empty text stays empty
        self.assertIn('layer="25">&gt;NAME</text>', out)
        self.assertIn('value="a > b"', out)
        self.assertTrue(out.startswith('<?xml version="1.0" encoding="utf-8"?>\n<!DOCTYPE eagle SYSTEM "eagle.dtd">\n<eagle'))
        self.assertEqual(self.canon(out), self.canon(self.COMPACT))
        self.assertEqual(fp.break_lines(out), out)  # idempotent: a library already broken is left as it is

    def test_a_raw_gt_in_text_is_not_split(self):
        t = '<a><text>></text><b/></a>'
        self.assertEqual(fp.break_lines(t), '<a>\n<text>></text>\n<b/>\n</a>')


class SchematicFailures(unittest.TestCase):
    def test_failure_codes_name_the_part_the_cause_and_the_reply(self):
        import io, contextlib
        for code, extra, want in (('add_failed', {'part': 'R7', 'failedAt': 3}, 'ADD of R7'),
                                  ('fusion_assert', {'fusionAsserts': ['PlacePartWorker.cpp, Line 23. Condition: assetAdded']}, '1.13.47'),
                                  ('no_3d_package', {}, '1.13.47')):
            buf = io.StringIO()
            with contextlib.redirect_stdout(buf), self.assertRaises(SystemExit):
                fp.schematic_failure({'group': 'power', 'parts': []}, {'success': False, 'errorCode': code, 'error': 'x', **extra}, Path('/run/fusion/schematic/run-01-power.json'))
            self.assertIn(want, buf.getvalue()); self.assertIn('run-01-power.json', buf.getvalue())

    def test_replies_are_kept_whole_in_the_run(self):
        d = Path(tempfile.mkdtemp()); old = fp.STATE; fp.STATE = d
        try:
            f = fp.keep_reply('run-01-power supply', {'reply': {'success': False, 'errorCode': 'add_failed'}})
            self.assertEqual(f.name, 'run-01-power-supply.json')
            import json as _j
            self.assertEqual(_j.loads(f.read_text())['reply']['errorCode'], 'add_failed')
        finally:
            fp.STATE = old


class CloudHome(unittest.TestCase):
    """fusion-bridge#136: a user's hub need not have a project called Main."""
    def run_place(self, reply, project=None, folder=None, remembered=None):
        import io, contextlib, types
        real = fp.ab; calls = []
        fp.ab = lambda verb, args, timeout=600: calls.append(verb) or reply
        fp._PROJECTS.clear()
        st = {'cloud': remembered} if remembered else {}
        buf = io.StringIO()
        try:
            with contextlib.redirect_stdout(buf):
                out = fp.cloud_place(types.SimpleNamespace(project=project, folder=folder), st)
            return out, st, buf.getvalue(), calls
        except SystemExit:
            return None, st, buf.getvalue(), calls
        finally:
            fp.ab = real; fp._PROJECTS.clear()

    @staticmethod
    def hubs(*names, active='(lookup failed)', timed_out=False):
        return {'data': {'activeProject': active, 'listTimedOut': timed_out, 'hubs': [{'name': 'H', 'projects': [{'name': n} for n in names]}]}}

    def test_main_stays_the_default_where_it_exists(self):
        out, st, _, _ = self.run_place(self.hubs('Personal', 'MAIN'))
        self.assertEqual(out, ('MAIN', 'Adom AI Flow'))
        self.assertEqual(st['cloud'], {'project': 'MAIN', 'folder': 'Adom AI Flow'})

    def test_no_main_takes_the_open_project_or_the_only_one(self):
        self.assertEqual(self.run_place(self.hubs('A', 'B', active='B'))[0][0], 'B')
        self.assertEqual(self.run_place(self.hubs('Only'))[0][0], 'Only')

    def test_no_way_to_tell_asks_and_names_them(self):
        out, _, text, _ = self.run_place(self.hubs('A', 'B'))
        self.assertIsNone(out)
        self.assertIn('A, B', text); self.assertIn('--project', text)

    def test_a_remembered_project_the_account_lacks_is_chosen_again(self):
        # Andy's rerun: the failed first run had saved Main in its state
        out, st, text, _ = self.run_place(self.hubs('Only'), remembered={'project': 'Main', 'folder': 'F'})
        self.assertEqual(out, ('Only', 'F'))
        self.assertIn('does not have', text)

    def test_an_asked_project_the_account_lacks_is_refused_with_the_list(self):
        out, _, text, _ = self.run_place(self.hubs('A', 'B', 'Main'), project='Nope')
        self.assertIsNone(out)
        self.assertIn('Nope', text); self.assertIn('A, B, Main', text)

    def test_an_asked_or_remembered_name_stands_when_the_list_cannot_be_read(self):
        self.assertEqual(self.run_place({'success': False, 'error': 'busy'}, project='X')[0][0], 'X')
        self.assertEqual(self.run_place(self.hubs('A', timed_out=True), remembered={'project': 'Y', 'folder': 'F'})[0][0], 'Y')
        out, _, text, _ = self.run_place({'success': False, 'error': 'busy'})
        self.assertIsNone(out); self.assertIn('--project', text)

    def test_several_hubs_never_guess_main(self):
        two = {'hubs': [{'name': 'A', 'projects': [{'name': 'Main'}]}, {'name': 'B', 'projects': [{'name': 'Mine'}]}], 'activeProject': '(lookup failed)'}
        self.assertIsNone(self.run_place(two)[0])
        two['activeProject'] = 'Mine'
        self.assertEqual(self.run_place(two)[0][0], 'Mine')

    def test_the_list_is_read_once_per_command(self):
        import types
        real = fp.ab; calls = []
        fp.ab = lambda verb, args, timeout=600: calls.append(verb) or self.hubs('Main')
        fp._PROJECTS.clear()
        try:
            st = {}; a = types.SimpleNamespace(project=None, folder=None)
            fp.cloud_place(a, st); fp.cloud_place(a, st)
            self.assertEqual(calls, ['fusion_list_cloud_projects'])
        finally:
            fp.ab = real; fp._PROJECTS.clear()


class SharedFootprint3D(unittest.TestCase):
    """fusion-bridge#136: adom-lbr assemble merges equal footprints, so PARTB's device uses package PARTA.
    A <package3d> naming the merged-away PARTB made Fusion 2706 assert on every ADD."""
    LIB = ('<?xml version="1.0" encoding="utf-8"?>\n<!DOCTYPE eagle SYSTEM "eagle.dtd">\n<eagle version="9.7"><drawing><library>'
           '<packages><package name="PARTA"><smd name="1" x="0" y="0" dx="1" dy="1" layer="1"/></package><package name="SOLO"><smd name="1" x="0" y="0" dx="1" dy="1" layer="1"/></package></packages>'
           '<symbols><symbol name="S"><pin name="1" x="0" y="0"/></symbol></symbols>'
           '<devicesets>' + ''.join('<deviceset name="%s" uservalue="yes"><gates><gate name="G$1" symbol="S" x="0" y="0"/></gates><devices><device name="" package="%s"><connects><connect gate="G$1" pin="1" pad="1"/></connects><technologies><technology name=""/></technologies></device></devices></deviceset>' % (d, p)
                                    for d, p in (('PARTA', 'PARTA'), ('PARTB', 'PARTA'), ('PARTC', 'PARTA'), ('SOLO', 'SOLO'))) +
           '</devicesets></library></drawing></eagle>')
    URNS = {'PARTA': 'urn:a', 'PARTB': 'urn:b', 'SOLO': 'urn:s', 'GONE': 'urn:gone'}  # PARTC has no model

    def device_urns(self, t):
        return {ds.get('name'): [i.get('package3d_urn') for i in ds.iter('package3dinstance')] for ds in ET.fromstring(t).iter('deviceset')}

    def test_each_part_keeps_its_own_model_on_the_package_that_survived(self):
        t = fp.bind_library_3d(self.LIB, self.URNS)
        self.assertEqual(fp.library_3d_problems(t), [])
        self.assertEqual(self.device_urns(t), {'PARTA': ['urn:a'], 'PARTB': ['urn:b'], 'PARTC': [], 'SOLO': ['urn:s']})
        models = {q.get('wip_urn'): (q.get('name'), [i.get('name') for i in q.iter('packageinstance')]) for q in ET.fromstring(t).iter('package3d')}
        self.assertEqual(models, {'urn:a': ('PARTA', ['PARTA']), 'urn:b': ('PARTB', ['PARTA']), 'urn:s': ('SOLO', ['SOLO'])})

    def test_binding_twice_is_the_same_as_once(self):
        once = fp.bind_library_3d(self.LIB, self.URNS)
        self.assertEqual(fp.bind_library_3d(once, self.URNS), once)

    def test_the_bug_is_caught(self):
        dead = fp.bind_library_3d(self.LIB, self.URNS).replace('</packages3d>', '<package3d name="PARTB" urn="" wip_urn="urn:x" locally_modified="yes" type="model"><packageinstances><packageinstance name="PARTB"/></packageinstances></package3d></packages3d>')
        self.assertTrue(any('PARTB, which is not in the library' in x for x in fp.library_3d_problems(dead)))
        wrong = fp.bind_library_3d(self.LIB, self.URNS).replace('package3d_urn="urn:b"', 'package3d_urn="urn:s"')
        self.assertTrue(any('PARTB uses a 3D model' in x for x in fp.library_3d_problems(wrong)))

    def test_board_restore_goes_by_part_not_by_package(self):
        d = Path(tempfile.mkdtemp()); old = (fp.STATE, fp.ab, fp.send, fp.box_dir, fp.time.sleep)
        lib = '<library name="Proj"><packages><package name="PARTA"><smd name="1" x="0" y="0" dx="1" dy="1" layer="1"/></package></packages>%s</library>'
        (d / 'restore3d').mkdir()
        (d / 'restore3d' / 'D-src.brd').write_text('<eagle><drawing><board><libraries>' + lib % '' + '</libraries><elements><element name="R1" library="Proj" package="PARTA"/><element name="R2" library="Proj" package="PARTA"/><element name="R3" library="Proj" package="PARTA"/></elements></board></drawing></eagle>')
        (d / 'restore3d' / 'D-src.sch').write_text('<eagle><drawing><schematic><libraries>' + lib % '<devicesets><deviceset name="PARTA"><devices><device name="" package="PARTA"><connects/></device></devices></deviceset><deviceset name="PARTB"><devices><device name="" package="PARTA"><connects/></device></devices></deviceset></devicesets>' + '</libraries><parts><part name="R1" library="Proj" deviceset="PARTA" device=""/><part name="R2" library="Proj" deviceset="PARTB" device=""/></parts></schematic></drawing></eagle>')
        fp.STATE = d; fp.send = lambda *a, **k: None; fp.box_dir = lambda: 'C:/x'; fp.time.sleep = lambda s: None
        fp.ab = lambda verb, args, timeout=600: {'savedTo': 'x', 'success': True, 'runId': 'r', 'stage': 'done', 'productType': 'EcadDesignProductType'}
        try:
            got = fp.restore_3d({'library': 'Proj', 'parts': {'PARTA': {'urn': 'urn:a'}, 'PARTB': {'urn': 'urn:b'}}}, 'D', 'P', 'F', by_ref={'R1': 'PARTA', 'R2': 'PARTB', 'R3': 'NOMODEL'})
        finally:
            fp.STATE, fp.ab, fp.send, fp.box_dir, fp.time.sleep = old
        self.assertEqual(got, ['R1', 'R2'])  # R3's part has no model, though its footprint is named after one that does
        brd = ET.parse(d / 'restore3d' / 'D.brd').getroot()
        self.assertEqual({e.get('name'): e.get('package3d_urn') for e in brd.iter('element')}, {'R1': 'urn:a', 'R2': 'urn:b', 'R3': None})
        self.assertEqual(fp.library_3d_problems(ET.tostring(brd, encoding='unicode')), [])
        sch = ET.parse(d / 'restore3d' / 'D.sch').getroot()
        self.assertEqual({e.get('name'): e.get('package3d_urn') for e in sch.iter('part')}, {'R1': 'urn:a', 'R2': 'urn:b'})
        self.assertEqual(fp.library_3d_problems(ET.tostring(sch, encoding='unicode')), [])


class CarriedTestPoint(unittest.TestCase):
    """Issue #143: the test point still works for an account that cannot read its private page."""
    def test_only_the_test_point_page_falls_back_and_only_when_the_files_are_there(self):
        with tempfile.TemporaryDirectory() as d:
            self.assertIsNone(fp.carried_part('adom/testpoint-pad-0-5mm', d))
            (Path(d) / 'testpoint').mkdir(); (Path(d) / 'testpoint' / 'page.json').write_text('{}')
            self.assertEqual(fp.carried_part('adom/testpoint-pad-0-5mm', d), Path(d) / 'testpoint')
            self.assertIsNone(fp.carried_part('adom/some-other-part', d))


class GerberLayers(unittest.TestCase):
    """Issue #144: the board's own copper count goes with the export and the zip is checked against it."""
    four = ['b.gtl', 'b.gbl', 'b.g1', 'b.g2', 'b.gts', 'b.gto', 'b.gko', 'b.xln']

    def test_the_count_comes_from_the_boards_stack(self):
        self.assertEqual(fp.copper_layer_count({'copperLayers': ['F.Cu', 'In1.Cu', 'In2.Cu', 'B.Cu']}), 4)
        self.assertEqual(fp.copper_layer_count({'data': {'copperLayers': ['F.Cu', 'B.Cu']}}), 2)
        self.assertIsNone(fp.copper_layer_count({'success': False}))

    def test_the_export_is_told_two_or_four_and_left_alone_otherwise(self):
        self.assertEqual(fp.gerber_args('x.zip', 4), {'outputPath': 'x.zip', 'layers': '4'})
        self.assertEqual(fp.gerber_args('x.zip', 2), {'outputPath': 'x.zip', 'layers': '2'})
        self.assertEqual(fp.gerber_args('x.zip', 6), {'outputPath': 'x.zip'})
        self.assertEqual(fp.gerber_args('x.zip', None), {'outputPath': 'x.zip'})

    def test_a_four_layer_board_with_outer_copper_only_is_a_problem(self):
        self.assertIsNone(fp.gerber_copper_problem(self.four, 4))
        self.assertIn('4 copper layers', fp.gerber_copper_problem(['b.gtl', 'b.gbl', 'b.gts', 'b.gto', 'b.gko', 'b.xln'], 4))

    def test_the_names_a_real_four_layer_export_writes_are_counted(self):
        names = [n.lower() for n in ['B_copper_l1.GTL', 'B_copper_l2.G1', 'B_copper_l3.G2', 'B_copper_l4.GBL', 'B_drill.XLN', 'B_Legend_Top.GTO', 'B_Profile_NP.GKO', 'B_Soldermask_Top.GTS']]
        self.assertIsNone(fp.gerber_copper_problem(names, 4))
        self.assertIn('carry 0 inner', fp.gerber_copper_problem([n for n in names if not n.endswith(('.g1', '.g2'))], 4))

    def test_two_layers_or_an_unknown_count_ask_nothing_of_the_zip(self):
        self.assertIsNone(fp.gerber_copper_problem(['b.gtl', 'b.gbl'], 2))
        self.assertIsNone(fp.gerber_copper_problem(['b.gtl', 'b.gbl'], None))


class LongRun(unittest.TestCase):
    """fusion_schematic_run past 300 commands can answer stillRunning with a runId and no nets yet (#137)."""
    def drive(self, replies, budget=None):
        calls, real = [], (fp.ab, fp.RUN_POLL_S, fp.RUN_BUDGET_S)
        def fake(verb, args, timeout=600):
            calls.append((verb, args))
            x = replies.pop(0)
            if isinstance(x, Exception): raise x
            return x
        fp.ab, fp.RUN_POLL_S = fake, 0
        if budget is not None: fp.RUN_BUDGET_S = budget
        try: return fp.run_group({'libraryPath': 'L', 'commands': ['ADD;'], 'pauseMs': 0}), calls
        finally: fp.ab, fp.RUN_POLL_S, fp.RUN_BUDGET_S = real

    def test_a_list_that_finishes_in_time_is_one_call(self):
        r, calls = self.drive([{'success': True, 'nets': {'A': ['R1.1']}}])
        self.assertEqual(r['nets'], {'A': ['R1.1']}); self.assertEqual(len(calls), 1)

    def test_it_polls_to_the_whole_reply_and_sends_the_list_once(self):
        r, calls = self.drive([{'success': True, 'stillRunning': True, 'runId': 'x'}, {'success': True, 'stillRunning': True, 'runId': 'x'},
                               {'success': True, 'stillRunning': False, 'runId': 'x', 'nets': {'A': ['R1.1']}}])
        self.assertEqual(r['nets'], {'A': ['R1.1']})
        self.assertEqual(len(calls), 3)
        self.assertIn('commands', calls[0][1])
        for _, args in calls[1:]: self.assertEqual(args, {'runId': 'x'})

    def test_a_failure_at_the_end_of_a_long_run_is_the_real_reply(self):
        r, _ = self.drive([{'success': True, 'stillRunning': True, 'runId': 'x'}, {'success': False, 'stillRunning': False, 'errorCode': 'add_failed', 'part': 'U1'}])
        self.assertEqual((r['success'], r['errorCode'], r['part']), (False, 'add_failed', 'U1'))

    def test_a_dropped_poll_does_not_end_the_run(self):
        r, _ = self.drive([{'success': True, 'stillRunning': True, 'runId': 'x'}, {'success': False, 'error': 'relay hiccup'},
                           fp.subprocess.TimeoutExpired('adom-bridge', 60), {'success': True, 'stillRunning': False, 'nets': {'A': []}}])
        self.assertEqual(r['nets'], {'A': []})

    def test_a_run_the_bridge_has_lost_stops_at_once(self):
        r, calls = self.drive([{'success': True, 'stillRunning': True, 'runId': 'x'}, {'success': False, 'errorCode': 'unknown_run_id', 'error': 'No schematic run x'}])
        self.assertEqual((r['success'], r['errorCode'], r['runId']), (False, 'run_lost', 'x')); self.assertEqual(len(calls), 2)

    def test_polls_that_never_answer_stop_without_resending(self):
        replies = [{'success': True, 'stillRunning': True, 'runId': 'x'}] + [{'success': False, 'error': 'no route'} for _ in range(fp.RUN_DROPPED_POLLS + 1)]
        r, calls = self.drive(replies)
        self.assertEqual(r['errorCode'], 'run_lost')
        self.assertEqual(sum('commands' in a for _, a in calls), 1)

    def test_past_the_budget_it_stops_and_names_the_run(self):
        r, calls = self.drive([{'success': True, 'stillRunning': True, 'runId': 'x'}], budget=-1)
        self.assertEqual((r['success'], r['errorCode'], r['runId']), (False, 'still_running', 'x')); self.assertEqual(len(calls), 1)


class BlankDesign(unittest.TestCase):
    """adom-aiflow#133: only a truly empty electronics design is reused, run against fake documents."""
    class Doc:
        def __init__(self, name, cls='EcadDesignDocument', ptype='EcadDesignProductType', cloud=False, mod=False, saved=False):
            self.name, self.cls, self.dataFile, self.isModified, self.isSaved = name, ('adsk::fusion::' if cls == 'FusionDocument' else 'adsk::electron::') + cls, (object() if cloud else None), mod, saved
            self.products = [type('P', (), {'productType': ptype})()]; self.activated = False
        def classType(self): return self.cls
        def activate(self): self.activated = True

    def run_with(self, docs):
        app = type('A', (), {'documents': type('D', (), {'count': len(docs), 'item': lambda s, i: docs[i]})()})()
        core = type('C', (), {'Application': type('X', (), {'get': staticmethod(lambda: app)})})
        adsk = type(sys)('adsk'); adsk.core = core
        with unittest.mock.patch.dict(sys.modules, {'adsk': adsk, 'adsk.core': core}):
            ns = {}; exec(fp.BLANK_DESIGN, ns)
        return ns['result']['reused'], [d for d in docs if d.activated]

    def test_an_empty_untitled_electronics_design_is_reused(self):
        d = self.Doc('Untitled')
        self.assertEqual(self.run_with([self.Doc('Untitled', 'FusionDocument', 'DesignProductType'), d]), (True, [d]))

    def test_anything_that_may_be_work_is_left_alone(self):
        for docs in ([self.Doc('Untitled', mod=True)], [self.Doc('Untitled', saved=True)], [self.Doc('MyBoard')], [self.Doc('Untitled', cloud=True)],
                     [self.Doc('Untitled', 'FusionDocument', 'DesignProductType')],
                     [self.Doc('Untitled'), self.Doc('Untitled', 'SchematicDocument', 'SchematicProductType')],
                     [self.Doc('Untitled'), self.Doc('Untitled', 'BoardDocument', 'BoardProductType', mod=True)]):
            self.assertEqual(self.run_with(docs), (False, []), [d.name + d.cls for d in docs])


class SheetLimit(unittest.TestCase):
    """adom-aiflow#133: a big block stays inside Fusion's sheet."""
    SIZES = {'R': (15.0, 2), 'BIG': (60.0, 150)}

    def parts(self, n, block='power', ds='R', start=1):
        return [{'ref': 'R%d' % (start + i), 'deviceset': ds, 'block': block} for i in range(n)]

    def test_a_block_that_fits_is_untouched(self):
        ps = self.parts(5) + self.parts(4, 'io', start=6)
        got, split = fp.split_blocks(ps, self.SIZES)
        self.assertEqual(got, ps); self.assertEqual(split, [])

    def test_a_wide_block_becomes_frames_that_fit_in_netlist_order(self):
        ps = self.parts(80, None)
        got, split = fp.split_blocks(ps, self.SIZES)
        self.assertEqual([q['ref'] for q in got], [q['ref'] for q in ps], 'nothing lost or reordered')
        self.assertEqual(split[0][:2], ('parts', 80)); frames = split[0][2]; self.assertGreater(frames, 1)
        names = list(dict.fromkeys(q['block'] for q in got))
        self.assertEqual(names, ['parts %d of %d' % (i + 1, frames) for i in range(frames)])
        for b in names:
            w = 7.62 + sum(self.SIZES['R'][0] + 25.4 for q in got if q['block'] == b)
            self.assertLessEqual(w, fp.FRAME_MAX_MM)
        # the planner's row: 3 of the widest frames, its gaps, its start and the fixed-parts shift
        self.assertLess(20.32 + 45.72 + 3 * fp.FRAME_MAX_MM + 2 * 10.16, fp.SHEET_LIMIT_MM)

    def test_a_block_with_too_many_commands_is_cut_even_when_narrow(self):
        got, split = fp.split_blocks(self.parts(4, 'mcu', 'BIG'), self.SIZES)
        self.assertEqual(split, [('mcu', 4, 4)], 'each part has 302 commands at most: two would pass the limit')

    def test_one_part_alone_never_loops(self):
        got, split = fp.split_blocks(self.parts(1, 'x', 'BIG'), {'BIG': (900.0, 400)})
        self.assertEqual(split, []); self.assertEqual(got[0]['block'], 'x')

    def test_an_unknown_symbol_still_splits(self):
        self.assertTrue(fp.split_blocks(self.parts(40, 'x', 'NOPE'), {})[1])

    def test_sizes_measure_pins_and_body(self):
        lib = ET.fromstring("""<library><symbols><symbol name="S"><wire x1="-5" y1="0" x2="5" y2="0"/><pin name="1" x="-10" y="0"/><pin name="2" x="10" y="0"/></symbol></symbols>
            <devicesets><deviceset name="Dev"><gates><gate symbol="S"/></gates></deviceset></devicesets></library>""")
        self.assertEqual(fp.symbol_sizes(lib), {'DEV': (20.0, 2)})

    def test_off_sheet_names_each_part_past_the_limit(self):
        groups = [{'group': 'g', 'commands': ['LAYER 91;', "ADD 'R@L' 'R1' R0 (100.000mm 139.700mm);", "ADD 'R@L' 'R2' R0 (1980.946mm 139.700mm);", "ADD 'R@L' 'R3' R0 (10.000mm -2000.000mm);", "NET 'A' (3000.000mm 0.000mm) (1.000mm 1.000mm);"]}]
        self.assertEqual(fp.off_sheet(groups), [('R2', 1980.946, 139.7, 'g'), ('R3', 10.0, -2000.0, 'g')])



class RelayoutKeeps(unittest.TestCase):
    """Issue #134: a re-lay moves only the placed names with a finding."""
    def test_only_names_with_a_finding_are_planned_again(self):
        # the lane runs silkscreen-plan.py and fusion-project.py as one script: give the helpers the plan's names
        sp = importlib.util.module_from_spec(importlib.util.spec_from_file_location('silk_plan', Path(__file__).with_name('silkscreen-plan.py')))
        sp.__spec__.loader.exec_module(sp)
        for k in ('tw_', '_ov', 'TEXT_H'):
            if hasattr(sp, k) and not hasattr(fp, k): setattr(fp, k, getattr(sp, k))
        els = [
            {'name': 'R1', 'bottom': False, 'nameAttr': 'x="10" y="10" size="0.8" layer="25"', 'nameAlign': 'bottom-left', 'homeName': (10, 10), 'pads': [(0, 0, 1, 1)], 'body': (0, 0, 2, 2)},
            {'name': 'R2', 'bottom': False, 'nameAttr': 'x="0.2" y="0.2" size="0.8" layer="25"', 'nameAlign': 'bottom-left', 'homeName': (0, 0), 'pads': [(0, 0, 1, 1)], 'body': (0, 0, 2, 2)},
            {'name': 'R3', 'bottom': False, 'nameAttr': 'x="10.5" y="10.2" size="0.8" layer="25"', 'nameAlign': 'bottom-left', 'homeName': (10, 10), 'pads': [(20, 20, 21, 21)], 'body': (20, 20, 22, 22)},
        ]
        boxes = fp.placed_name_boxes(els, set())
        probs = fp.name_problems(boxes, els, (-5, -5, 40, 40), [])
        self.assertNotIn('R1', probs, probs)
        self.assertTrue(any('over a pad' in p for p in probs['R2']), probs)
        self.assertTrue(any('overlap' in p for p in probs['R3']), 'of two overlapping names only the second moves')


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