"""KiCad project folder ingest: *.kicad_pro + siblings, no netlist.""" from __future__ import annotations import json import re from pathlib import Path import pytest from backend.periscopex.kicad_project import load_kicad_project, sniff_kicad_pro from backend.periscopex.netlist_bundle import materialize_netlist_upload, sniff_netlist_kind from backend.periscopex.parsers import parse_netlist_any HUBAUDIO = Path("/Users/michelebigi/Development/HubAudio/hardware/kicad/HubAudio") _LIB_R = """ (lib_symbols (symbol "Device:R" (pin passive (at 0 3.81 90) (length 2.54) (name "~" (effects (font (size 1.27 1.27)))) (number "1" (effects (font (size 1.27 1.27)))) ) (pin passive (at 0 -3.81 90) (length 2.54) (name "~" (effects (font (size 1.27 1.27)))) (number "2" (effects (font (size 1.27 1.27)))) ) ) (symbol "Device:U" (pin passive (at 0 3.81 90) (length 2.54) (name "~" (effects (font (size 1.27 1.27)))) (number "1" (effects (font (size 1.27 1.27)))) ) ) ) """ def _sch(*body: str) -> str: return ( "(kicad_sch (version 20250114) (uuid \"11111111-1111-1111-1111-111111111111\")" + _LIB_R + "".join(body) + "\n)\n" ) def _resistor(ref: str, value: str) -> str: uid = "aaaaaaaa-aaaa-aaaa-aaaa-" + ref.encode().hex()[:12].ljust(12, "0") return f""" (symbol (lib_id "Device:R") (at 0 0 0) (unit 1) (uuid "{uid}") (property "Reference" "{ref}" (at 0 0 0) (effects (font (size 1.27 1.27)))) (property "Value" "{value}" (at 0 0 0) (effects (font (size 1.27 1.27)))) (pin "1" (uuid "p1")) (pin "2" (uuid "p2")) ) """ def _ic(ref: str, value: str, pnm: str) -> str: uid = "bbbbbbbb-bbbb-bbbb-bbbb-" + ref.encode().hex()[:12].ljust(12, "0") return f""" (symbol (lib_id "Device:U") (at 0 0 0) (unit 1) (uuid "{uid}") (property "Reference" "{ref}" (at 0 0 0) (effects (font (size 1.27 1.27)))) (property "Value" "{value}" (at 0 0 0) (effects (font (size 1.27 1.27)))) (property "PNM" "{pnm}" (at 0 0 0) (effects (font (size 1.27 1.27)))) (pin "1" (uuid "u1")) ) """ def _pro(root_name: str) -> str: return json.dumps({ "meta": {"filename": root_name.replace(".kicad_sch", ".kicad_pro")}, "schematic": { "top_level_sheets": [{"filename": root_name, "name": "root"}], }, }) def _write_project(folder: Path) -> None: child = _sch( _ic("U9", "TPD2E007DCKR", "TPD2E007DCKR"), """ (global_label "GND" (at 0 3.81 0) (uuid "cccccccccccccccccccccccccccccccccccc")) """, ) root = _sch( _resistor("R1", "10k"), """ (global_label "GND" (at 0 3.81 0) (uuid "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb")) (sheet (at 50 0) (size 20 20) (property "Sheetname" "Codec" (at 50 0 0) (effects (font (size 1.27 1.27)))) (property "Sheetfile" "Codec.kicad_sch" (at 50 0 0) (effects (font (size 1.27 1.27)))) ) """, ) folder.mkdir(parents=True, exist_ok=True) (folder / "HubAudio.kicad_pro").write_text(_pro("HubAudio.kicad_sch")) (folder / "HubAudio.kicad_sch").write_text(root) (folder / "Codec.kicad_sch").write_text(child) (folder / "HubAudio.kicad_pcb").write_text( '(kicad_pcb (version 20240108) (generator pcbnew)\n (net 0 "")\n)\n' ) (folder / "netlist.asc").write_text( "*PADS-PCB*\n*PART*\nU13 SOT23\nR1 0603\n*NET*\n" "*SIGNAL* GND\nU13.1 R1.1\n*END*\n" ) def test_sniff_kicad_pro_json(): body = _pro("HubAudio.kicad_sch").encode() assert sniff_kicad_pro(body, "HubAudio.kicad_pro") assert sniff_netlist_kind(body, "HubAudio.kicad_pro") == "kicad_pro" def test_lone_kicad_pro_bytes_are_rejected(tmp_path: Path): with pytest.raises(ValueError, match="da solo non basta"): materialize_netlist_upload( [("HubAudio.kicad_pro", _pro("HubAudio.kicad_sch").encode())], tmp_path / "work", ) def test_folder_next_to_pro_loads_sheets_and_pcb(tmp_path: Path): src = tmp_path / "HubAudio" _write_project(src) files = [ (str(p.relative_to(tmp_path)), p.read_bytes()) for p in src.iterdir() if p.is_file() ] parsed = materialize_netlist_upload(files, tmp_path / "work") parts, _nets, fmt = parse_netlist_any(parsed.root) assert fmt == "kicad_sch" assert parsed.pcb is not None and parsed.pcb.name == "HubAudio.kicad_pcb" assert "R1" in parts and "U9" in parts assert "U13" not in parts assert {p.name for p in parsed.extra_sch} == {"Codec.kicad_sch"} def test_stale_pads_asc_ignored_when_pro_present(tmp_path: Path): src = tmp_path / "HubAudio" _write_project(src) loaded = load_kicad_project(src) assert loaded.root_sch.name == "HubAudio.kicad_sch" parts, _nets, fmt = parse_netlist_any(loaded.root_sch) assert fmt == "kicad_sch" assert "U13" not in parts def test_disk_path_to_pro_copies_siblings(tmp_path: Path): src = tmp_path / "HubAudio" _write_project(src) parsed = materialize_netlist_upload( [(str(src / "HubAudio.kicad_pro"), (src / "HubAudio.kicad_pro").read_bytes())], tmp_path / "work", ) parts, _nets, _fmt = parse_netlist_any(parsed.root) assert "U9" in parts and "U13" not in parts assert parsed.pcb is not None def test_graph_without_bom_csv_uses_schematic_pnm(tmp_path: Path): from backend.periscopex.graph import build_graph src = tmp_path / "HubAudio" _write_project(src) missing = tmp_path / "no-bom.csv" g = build_graph( src / "HubAudio.kicad_sch", missing, tmp_path / "ex", tmp_path / "pat", tmp_path / "mod", ) assert g.components["U9"].mpn == "TPD2E007DCKR" assert "U13" not in g.components def test_empty_child_pnm_not_invented(tmp_path: Path): from backend.periscopex.graph import build_graph folder = tmp_path / "proj" folder.mkdir() child = _sch( _ic("U15", "", ""), """ (global_label "GND" (at 0 3.81 0) (uuid "cccccccccccccccccccccccccccccccccccc")) """, ) root = _sch( _resistor("R1", "10k"), """ (global_label "GND" (at 0 3.81 0) (uuid "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb")) (sheet (at 50 0) (size 20 20) (property "Sheetname" "Codec" (at 50 0 0) (effects (font (size 1.27 1.27)))) (property "Sheetfile" "Codec.kicad_sch" (at 50 0 0) (effects (font (size 1.27 1.27)))) ) """, ) (folder / "p.kicad_pro").write_text(_pro("p.kicad_sch")) (folder / "p.kicad_sch").write_text(root) (folder / "Codec.kicad_sch").write_text(child) g = build_graph( folder / "p.kicad_sch", tmp_path / "missing.csv", tmp_path / "ex", tmp_path / "pat", tmp_path / "mod", ) assert not (g.components["U15"].mpn or "").strip() @pytest.mark.skipif(not (HUBAUDIO / "HubAudio.kicad_pro").is_file(), reason="HubAudio tree not on disk") def test_hubaudio_folder_siblings_only(): loaded = load_kicad_project(HUBAUDIO) assert loaded.pro.name == "HubAudio.kicad_pro" assert loaded.root_sch.name == "HubAudio.kicad_sch" names = {p.name for p in loaded.sheets} assert "Codec.kicad_sch" in names assert "POWER.kicad_sch" in names assert loaded.pcb is not None and loaded.pcb.name == "HubAudio.kicad_pcb" assert loaded.folder == HUBAUDIO.resolve() parts, _nets, fmt = parse_netlist_any(loaded.root_sch) assert fmt == "kicad_sch" assert "U13" not in parts assert "U9" in parts def test_api_folder_upload_no_csv(tmp_path: Path): from fastapi.testclient import TestClient from backend.main import app from backend.services.storage import LocalStorageBackend app.state.storage = LocalStorageBackend(tmp_path) client = TestClient(app) pid = client.post("/api/projects", json={"name": "ha"}).json()["id"] src = tmp_path / "src" _write_project(src) files = [ ("files", (p.name, p.read_bytes(), "application/octet-stream")) for p in src.iterdir() if p.suffix in {".kicad_pro", ".kicad_sch", ".kicad_pcb"} ] paths = [p.name for p in src.iterdir() if p.suffix in {".kicad_pro", ".kicad_sch", ".kicad_pcb"}] resp = client.post( f"/api/projects/{pid}/upload/netlist", files=files, data={"paths": json.dumps(paths)}, ) assert resp.status_code == 200, resp.text body = resp.json() assert body["format"] == "kicad_sch" assert body["pcb_saved"] is True assert body["bom_saved"] is True assert "U13" not in str(body) assert body["parts"] >= 2 def _looks_like_kicad_pro(name: str, relative: str = "") -> bool: """Same rule as frontend looksLikeKicadPro (name and/or webkitRelativePath).""" return any(".kicad_pro" in s.lower() for s in (name, relative, Path(relative or name).name)) def test_safari_stem_name_is_still_hubaudio_kicad_pro(): """Live 2.60.6 checked only File.name.endsWith('.kicad_pro') and missed Safari.""" assert not "HubAudio".lower().endswith(".kicad_pro") assert _looks_like_kicad_pro("HubAudio", "HubAudio/HubAudio.kicad_pro") assert _looks_like_kicad_pro("HubAudio.kicad_pro", "") assert sniff_netlist_kind( _pro("HubAudio.kicad_sch").encode(), "HubAudio", ) == "kicad_pro" def test_safari_stem_upload_writes_kicad_pro_suffix(tmp_path: Path): src = tmp_path / "HubAudio" _write_project(src) pro = (src / "HubAudio.kicad_pro").read_bytes() files = [ ("HubAudio", pro), ("HubAudio.kicad_sch", (src / "HubAudio.kicad_sch").read_bytes()), ("Codec.kicad_sch", (src / "Codec.kicad_sch").read_bytes()), ("HubAudio.kicad_pcb", (src / "HubAudio.kicad_pcb").read_bytes()), ] parsed = materialize_netlist_upload(files, tmp_path / "work") assert parsed.root.name == "HubAudio.kicad_sch" assert any(p.name == "HubAudio.kicad_pro" for p in (tmp_path / "work").rglob("*")) parts, _nets, fmt = parse_netlist_any(parsed.root) assert fmt == "kicad_sch" assert "U9" in parts and "U13" not in parts @pytest.mark.skipif(not (HUBAUDIO / "HubAudio.kicad_pro").is_file(), reason="HubAudio tree not on disk") def test_real_hubaudio_kicad_pro_filename_and_folder_ingest(tmp_path: Path): pro_path = HUBAUDIO / "HubAudio.kicad_pro" assert pro_path.name == "HubAudio.kicad_pro" data = pro_path.read_bytes() assert sniff_kicad_pro(data, "HubAudio.kicad_pro") assert sniff_netlist_kind(data, "HubAudio") == "kicad_pro" files: list[tuple[str, bytes]] = [] for path in HUBAUDIO.iterdir(): if path.suffix.lower() in {".kicad_pro", ".kicad_sch", ".kicad_pcb"}: files.append((f"HubAudio/{path.name}", path.read_bytes())) assert any(name.endswith("HubAudio.kicad_pro") for name, _ in files) parsed = materialize_netlist_upload(files, tmp_path / "work") assert parsed.root.name == "HubAudio.kicad_sch" assert parsed.pcb is not None parts, _nets, fmt = parse_netlist_any(parsed.root) assert fmt == "kicad_sch" assert "U13" not in parts # Mirrors periscope/src/frontend/src/lib/kicad-project-files.ts (folder walk + Next). def _skip_walk_dir_name(name: str) -> bool: n = name.rstrip("/").lower() if not n or n.startswith("."): return True return bool(re.search(r"(-backups|\.pretty|3dmodels|__macosx|node_modules)$", n)) def _is_kicad_project_file_name(name: str) -> bool: base = name.replace("\\", "/").rsplit("/", 1)[-1].lower() return base.endswith((".kicad_pro", ".kicad_sch", ".kicad_pcb")) def _skip_rel(rel: str) -> bool: parts = [p for p in rel.replace("\\", "/").lower().split("/") if p] return any(_skip_walk_dir_name(p) for p in parts) def _details_step_ready(project_name: str, files: list[dict[str, str]]) -> bool: if not project_name.strip(): return False return any(_looks_like_kicad_pro(f.get("name", ""), f.get("webkitRelativePath", "")) for f in files) def test_folder_walk_skips_history_and_keeps_project_dir_only(): """Live 2.60.8 recursed handle.entries() into HubAudio/.history (~2000 files) so Next never enabled.""" names = [ "HubAudio.kicad_pro", "HubAudio.kicad_sch", "Codec.kicad_sch", "HubAudio.kicad_pcb", "HubAudio.kicad_prl", "~HubAudio.kicad_pcb.lck", ".history", ".git", "fp-lib-table", ] kept = [n for n in names if _is_kicad_project_file_name(n)] skipped_dirs = [n for n in names if _skip_walk_dir_name(n)] assert kept == [ "HubAudio.kicad_pro", "HubAudio.kicad_sch", "Codec.kicad_sch", "HubAudio.kicad_pcb", ] assert ".history" in skipped_dirs and ".git" in skipped_dirs assert _skip_rel("HubAudio/.history/objects/aa") assert not _skip_rel("HubAudio/HubAudio.kicad_pro") def test_details_next_enables_when_kicad_pro_is_in(): empty: list[dict[str, str]] = [] assert not _details_step_ready("HubAudio", empty) assert not _details_step_ready( "", [{"name": "HubAudio.kicad_pro", "webkitRelativePath": "HubAudio/HubAudio.kicad_pro"}], ) assert _details_step_ready( "HubAudio", [{"name": "HubAudio.kicad_pro", "webkitRelativePath": "HubAudio/HubAudio.kicad_pro"}], ) assert _details_step_ready( "HubAudio", [{"name": "HubAudio", "webkitRelativePath": "HubAudio/HubAudio.kicad_pro"}], ) @pytest.mark.skipif(not (HUBAUDIO / "HubAudio.kicad_pro").is_file(), reason="HubAudio tree not on disk") def test_real_hubaudio_recursive_tree_is_huge_project_dir_is_small(): tree = list(HUBAUDIO.rglob("*")) project_dir = [ p for p in HUBAUDIO.iterdir() if p.is_file() and _is_kicad_project_file_name(p.name) ] assert len(tree) > 100 assert (HUBAUDIO / ".history").is_dir() assert any(p.name == "HubAudio.kicad_pro" for p in project_dir) assert any(p.name.endswith(".kicad_sch") for p in project_dir) assert any(p.name.endswith(".kicad_pcb") for p in project_dir) assert all(".history" not in str(p) for p in project_dir) assert len(project_dir) < 40 sch_names = {p.name for p in project_dir if p.suffix.lower() == ".kicad_sch"} assert "HubAudio.kicad_sch" in sch_names assert "USB.kicad_sch" in sch_names assert "Codec.kicad_sch" in sch_names assert "POWER.kicad_sch" in sch_names assert len(sch_names) > 1 def _sheet_block(name: str, file: str) -> str: return f""" (sheet (at 50 0) (size 20 20) (property "Sheetname" "{name}" (at 50 0 0) (effects (font (size 1.27 1.27)))) (property "Sheetfile" "{file}" (at 50 0 0) (effects (font (size 1.27 1.27)))) ) """ def _write_hier_modules(folder: Path) -> None: """Root + USB/Codec/POWER siblings + sheets/nested.kicad_sch; junk in .history.""" folder.mkdir(parents=True, exist_ok=True) nested_dir = folder / "sheets" nested_dir.mkdir() history = folder / ".history" history.mkdir() (history / "HubAudio.kicad_sch").write_text(_sch(_ic("U99", "FAKE", "FAKE"))) (history / "evil.kicad_pro").write_text(_pro("HubAudio.kicad_sch")) usb = _sch( _ic("U2", "CH340E", "CH340E"), """ (global_label "GND" (at 0 3.81 0) (uuid "dddddddd-dddd-dddd-dddd-dddddddddddd")) """, ) codec = _sch( _ic("U9", "TPD2E007DCKR", "TPD2E007DCKR"), """ (global_label "GND" (at 0 3.81 0) (uuid "cccccccccccccccccccccccccccccccccccc")) """, ) power = _sch( _ic("U1", "SPX3819M5-L-3-3", "SPX3819M5-L-3-3"), """ (global_label "GND" (at 0 3.81 0) (uuid "eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee")) """, ) nested = _sch( _ic("U4", "PCA9534ARGTR", "PCA9534ARGTR"), """ (global_label "GND" (at 0 3.81 0) (uuid "ffffffffffffffffffffffffffffffffffffffff")) """, ) root = _sch( _resistor("R1", "10k"), """ (global_label "GND" (at 0 3.81 0) (uuid "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb")) """ + _sheet_block("USB", "USB.kicad_sch") + _sheet_block("Codec", "Codec.kicad_sch") + _sheet_block("POWER", "POWER.kicad_sch") + _sheet_block("Nested", "sheets/nested.kicad_sch"), ) (folder / "HubAudio.kicad_pro").write_text(_pro("HubAudio.kicad_sch")) (folder / "HubAudio.kicad_sch").write_text(root) (folder / "USB.kicad_sch").write_text(usb) (folder / "Codec.kicad_sch").write_text(codec) (folder / "POWER.kicad_sch").write_text(power) (nested_dir / "nested.kicad_sch").write_text(nested) (folder / "HubAudio.kicad_pcb").write_text( '(kicad_pcb (version 20240108) (generator pcbnew)\n (net 0 "")\n)\n' ) def _sheetfiles_from_text(text: str) -> list[str]: found: list[str] = [] for pat in ( r'\(\s*property\s+"Sheetfile"\s+"([^"]+)"', r'\(\s*sheetfile\s+"([^"]+)"', r'\(\s*file\s+"([^"]+\.kicad_sch)"', r'file\s*=\s*"([^"]+\.kicad_sch)"', ): found.extend(re.findall(pat, text, flags=re.I)) return list(dict.fromkeys(found)) def _join_sheet_rel(sch_rel: str, sheetfile: str) -> str | None: file = sheetfile.replace("\\", "/").strip() if not file or file.startswith("/") or ".." in file.split("/"): return None parent = sch_rel.replace("\\", "/").rsplit("/", 1)[0] if "/" in sch_rel.replace("\\", "/") else "" joined = f"{parent}/{file}" if parent else file if _skip_rel(joined): return None return joined def _files_beside(incoming: list[dict[str, str]]) -> list[dict[str, str]]: """Same keep rule as frontend filesBesideKicadPro (all sch under project dir).""" hits = [ f for f in incoming if _looks_like_kicad_pro(f["name"], f.get("webkitRelativePath", "")) and not _skip_rel(f.get("webkitRelativePath") or f["name"]) ] if not hits: return [] pro_rel = (hits[0].get("webkitRelativePath") or hits[0]["name"]).replace("\\", "/") pro_dir = pro_rel.rsplit("/", 1)[0] if "/" in pro_rel else "" kept: list[dict[str, str]] = [] for f in incoming: rel = (f.get("webkitRelativePath") or f["name"]).replace("\\", "/") if _skip_rel(rel): continue parent = rel.rsplit("/", 1)[0] if "/" in rel else "" under = parent == pro_dir or ( bool(pro_dir) and (parent == pro_dir or parent.startswith(pro_dir + "/")) ) or (not parent and not pro_dir) if not under: continue base = rel.rsplit("/", 1)[-1].lower() if not ( ".kicad_pro" in rel.lower() or base.endswith((".kicad_sch", ".kicad_pcb")) ): continue kept.append(f) return kept def test_sheetfiles_from_root_text_usb_codec_power(): text = ( _sheet_block("USB", "USB.kicad_sch") + _sheet_block("Codec", "Codec.kicad_sch") + _sheet_block("POWER", "POWER.kicad_sch") + '(file "sheets/nested.kicad_sch")' + 'file="also.kicad_sch"' ) names = _sheetfiles_from_text(text) assert "USB.kicad_sch" in names assert "Codec.kicad_sch" in names assert "POWER.kicad_sch" in names assert "sheets/nested.kicad_sch" in names assert "also.kicad_sch" in names assert _join_sheet_rel("HubAudio/HubAudio.kicad_sch", "USB.kicad_sch") == "HubAudio/USB.kicad_sch" assert _join_sheet_rel("HubAudio/HubAudio.kicad_sch", "sheets/nested.kicad_sch") == "HubAudio/sheets/nested.kicad_sch" assert _join_sheet_rel("HubAudio/HubAudio.kicad_sch", ".history/x.kicad_sch") is None assert _join_sheet_rel("HubAudio/HubAudio.kicad_sch", "../escape.kicad_sch") is None def test_keep_every_sibling_sch_not_only_root(): incoming = [ {"name": "HubAudio.kicad_pro", "webkitRelativePath": "HubAudio/HubAudio.kicad_pro"}, {"name": "HubAudio.kicad_sch", "webkitRelativePath": "HubAudio/HubAudio.kicad_sch"}, {"name": "USB.kicad_sch", "webkitRelativePath": "HubAudio/USB.kicad_sch"}, {"name": "Codec.kicad_sch", "webkitRelativePath": "HubAudio/Codec.kicad_sch"}, {"name": "POWER.kicad_sch", "webkitRelativePath": "HubAudio/POWER.kicad_sch"}, {"name": "nested.kicad_sch", "webkitRelativePath": "HubAudio/sheets/nested.kicad_sch"}, {"name": "HubAudio.kicad_pcb", "webkitRelativePath": "HubAudio/HubAudio.kicad_pcb"}, {"name": "HubAudio.kicad_sch", "webkitRelativePath": "HubAudio/.history/HubAudio.kicad_sch"}, ] kept = _files_beside(incoming) names = { (f.get("webkitRelativePath") or f["name"]).rsplit("/", 1)[-1] for f in kept } rels = { f.get("webkitRelativePath") or f["name"] for f in kept } assert "HubAudio.kicad_sch" in names assert "USB.kicad_sch" in names assert "Codec.kicad_sch" in names assert "POWER.kicad_sch" in names assert "nested.kicad_sch" in names assert names != {"HubAudio.kicad_sch"} assert "HubAudio/.history/HubAudio.kicad_sch" not in rels def test_hier_modules_and_nested_sheet_history_ignored(tmp_path: Path): src = tmp_path / "HubAudio" _write_hier_modules(src) loaded = load_kicad_project(src) names = {p.name for p in loaded.sheets} assert names == { "HubAudio.kicad_sch", "USB.kicad_sch", "Codec.kicad_sch", "POWER.kicad_sch", "nested.kicad_sch", } assert all(".history" not in p.parts for p in loaded.sheets) assert loaded.pro.name == "HubAudio.kicad_pro" parts, _nets, fmt = parse_netlist_any(loaded.root_sch) assert fmt == "kicad_sch" assert "U2" in parts and "U9" in parts and "U1" in parts and "U4" in parts assert "U99" not in parts files = [ (str(p.relative_to(tmp_path)), p.read_bytes()) for p in src.rglob("*") if p.is_file() and p.suffix.lower() in {".kicad_pro", ".kicad_sch", ".kicad_pcb"} and ".history" not in p.parts ] parsed = materialize_netlist_upload(files, tmp_path / "work") extra = {p.name for p in parsed.extra_sch} assert extra == {"USB.kicad_sch", "Codec.kicad_sch", "POWER.kicad_sch", "nested.kicad_sch"} copied = {p.name for p in (tmp_path / "work").rglob("*.kicad_sch")} assert "nested.kicad_sch" in copied assert not list((tmp_path / "work").rglob(".history/**/*")) def test_history_kicad_pro_is_not_the_project(tmp_path: Path): src = tmp_path / "HubAudio" _write_hier_modules(src) loaded = load_kicad_project(src) assert ".history" not in loaded.pro.parts assert loaded.pro.parent == src.resolve()