feat: create debian packages without external processes

src/ops2deb/builder.py

@@ -1,11 +1,15 @@
 import asyncio
-import re
+import tarfile
 from pathlib import Path
 from typing import Any
 
+import unix_ar
+from debian.deb822 import Deb822
+
 from ops2deb import logger
 from ops2deb.exceptions import Ops2debBuilderError
-from ops2deb.utils import log_and_raise
+from ops2deb.templates import environment
+from ops2deb.utils import log_and_raise, working_directory
 
 
 def parse_debian_control(cwd: Path) -> dict[str, str]:
@@ -14,28 +18,97 @@ def parse_debian_control(cwd: Path) -> dict[str, str]:
     :param cwd: Path to debian source package
     :return: Dict object with fields as keys
     """
-    field_re = re.compile(r"^([\w-]+)\s*:\s*(.+)")
 
-    content = (cwd / "debian" / "control").read_text()
-    control = {}
-    for line in content.split("\n"):
-        m = field_re.search(line)
-        if m:
-            g = m.groups()
-            control[g[0]] = g[1]
+    raw_control = (cwd / "debian" / "control").read_text()
+    parsed_control: dict[str, str] = {}
+    for paragraph in Deb822.iter_paragraphs(raw_control):
+        parsed_control.update(paragraph)
+    parsed_control.pop("Source")
+    return parsed_control
+
+
+def _make_tar_xz_archive(archive_path: Path, base_path: Path) -> None:
+    archive_path.parent.mkdir(exist_ok=True, parents=True)
+    tar_compression = "xz"
+
+    def _set_uid_gid(tarinfo: Any) -> Any:
+        tarinfo.gid = 0
+        tarinfo.gname = "root"
+        tarinfo.uid = 0
+        tarinfo.uname = "root"
+        return tarinfo
+
+    tar = tarfile.open(archive_path, "w|%s" % tar_compression)
+
+    try:
+        with working_directory(base_path):
+            tar.add(".", filter=_set_uid_gid)
+    finally:
+        tar.close()
 
-    return control
 
+def _make_ar_archive(archive_path: Path, base_path: Path) -> None:
+    if archive_path.is_file():
+        archive_path.unlink()
+    else:
+        archive_path.parent.mkdir(exist_ok=True, parents=True)
+    ar = unix_ar.open(str(archive_path), "w")
+    try:
+        with working_directory(base_path):
+            for file in Path(".").glob("*"):
+                ar.addfile(str(file))
+    finally:
+        ar.close()
+
+
+def _build_package_with_python(package_path: Path) -> None:
+    """
+    Simple implementation of a debian package builder
+    Can be used instead of dpkg-buildpackage to speed up
+    the build process.
+    """
+
+    build_path = Path("/tmp/ops2deb_builder") / package_path.name
 
-async def build_package(cwd: Path) -> None:
+    control_path = build_path / "control"
+    control_path.mkdir(exist_ok=True, parents=True)
+    control = parse_debian_control(package_path)
+    template = environment.get_template("package-control")
+    template.stream(control=control).dump(str(control_path / "control"))
+
+    (control_path / "md5sums").write_text(
+        """ff417446446fd12c5b6164eec36c04c7  usr/bin/ops2deb
+37693f15b118085f502da70d74de9f6b  usr/share/doc/ops2deb/changelog.Debian.gz
+f68c33e9d9a8bdffc617e8ddb2fb94b8  usr/share/lintian/overrides/ops2deb
+"""
+    )
+
+    # create content of .deb tar archive
+    tar_path = build_path / "tar"
+    tar_path.mkdir(exist_ok=True, parents=True)
+    (tar_path / "debian-binary").write_text("2.0\n")
+    _make_tar_xz_archive(tar_path / "data.tar.xz", package_path / "src")
+    _make_tar_xz_archive(tar_path / "control.tar.xz", control_path)
+
+    # create final .deb tar archive
+    version = "1.0.0-1"
+    control = parse_debian_control(package_path)
+    package_name = f'{control["Package"]}_{version}_{control["Architecture"]}'
+    _make_ar_archive(package_path.parent / f"{package_name}.tar", tar_path)
+
+
+async def build_package_with_python(package_path: Path) -> None:
+    loop = asyncio.get_running_loop()
+    await loop.run_in_executor(None, _build_package_with_python, package_path)
+
+
+async def build_package_with_dpkgbuildpackage(cwd: Path) -> None:
     """Run dpkg-buildpackage in specified path."""
     args = ["-us", "-uc"]
     arch = parse_debian_control(cwd)["Architecture"]
     if arch != "all":
         args += ["--host-arch", arch]
 
-    logger.info(f"Building {cwd}...")
-
     proc = await asyncio.create_subprocess_exec(
         "/usr/bin/dpkg-buildpackage",
         *args,
@@ -52,8 +125,6 @@ async def build_package(cwd: Path) -> None:
 
     if proc.returncode:
         log_and_raise(Ops2debBuilderError(f"Failed to build package in {str(cwd)}"))
-    else:
-        logger.info(f"Successfully built {str(cwd)}")
 
 
 def build(package_paths: list[Path], workers: int) -> None:
@@ -65,14 +136,20 @@ def build(package_paths: list[Path], workers: int) -> None:
 
     logger.title(f"Building {len(package_paths)} source packages...")
 
+    async def _builder(package_path: Path) -> None:
+        logger.info(f"Building {package_path}...")
+        # await build_package_with_dpkgbuildpackage(package_path)
+        await build_package_with_python(package_path)
+        logger.info(f"Successfully built {package_path}")
+
     async def _build_package(sem: asyncio.Semaphore, _path: Path) -> None:
         async with sem:  # semaphore limits num of simultaneous builds
-            await build_package(_path)
+            await _builder(_path)
 
     async def _build_packages() -> Any:
         sem = asyncio.Semaphore(workers)
         return await asyncio.gather(
-            *[_build_package(sem, p) for p in package_paths], return_exceptions=True
+            *[_build_package(sem, p) for p in package_paths], return_exceptions=False
         )
 
     results = asyncio.run(_build_packages())

src/ops2deb/templates.py

@@ -60,6 +60,22 @@
 """
 
 
