AI Flow

"""Tests for tools/project-page.py against a stub adom-wiki and a stub kicad-cli (no network, no KiCad):
python3 tools/test_project_page.py"""
import importlib.util, json, os, stat, sys, tempfile, unittest, zipfile
from pathlib import Path
from unittest import mock

_spec = importlib.util.spec_from_file_location('project_page', Path(__file__).with_name('project-page.py'))
pp = importlib.util.module_from_spec(_spec); _spec.loader.exec_module(pp)

HEADER = '(kicad_symbol_lib (version 20250114) (generator t)\n'
GOOD = '(symbol "GOOD1" (property "Reference" "R" (at 0 0 0)) (property "Footprint" "x:GOOD1" (at 0 0 0)))'
CHILD = '(symbol "CHILD1" (extends "GOOD1"))'
# noah's two pages: written without quotes
BAD = '(symbol bsc016n06ns (property Reference Q (at 0 0 0)))'
# a symbol KiCad cannot load whatever the quoting
BROKEN = '(symbol "bsc016n06ns" (property "Reference" "Q" (at 0 0 0)) (property "Footprint" "x:GOOD1" (at 0 0 0)) (unknown_token))'
# issue #196: a page written with unquoted atoms, which KiCad reads; project-page puts the quotes in
UNQUOTED = '(symbol cl05b104kb54pnc (property Reference C (at 0 0 0)) (property Footprint cl05b104kb54pnc:GOOD1 (at 0 0 0)))'

# refuses a library holding an unquoted symbol name, the way KiCad 10 refuses it
KICAD_CLI = '''#!/usr/bin/env python3
import re, sys
args = sys.argv[1:]
out = args[args.index('-o') + 1]
text = open(args[-1]).read()
if re.search(r'\\(symbol [^"(]', text) or '(unknown_token)' in text:
    print('Unable to load library'); sys.exit(1)
if '(extends' in text and '(symbol "GOOD1"' not in text:
    print('Unable to load library'); sys.exit(1)
open(out, 'w').write(text)
'''

# stateful adom-wiki: pages and library items live in a JSON file, every call is logged
WIKI = '''#!/usr/bin/env python3
import json, os, sys, zipfile
state_path = os.environ['STUB_STATE']
st = json.load(open(state_path))
a = sys.argv[1:]
st['calls'].append(a)
def done(obj=None, code=0):
    json.dump(st, open(state_path, 'w'))
    if obj is not None: print(json.dumps(obj))
    sys.exit(code)
if a[:2] == ['repo', 'show']:
    done(code=0 if a[2] in st['pages'] else 1)
if a[:2] == ['page', 'create']:
    st['pages'].append('adom/' + a[2]); done({'status': 'ok'})
if a[0] == 'whoami':
    done({'data': {'user': {'username': 'adom'}}})
if a[:2] == ['repo', 'push']:
    done({'commit': 'abcdef1'}, 1 if st.get('pushFails') else 0)
if a[0] == 'library':
    v = a[1]
    if v == 'list': done({'data': {'libraries': st['libs']}})
    if v == 'create':
        st['libs'].append({'ref': 'lib_x', 'name': a[2], 'active': True}); done({'data': {'ref': 'lib_x'}})
    if v == 'activate': done({})
    if v == 'show': done({'status': 'ok', 'data': {'data': {'items': st['items']}}})
    if v == 'remove':
        st['items'] = [i for i in st['items'] if str(i['id']) != a[3]]; done({'removed': True})
    if v == 'add':
        for pg in [x for x in a[3:] if '/' in x]:
            if not any(f"{i['owner']}/{i['slug']}" == pg for i in st['items']):
                o, s = pg.split('/'); st['items'].append({'id': len(st['calls']) + 1000, 'ref': 'pg_' + s, 'owner': o, 'slug': s, 'mpn': s.upper()})
        done({'status': 'ok'})
    if v == 'export':
        out = a[a.index('-o') + 1]
        with zipfile.ZipFile(out, 'w') as z:
            z.writestr('lib/lib.kicad_sym', st['symbols'])
            z.writestr('lib/lib.pretty/GOOD1.kicad_mod', '(footprint "GOOD1" (model "3dmodels/GOOD1.step"))')
            z.writestr('lib/3dmodels/GOOD1.step', 'x')
        done({'status': 'ok', 'data': {'report': {}}})
done(code=2)
'''


