Fix ACD export for non-program collection records

acd/l5x/elements.py

@@ -43,6 +43,37 @@ class L5xElementBuilder:
     _object_id: int = -1
 
 
+def _get_program_records(cur: Cursor, collection_id: int) -> List[Tuple[str, int, int, bytes]]:
+    cur.execute(
+        "SELECT comp_name, object_id, record_type, record FROM comps "
+        "WHERE parent_id=? AND record_type=256",
+        (collection_id,),
+    )
+    return cur.fetchall()
+
+
+def _required_program_collection(
+    cur: Cursor,
+    program_name: str,
+    program_id: int,
+    record_type: int,
+    collection_name: str,
+) -> int:
+    cur.execute(
+        "SELECT object_id FROM comps WHERE parent_id=? AND comp_name=?",
+        (program_id, collection_name),
+    )
+    rows = cur.fetchall()
+    context = (
+        f"program {program_name!r} (object_id={program_id}, record_type={record_type})"
+    )
+    if not rows:
+        raise ValueError(f"{context} is missing required {collection_name}")
+    if len(rows) > 1:
+        raise ValueError(f"{context} has multiple {collection_name} children")
+    return rows[0][0]
+
+
 # Maps Python attribute names to L5X XML section wrapper tag names.
 # Entries here also control which list attributes are serialized as child sections.
 _LIST_SECTION_NAMES = {
@@ -2122,16 +2153,17 @@ class ProgramBuilder(L5xElementBuilder):
 
     def build(self) -> Program:
         self._cur.execute(
-            "SELECT comp_name, object_id, parent_id, record FROM comps WHERE object_id="
-            + str(self._object_id)
+            "SELECT comp_name, record_type, record FROM comps WHERE object_id=?",
+            (self._object_id,),
         )
-        results = self._cur.fetchall()
+        row = self._cur.fetchone()
+        if row is None:
+            raise ValueError(f"program object_id={self._object_id} was not found")
 
-        prog_record = bytes(results[0][3])
+        name, record_type, record = row
+        prog_record = bytes(record)
         r = RxGeneric.from_bytes(prog_record)
 
-        name = results[0][0]
-
         # --- MainRoutineName and FaultRoutineName from extended records ---
         # ext[0x12D] = MainRoutine object_id, ext[0x066] = FaultRoutine object_id
         exts: Dict[int, bytes] = {e.attribute_id: bytes(e.value) for e in r.extended_records}
@@ -2160,13 +2192,13 @@ def build(self) -> Program:
         )
         disabled = "true" if disabled_flag else "false"
 
-        self._cur.execute(
-            "SELECT comp_name, object_id, parent_id, record FROM comps WHERE parent_id="
-            + str(self._object_id)
-            + " AND comp_name='RxRoutineCollection'"
+        collection_id = _required_program_collection(
+            self._cur,
+            name,
+            self._object_id,
+            record_type,
+            "RxRoutineCollection",
         )
-        collection_results = self._cur.fetchall()
-        collection_id = collection_results[0][1]
 
         self._cur.execute(
             "SELECT comp_name, object_id, parent_id, record FROM comps WHERE parent_id="
@@ -2179,18 +2211,17 @@ def build(self) -> Program:
             routines.append(RoutineBuilder(self._cur, child[1]).build())
 
         # Get the Program Scoped Tags
-        self._cur.execute(
-            "SELECT comp_name, object_id, parent_id, record_type FROM comps WHERE parent_id="
-            + str(self._object_id)
-            + " AND comp_name='RxTagCollection'"
+        tag_collection_id = _required_program_collection(
+            self._cur,
+            name,
+            self._object_id,
+            record_type,
+            "RxTagCollection",
         )
-        results = self._cur.fetchall()
-        if len(results) > 1:
-            raise Exception("Contains more than one program tag collection")
 
         self._cur.execute(
             "SELECT comp_name, object_id, parent_id, record_type FROM comps WHERE parent_id="
-            + str(results[0][1])
+            + str(tag_collection_id)
         )
         results = self._cur.fetchall()
         tags: List[Tag] = []
@@ -2430,26 +2461,19 @@ def _decode_utf16(key):
             raise Exception("Contains more than one controller program collection")
 
         _program_collection_object_id = results[0][1]
-        self._cur.execute(
-            "SELECT comp_name, object_id, parent_id, record_type FROM comps WHERE parent_id="
-            + str(_program_collection_object_id)
-        )
-        results = self._cur.fetchall()
+        program_records = _get_program_records(self._cur, _program_collection_object_id)
         programs: List[Program] = []
-        for result in results:
+        for result in program_records:
             _program_object_id = result[1]
             programs.append(
                 ProgramBuilder(self._cur, _program_object_id, data_types_map, redundancy_enabled).build()
             )
 
         # Build comment_id → program name map for task scheduled-program resolution.
         # comment_id is a u16 at BLOB offset 0x0C in each program's RxGeneric record.
-        self._cur.execute(
-            "SELECT comp_name, record FROM comps WHERE parent_id=" + str(_program_collection_object_id)
-        )
         comment_id_to_program: Dict[int, str] = {
             struct.unpack_from("<H", rec, 0x0C)[0]: pname
-            for pname, rec in self._cur.fetchall()
+            for pname, _, _, rec in program_records
         }
 
         # Get the Task Collection and build Tasks

test/test_program_collections.py

@@ -0,0 +1,48 @@
+import sqlite3
+
+import pytest
+
+from acd.l5x.elements import _get_program_records, _required_program_collection
+
+
+def _comps_cursor():
+    connection = sqlite3.connect(":memory:")
+    cursor = connection.cursor()
+    cursor.execute(
+        "CREATE TABLE comps("
+        "object_id int, parent_id int, comp_name text, seq_number int, "
+        "record_type int, record BLOB NOT NULL)"
+    )
+    return connection, cursor
+
+
+def test_program_records_exclude_non_program_metadata():
+    connection, cursor = _comps_cursor()
+    try:
+        cursor.executemany(
+            "INSERT INTO comps VALUES (?, ?, ?, ?, ?, ?)",
+            [
+                (10, 1, "Program", 0, 256, b"program"),
+                (20, 1, "Metadata", 1, 512, b"metadata"),
+            ],
+        )
+
+        assert _get_program_records(cursor, 1) == [("Program", 10, 256, b"program")]
+    finally:
+        connection.close()
+
+
+@pytest.mark.parametrize("collection_name", ["RxRoutineCollection", "RxTagCollection"])
+def test_required_program_collection_reports_program_context(collection_name):
+    connection, cursor = _comps_cursor()
+    try:
+        with pytest.raises(
+            ValueError,
+            match=(
+                r"program 'Program' \(object_id=10, record_type=256\) "
+                rf"is missing required {collection_name}"
+            ),
+        ):
+            _required_program_collection(cursor, "Program", 10, 256, collection_name)
+    finally:
+        connection.close()