+DEBIAN_PACKAGE_CONTROL = """\
+Package: {{ control.Package }}
+Version: 0.18.0-1~ops2deb
+Achitecture: {{ control.Architecture }}
+Maintainer: {{ control.Maintainer }}
+Installed-Size: 11367
+Priority: {{ control.Priority }}
+{%- if control.Provides %}{{ '\n' }}Provides: {{ control.Provides }}{% endif %}
+{%- if control.Depends %}{{ '\n' }}Depends: {{ control.Depends }}{% endif %}
+{%- if control.Recommends %}{{ '\n' }}Recommends: {{ control.Recommends }}{% endif %}
+{%- if control.Replaces %}{{ '\n' }}Replaces: {{ control.Replaces }}{% endif %}
+{%- if control.Conflicts %}{{ '\n' }}Conflicts: {{ control.Conflicts }}{% endif %}
+Description: {% for line in control.Description.split('\n') %} {{ line or '.' }}{{ '\n' if not loop.last else '' }}{% endfor %}
+"""
+
+
 def template_loader(name: str) -> Optional[str]:
     variable_name = f"DEBIAN_{name.upper().replace('-', '_')}"
     template_content: str = globals()[variable_name]

tests/test_builder.py

@@ -0,0 +1,46 @@
+from textwrap import dedent
+
+from ops2deb.builder import parse_debian_control
+
+
+def test_parse_debian_control__should_return_dict_with_all_fieds_from_debian_control(
+    tmp_path,
+):
+    raw_control = dedent(
+        """\
+    Source: great-app
+    Priority: optional
+    Maintainer: ops2deb <[email protected]>
+    Build-Depends: debhelper, build-dep-1, build-dep-2
+    Standards-Version: 3.9.6
+    Homepage: http://great-app.io
+
+    Package: great-app
+    Architecture: all
+    Provides: virtual_package
+    Depends: package_a
+    Recommends: package_b
+    Replaces: package_c
+    Conflicts: package_d
+    Description: My great app
+     A detailed description of the super package
+    """
+    )
+    (tmp_path / "debian").mkdir()
+    (tmp_path / "debian" / "control").write_text(raw_control)
+    control = parse_debian_control(tmp_path)
+    assert control == {
+        "Priority": "optional",
+        "Maintainer": "ops2deb <[email protected]>",
+        "Build-Depends": "debhelper, build-dep-1, build-dep-2",
+        "Standards-Version": "3.9.6",
+        "Homepage": "http://great-app.io",
+        "Package": "great-app",
+        "Architecture": "all",
+        "Provides": "virtual_package",
+        "Depends": "package_a",
+        "Recommends": "package_b",
+        "Replaces": "package_c",
+        "Conflicts": "package_d",
+        "Description": "My great app\n A detailed description of the super package",
+    }