def script(path, text):
    Path(path).write_text(text); os.chmod(path, os.stat(path).st_mode | stat.S_IEXEC)


class Base(unittest.TestCase):
    def setUp(self):
        self.d = tempfile.mkdtemp(prefix='aiflow-page-test-')
        self.bin = os.path.join(self.d, 'bin'); os.makedirs(self.bin)
        self.state = os.path.join(self.d, 'state.json')
        script(os.path.join(self.bin, 'adom-wiki'), WIKI)
        script(os.path.join(self.bin, 'kicad-cli'), KICAD_CLI)
        env = {'PATH': self.bin + os.pathsep + os.environ['PATH'], 'STUB_STATE': self.state,
               'ADOM_AIFLOW_KICAD_CLI': os.path.join(self.bin, 'kicad-cli')}
        patcher = mock.patch.dict(os.environ, env); patcher.start(); self.addCleanup(patcher.stop)

    def write_state(self, **kw):
        st = dict(pages=[], libs=[], items=[], calls=[], symbols=HEADER + GOOD + '\n)\n'); st.update(kw)
        json.dump(st, open(self.state, 'w'))

    def read_state(self):
        return json.load(open(self.state))


class Blame(Base):
    def test_names_the_parts_that_break_the_library(self):
        path = os.path.join(self.d, 'lib.kicad_sym')
        Path(path).write_text(HEADER + GOOD + '\n' + BAD + '\n)\n')
        self.assertEqual(pp.kicad_check(path)[0], 'fail')
        bad = pp.broken_symbols(path, [{'mpn': 'BSC016N06NS', 'wiki': 'adom/bsc016n06ns'}, {'mpn': 'GOOD1', 'wiki': 'adom/good1'}])
        self.assertEqual([b['symbol'] for b in bad], ['bsc016n06ns'])
        self.assertEqual(bad[0]['label'], 'BSC016N06NS (adom/bsc016n06ns)')

    def test_a_child_symbol_is_tested_with_its_parent(self):
        path = os.path.join(self.d, 'lib.kicad_sym')
        Path(path).write_text(HEADER + GOOD + '\n' + CHILD + '\n)\n')
        self.assertEqual(pp.broken_symbols(path, []), [])

    def test_no_kicad_cli_blames_nothing(self):
        path = os.path.join(self.d, 'lib.kicad_sym')
        Path(path).write_text(HEADER + BAD + '\n)\n')
        with mock.patch.dict(os.environ, {'ADOM_AIFLOW_KICAD_CLI': '', 'PATH': self.bin.replace('bin', 'nothing')}):
            self.assertEqual(pp.broken_symbols(path, []), [])


class Stale(Base):
    def item(self, i, slug):
        return {'id': i, 'ref': 'pg_' + slug, 'owner': 'adom', 'slug': slug, 'mpn': slug.upper()}

    def test_parts_off_the_bom_leave_the_library_and_nothing_else_does(self):
        self.write_state(items=[self.item(1, 'keep'), self.item(2, 'gone'), {'id': 3, 'ref': 'pg_dead', 'resolved': False}])
        removed, _ = pp.drop_stale_items('lib_x', ['adom/keep'])
        self.assertEqual(removed, ['GONE (adom/gone)', 'pg_dead (page gone)'])
        self.assertEqual([i['id'] for i in self.read_state()['items']], [1])

    def test_a_page_named_by_its_id_or_in_other_case_stays(self):
        self.write_state(items=[self.item(1, 'keep'), self.item(2, 'other')])
        self.assertEqual(pp.drop_stale_items('lib_x', ['ADOM/KEEP', 'pg_other'])[0], [])

    def test_an_unreadable_library_removes_nothing(self):
        self.write_state(items=[self.item(1, 'gone')])
        with mock.patch.object(pp, 'run_retrying', return_value=(False, 'boom', [])):
            with self.assertRaises(SystemExit) as e:
                pp.drop_stale_items('lib_x', [])
        self.assertIn('could not read', str(e.exception))


