src/python/jw/pkg/App.py

@@ -10,12 +10,13 @@ import re
 import sys
 
 from enum import Enum, auto
-from functools import lru_cache
+from functools import cache
 from typing import TYPE_CHECKING
 
 from .lib.App import App as Base
 from .lib.Distro import Distro
 from .lib.log import DEBUG, ERR, log
+from .lib.ProjectConf import ProjectConf
 
 if TYPE_CHECKING:
     import argparse
@@ -180,6 +181,17 @@ class App(Base):
                 return ret
         return None
 
+    @cache
+    def __read_project_conf(self, project_dir: str) -> ProjectConf:
+        return ProjectConf.read(project_dir + '/make/project.conf')
+
+    @cache
+    def __get_project_conf(self, project: str) -> ProjectConf:
+        pd = self.__proj_dir(project, False)
+        if pd is None:
+            raise Exception(f'Failed to find directory of project {project}')
+        return self.__read_project_conf(pd)
+
     def __get_project_refs_cached(
         self, buf, visited, spec, section, key, add_self, scope, names_only
     ):
@@ -405,9 +417,11 @@ class App(Base):
         self.___pretty_topdir = self.__format_topdir(self.___topdir, args.topdir_format)
         self.__topdir_fmt = args.topdir_format
         if self.___topdir is not None:
-            self.__top_name = self.read_value(
-                self.___topdir + '/make/project.conf', 'build', 'name'
-            )
+            try:
+                conf = self.__read_project_conf(self.__topdir)
+                self.__top_name = conf.get_str_or_none('build', 'name')
+            except FileNotFoundError:
+                pass
             if not self.__top_name:
                 self.__top_name = re.sub(
                     '-[0-9.-]*$',
@@ -512,83 +526,13 @@ class App(Base):
     def strip_module_from_spec(self, mod):
         return re.sub(r'-dev$|-devel$|-run$', '', re.split('([=><]+)', mod)[0].strip())
 
-    @lru_cache(maxsize = None)
-    def get_section(self, path: str, section: str) -> str:
-        ret = ''
-        pat = '[' + section + ']'
-        in_section = False
-        file = open(path)
-        for line in file:
-            if line.rstrip() == pat:
-                in_section = True
-                continue
-            if in_section:
-                if len(line) and line[0] == '[':
-                    break
-                ret += line
-        file.close()
-        return ret.rstrip()
-
-    @lru_cache(maxsize = None)
-    def read_value(self, path: str, section: str, key: str) -> str | None:
-
-        def scan_section(f, key: str) -> str | None:
-            if key is None:
-                ret = ''
-                for line in f:
-                    if len(line) and line[0] == '[':
-                        break
-                    ret += line
-                return ret if len(ret) else None
-            lines: list[str] = []
-            cont_line = ''
-            for line in f:
-                if len(line) and line[0] == '[':
-                    break
-                cont_line += line.rstrip()
-                if len(cont_line) and cont_line[-1] == '\\':
-                    cont_line = cont_line[0:-1]
-                    continue
-                lines.append(cont_line)
-                cont_line = ''
-            rx = re.compile(r'^\s*' + key + r'\s*=\s*(.*)\s*$')
-            for line in lines:
-                # log(DEBUG, '  looking for "%s" in line="%s"' % (key, line))
-                m = re.search(rx, line)
-                if m is not None:
-                    return m.group(1)
-            return None
-
-        def scan_section_debug(f, key: str) -> str | None:
-            ret = scan_section(f, key)
-            # log(DEBUG, '  returning', rr)
-            return ret
-
-        try:
-            # log(DEBUG, 'looking for {}::[{}].{}'.format(path, section, key))
-            # TODO: Parse
-            with open(path, 'r') as f:
-                if not len(section):
-                    return scan_section(f, key)
-                pat = '[' + section + ']'
-                for line in f:
-                    if line.rstrip() == pat:
-                        return scan_section(f, key)
-                return None
-        except Exception:
-            log(DEBUG, f'Not found: {path}')
-            # TODO: handle this special case cleaner somewhere up the stack
-            if section == 'build' and key == 'libname':
-                return 'none'
-            return None
-
-    @lru_cache(maxsize = None)
+    @cache
     def get_value(self, project: str, section: str, key: str) -> str | None:
         ret: str | None
+        if section == 'version':
             proj_dir = self.__proj_dir(project, pretty = False)
             if proj_dir is None:
                 raise Exception(f"Can't get project directory for {project}")
-        if section == 'version':
             proj_version_dirs = [proj_dir]
             if proj_dir != self.___topdir:
                 proj_version_dirs.append('/usr/share/doc/packages/' + project)
@@ -603,8 +547,7 @@ class App(Base):
                     log(DEBUG, f'Ignoring unreadable file "{version_path}"')
                     continue
             raise Exception(f'No version file found for project "{project}"')
-        path = proj_dir + '/make/project.conf'
-        ret = self.read_value(path, section, key)
+        ret = self.__get_project_conf(project).get_str_or_none(section, key)
         log(
             DEBUG,
             'Lookup %s -> %s / [%s%s] -> "%s"' %
@@ -612,7 +555,7 @@ class App(Base):
         )
         return ret
 
-    @lru_cache(maxsize = None)
+    @cache
     def get_version(self, project) -> str:
         ret = self.get_value(project, 'version', '')
         if ret is None:

src/python/jw/pkg/lib/ProjectConf.py

@@ -0,0 +1,238 @@
+from __future__ import annotations
+
+import re
+
+from pathlib import Path
+from typing import ClassVar, Self
+
+class ProjectConf:
+
+    class Error(ValueError):
+        pass
+
+    __SECTION_RE: ClassVar[re.Pattern[str]] = re.compile(
+        r'^\s*\[([^\]]+)\]\s*(?:#.*)?$',
+    )
+
+    @classmethod
+    def __parse_key_value_section(cls, raw_section: str) -> dict[str, str]:
+
+        ret: dict[str, str] = {}
+        pending_line: str | None = None
+
+        for line_number, physical_line in enumerate(
+            raw_section.splitlines(),
+            start = 1,
+        ):
+            line = cls.__strip_comment_outside_quotes(physical_line).rstrip()
+
+            if not line.strip() and pending_line is None:
+                continue
+
+            is_continued = line.endswith('\\')
+
+            if is_continued:
+                line = line[:-1]
+
+            if pending_line is None:
+                pending_line = line
+            else:
+                pending_line += line.lstrip()
+
+            if is_continued:
+                continue
+
+            logical_line = pending_line.strip()
+            pending_line = None
+
+            if not logical_line:
+                continue
+
+            if '=' not in logical_line:
+                raise cls.Error(
+                    f'Line {line_number} is not an assignment: {physical_line!r}',
+                )
+
+            key, value = logical_line.split('=', 1)
+            key = key.strip()
+
+            if not key:
+                raise cls.Error(f'Line {line_number} has an empty key')
+
+            ret[key] = value.strip()
+
+        if pending_line is not None:
+            raise cls.Error('Section ends with an unfinished continuation')
+
+        return ret
+
+    @staticmethod
+    def __strip_comment_outside_quotes(line: str) -> str:
+
+        in_quotes = False
+
+        for index, char in enumerate(line):
+            if char == '"':
+                in_quotes = not in_quotes
+                continue
+
+            if char != '#' or in_quotes:
+                continue
+
+            return line[:index]
+
+        return line
+
+    @classmethod
+    def __split_value(cls, value: str) -> list[str]:
+
+        if value == '':
+            return []
+
+        ret: list[str] = []
+        start = 0
+        in_quotes = False
+
+        for index, char in enumerate(value):
+            if char == '"':
+                in_quotes = not in_quotes
+                continue
+
+            if char != ',' or in_quotes:
+                continue
+
+            elem = cls.__unquote(value[start:index].strip())
+            if elem:
+                ret.append(elem)
+            start = index + 1
+
+        if in_quotes:
+            raise cls.Error(f'Unclosed quote in value: {value!r}')
+
+        last = cls.__unquote(value[start:].strip())
+        if last:
+            ret.append(last)
+        return ret
+
+    @staticmethod
+    def __unquote(value: str) -> str:
+
+        if len(value) < 2:
+            return value
+
+        if value[0] != '"':
+            return value
+
+        if value[-1] != '"':
+            return value
+
+        return value[1:-1]
+
+    # -- Public API
+
+    def __init__(self, sections: dict[str, str]) -> None:
+
+        self.__sections = sections
+        self.__parsed_cache: dict[str, dict[str, str]] = {}
+
+    @classmethod
+    def read(cls, path: str) -> Self:
+
+        text = Path(path).read_text(encoding = 'utf-8')
+
+        sections: dict[str, list[str]] = {}
+        current_section: str | None = None
+        current_lines: list[str] = []
+
+        def __flush() -> None:
+
+            nonlocal current_section, current_lines
+
+            if current_section is not None:
+                sections.setdefault(current_section, []).append(
+                    ''.join(current_lines),
+                )
+
+            current_lines = []
+
+        for line in text.splitlines(keepends = True):
+            match = cls.__SECTION_RE.match(line.rstrip('\r\n'))
+
+            if not match:
+                if current_section is not None:
+                    current_lines.append(line)
+
+                continue
+
+            __flush()
+            current_section = match.group(1).strip()
+
+        __flush()
+
+        ret = cls(
+            {
+                name: ''.join(section_parts)
+                for name, section_parts in sections.items()
+            }
+        )
+
+        return ret
+
+    def get_section(self, section: str) -> str | None:
+
+        return self.__sections.get(section)
+
+    def get_str(self, section: str, key: str) -> str:
+
+        ret = self.get_str_or_none(section, key)
+
+        if ret is None:
+            raise self.Error(f'Missing key {section}.{key}')
+
+        return ret
+
+    def get_str_or_none(self, section: str, key: str) -> str | None:
+
+        value = self.__get_value_or_none(section, key)
+
+        if value is None:
+            return None
+
+        return self.__unquote(value)
+
+    def get_list(self, section: str, key: str) -> list[str]:
+
+        ret = self.get_list_or_none(section, key)
+
+        if ret is None:
+            raise self.Error(f'Missing key {section}.{key}')
+
+        return ret
+
+    def get_list_or_none(self, section: str, key: str) -> list[str] | None:
+
+        value = self.__get_value_or_none(section, key)
+
+        if value is None:
+            return None
+
+        return self.__split_value(value)
+
+    def __get_value_or_none(self, section: str, key: str) -> str | None:
+
+        raw_section = self.__sections.get(section)
+
+        if raw_section is None:
+            return None
+
+        if section not in self.__parsed_cache:
+            try:
+                self.__parsed_cache[section] = self.__parse_key_value_section(
+                    raw_section,
+                )
+            except self.Error as exc:
+                raise self.Error(
+                    f'Section {section!r} cannot be parsed as key-value pairs',
+                ) from exc
+
+        return self.__parsed_cache[section].get(key)

test/unit/python/jw/pkg/lib/ProjectConf/Makefile

@@ -0,0 +1,7 @@
+TOPDIR = ../../../../../../..
+
+include $(TOPDIR)/make/proj.mk
+include $(JWBDIR)/make/py-run.mk
+
+all:
+test: run

test/unit/python/jw/pkg/lib/ProjectConf/test.py

@@ -0,0 +1,253 @@
+import tempfile
+from jw.pkg.lib.ProjectConf import ProjectConf
+
+# -- Helper: create temp file and read it --
+
+def _load(text: str) -> ProjectConf:
+    with tempfile.NamedTemporaryFile(mode='w', suffix='.conf', delete=False) as f:
+        f.write(text)
+        f.flush()
+        return ProjectConf.read(f.name)
+
+# -- Basic construction --
+
+pc = _load('[section]\nkey = value\n')
+assert pc.get_str('section', 'key') == 'value'
+
+# -- get_section returns raw section text --
+
+pc = _load('[meta]\n# header comment\nkey = val\n\n')
+section = pc.get_section('meta')
+assert section is not None
+assert '# header comment' in section
+assert 'key = val' in section
+
+# -- get_section returns None for missing section --
+
+pc = _load('[section]\nkey = value\n')
+assert pc.get_section('nonexistent') is None
+
+# -- get_str returns unquoted value --
+
+pc = _load('[section]\nkey = "hello world"\n')
+assert pc.get_str('section', 'key') == 'hello world'
+
+# -- get_str returns raw value when not quoted --
+
+pc = _load('[section]\nkey = plain value\n')
+assert pc.get_str('section', 'key') == 'plain value'
+
+# -- get_str raises Error when key missing --
+
+pc = _load('[section]\nother = val\n')
+try:
+    pc.get_str('section', 'missing')
+    assert False, 'Should have raised'
+except ProjectConf.Error:
+    pass
+
+# -- get_str raises Error when section missing --
+
+pc = _load('[section]\nkey = val\n')
+try:
+    pc.get_str('nonexistent', 'key')
+    assert False, 'Should have raised'
+except ProjectConf.Error:
+    pass
+
+# -- get_str_or_none returns None for missing key --
+
+pc = _load('[section]\nother = val\n')
+assert pc.get_str_or_none('section', 'missing') is None
+
+# -- get_str_or_none returns None for missing section --
+
+pc = _load('[section]\nkey = val\n')
+assert pc.get_str_or_none('nonexistent', 'key') is None
+
+# -- get_str returns string with trailing whitespace stripped --
+
+pc = _load('[section]\nkey = value  \n')
+assert pc.get_str('section', 'key') == 'value'
+
+# -- Quoted value with spaces preserved --
+
+pc = _load('[section]\nkey = "  spaced  "\n')
+assert pc.get_str('section', 'key') == '  spaced  '
+
+# -- Quoted hash inside quotes is preserved --
+
+pc = _load('[section]\nkey = "value # not a comment"\n')
+assert pc.get_str('section', 'key') == 'value # not a comment'
+
+# -- Inline comment outside quotes --
+
+pc = _load('[section]\nkey = value # this is a comment\n')
+assert pc.get_str('section', 'key') == 'value'
+
+# -- Quoted value containing an inline comment delimiter --
+
+pc = _load('[section]\nkey = "has # inside"\n')
+assert pc.get_str('section', 'key') == 'has # inside'
+
+# -- get_list returns split values --
+
+pc = _load('[section]\nlist = a, b, c\n')
+assert pc.get_list('section', 'list') == ['a', 'b', 'c']
+
+# -- get_list returns single value --
+
+pc = _load('[section]\nlist = single\n')
+assert pc.get_list('section', 'list') == ['single']
+
+# -- get_list_or_none returns None for missing key --
+
+pc = _load('[section]\nother = val\n')
+assert pc.get_list_or_none('section', 'missing') is None
+
+# -- get_list raises Error when missing key --
+
+pc = _load('[section]\nother = val\n')
+try:
+    pc.get_list('section', 'missing')
+    assert False, 'Should have raised'
+except ProjectConf.Error:
+    pass
+
+# -- get_list handles quoted commas --
+
+pc = _load('[section]\nlist = "a, b", c, "d, e"\n')
+assert pc.get_list('section', 'list') == ['a, b', 'c', 'd, e']
+
+# -- get_list returns empty list for empty value --
+
+pc = _load('[section]\nlist = \n')
+assert pc.get_list('section', 'list') == []
+
+# -- get_list_or_none returns empty list for empty value --
+
+pc = _load('[section]\nlist = \n')
+assert pc.get_list_or_none('section', 'list') == []
+
+# -- Unclosed quote in list raises --
+
+pc = _load('[section]\nlist = "unclosed, b\n')
+try:
+    pc.get_list('section', 'list')
+    assert False, 'Should have raised'
+except ProjectConf.Error as exc:
+    assert 'Unclosed quote' in str(exc)
+
+# -- Line continuation --
+
+pc = _load('[section]\nkey = first \\\n  second \\\n  third\n')
+assert pc.get_str('section', 'key') == 'first second third'
+
+# -- Line continuation with inline comment at end --
+
+pc = _load('[section]\nkey = first \\\n  second # comment\n')
+assert pc.get_str('section', 'key') == 'first second'
+
+# -- get_section returns None for non-existent section --
+
+pc = _load('[section]\nkey = val\n')
+assert pc.get_section('does_not_exist') is None
+
+# -- Multiple keys in same section --
+
+pc = _load('[section]\nfirst = one\nsecond = two\nthird = three\n')
+assert pc.get_str('section', 'first') == 'one'
+assert pc.get_str('section', 'second') == 'two'
+assert pc.get_str('section', 'third') == 'three'
+
+# -- get_str_or_none does not raise for missing section --
+
+pc = _load('[section]\nkey = val\n')
+assert pc.get_str_or_none('nope', 'key') is None
+
+# -- get_list_or_none does not raise for missing section --
+
+pc = _load('[section]\nkey = val\n')
+assert pc.get_list_or_none('nope', 'key') is None
+
+# -- get_list_or_none does not raise for missing key --
+
+pc = _load('[section]\nother = val\n')
+assert pc.get_list_or_none('section', 'nope') is None
+
+# -- Empty key raises --
+
+pc = _load('[section]\n= value\n')
+try:
+    pc.get_str('section', '')
+    assert False, 'Should have raised'
+except ProjectConf.Error:
+    pass
+
+# -- Not an assignment raises --
+
+pc = _load('[section]\nnot an assignment\n')
+try:
+    pc.get_str('section', 'not an assignment')
+    assert False, 'Should have raised'
+except ProjectConf.Error:
+    pass
+
+# -- Unfinished continuation raises --
+
+pc = _load('[section]\nkey = value \\\n')
+try:
+    pc.get_str('section', 'key')
+    assert False, 'Should have raised'
+except ProjectConf.Error:
+    pass
+
+# -- Error class is ValueError --
+
+pc = _load('[section]\nkey = val\n')
+try:
+    pc.get_str('missing', 'key')
+except ValueError as e:
+    assert isinstance(e, ValueError)
+else:
+    assert False, 'Should have raised ValueError'
+
+# -- Whitespace around key and value --
+
+pc = _load('[section]\n  key  =  value  \n')
+assert pc.get_str('section', 'key') == 'value'
+
+# -- Whitespace around key in list --
+
+pc = _load('[section]\nlist =  a  ,  b  ,  c  \n')
+assert pc.get_list('section', 'list') == ['a', 'b', 'c']
+
+# -- Multiple sections --
+
+pc = _load('[one]\nkey = val1\n[two]\nkey = val2\n')
+assert pc.get_str('one', 'key') == 'val1'
+assert pc.get_str('two', 'key') == 'val2'
+
+# -- Section with only comments and blanks --
+
+pc = _load('[empty]\n# just a comment\n\n')
+section = pc.get_section('empty')
+assert section is not None
+assert pc.get_str_or_none('empty', 'key') is None
+
+# -- Section with trailing comment on section header line --
+
+pc = _load('[section] # inline comment\nkey = val\n')
+assert pc.get_str('section', 'key') == 'val'
+
+# -- Value that looks like a section header --
+
+pc = _load('[section]\nnot_a_key = [bracket]\n')
+assert pc.get_str('section', 'not_a_key') == '[bracket]'
+
+# -- Multiple values with trailing backslash comments --
+
+pc = _load('[section]\nlist = "a", "b", "c"\n')
+assert pc.get_list('section', 'list') == ['a', 'b', 'c']
+
+print('All ProjectConf tests passed')