class Create(Base):
    PARTS = [{'mpn': 'GOOD1', 'wiki': 'adom/good1', 'refs': ['R1'], 'noModelExpected': True}]

    def run_main(self, **over):
        cfg = dict(run=self.d, eda='kicad', page='adom/board1', pageDir=None, title='Board', brief='A board', summary='',
                   parts=self.PARTS, create=True, push=True, private=False, replaceReadme=False, existing=False, version='1.0.0', libRef=None)
        cfg.update(over)
        path = os.path.join(self.d, 'config.json'); json.dump(cfg, open(path, 'w'))
        with mock.patch.object(sys, 'argv', ['project-page', path]), mock.patch('builtins.print') as out:
            try:
                pp.main()
            except SystemExit as e:
                return json.loads(str(e))
        return json.loads(out.call_args[0][0])

    def pushed_readme(self):
        return any(c[:2] == ['repo', 'push'] and 'README.md' in c for c in self.read_state()['calls'])

    def test_a_failed_library_check_leaves_no_page_and_names_the_part(self):
        self.write_state(symbols=HEADER + GOOD + '\n' + BROKEN + '\n)\n')
        r = self.run_main(parts=self.PARTS + [{'mpn': 'BSC016N06NS', 'wiki': 'adom/bsc016n06ns', 'refs': ['Q1']}])
        self.assertFalse(r['ok'])
        self.assertIn('BSC016N06NS (adom/bsc016n06ns): KiCad cannot load this symbol', r['error'])
        self.assertEqual(r['libraryCheck']['brokenSymbols'][0]['page'], 'adom/bsc016n06ns')
        self.assertEqual(self.read_state()['pages'], [], 'nothing was published, and no page was made')
        # fixed, the same --create goes through
        st = self.read_state(); st['symbols'] = HEADER + GOOD + '\n)\n'; json.dump(st, open(self.state, 'w'))
        r = self.run_main()
        self.assertTrue(r['ok'], r); self.assertTrue(r['created']); self.assertTrue(self.pushed_readme())

    def test_a_page_with_unquoted_atoms_is_packaged(self):
        # issue #196: before, its Footprint read as missing and the PCM rewrite skipped it
        self.write_state(symbols=HEADER + GOOD + '\n' + UNQUOTED + '\n)\n')
        r = self.run_main(parts=self.PARTS + [{'mpn': 'CL05B104KB54PNC', 'wiki': 'adom/cl05b104kb54pnc', 'refs': ['C1'], 'noModelExpected': True}])
        self.assertTrue(r['ok'], r)
        sym = next(Path(self.d).rglob('*.kicad_sym')).read_text()
        self.assertIn('(symbol "cl05b104kb54pnc"', sym)
        self.assertRegex(sym, r'\(property "Footprint" "PCM_[^":]+:GOOD1"')

    def test_a_failed_push_can_be_rerun_with_create_and_finishes_the_page(self):
        self.write_state(pushFails=True)
        self.assertFalse(self.run_main()['ok'])
        self.assertEqual(self.read_state()['pages'], ['adom/board1'])
        st = self.read_state(); st['pushFails'] = False; st['calls'] = []; json.dump(st, open(self.state, 'w'))
        r = self.run_main()
        self.assertTrue(r['ok'], r); self.assertTrue(r['created']); self.assertTrue(self.pushed_readme())
        self.assertFalse(any(c[:2] == ['page', 'create'] for c in self.read_state()['calls']))

    def test_a_finished_page_or_someone_elses_is_still_refused(self):
        self.write_state()
        self.assertTrue(self.run_main()['ok'])
        r = self.run_main()
        self.assertFalse(r['ok']); self.assertIn('already exists', r['error'])
        self.write_state(pages=['adom/board2'])
        r = self.run_main(page='adom/board2')
        self.assertFalse(r['ok']); self.assertIn('already exists', r['error'])

    def test_a_rerun_drops_what_left_the_bom_before_exporting(self):
        self.write_state(items=[{'id': 9, 'ref': 'pg_old', 'owner': 'adom', 'slug': 'old', 'mpn': 'OLD'}])
        r = self.run_main()
        self.assertTrue(r['ok'], r)
        self.assertEqual(r['adds']['removedStale'], ['OLD (adom/old)'])
        self.assertTrue(any('no longer on the BOM' in n for n in r['adds']['notes']))
        calls = self.read_state()['calls']
        order = [c[1] for c in calls if c[0] == 'library' and c[1] in ('remove', 'add', 'export')]
        self.assertEqual(order, ['remove', 'add', 'export'])


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