feat(OMN-10773): atomic env file writer for interactive onboarding config output

src/omnibase_infra/onboarding/config_writer.py

@@ -1,9 +1,10 @@
 # SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
 # SPDX-License-Identifier: MIT
 
-"""Config writer for onboarding-generated env files (OMN-10783).
+"""Config writer for onboarding-generated env files.
 
-Separate, explicit invocation only — never called automatically.
+Separate, explicit invocation only: callers choose when to write, and dry-run
+paths should call ``ConfigWriter.render`` instead of touching the filesystem.
 """
 
 from __future__ import annotations
@@ -15,7 +16,7 @@
 
 
 class ConfigWriterError(ValueError):
-    """Raised when a key or value contains characters that would corrupt the env format."""
+    """Raised when a key or value contains characters that would corrupt env output."""
 
 
 def _validate_env_pair(key: str, value: str) -> None:
@@ -36,26 +37,15 @@ def _validate_env_pair(key: str, value: str) -> None:
 class ConfigWriter:
     """Writes env key=value files with merge-and-preserve semantics.
 
-    Explicit invocation only. Never write to ~/.omnibase/ from tests.
+    Explicit invocation only. Never write to ``~/.omnibase/`` from tests.
     """
 
     def render(
         self,
         env_dict: dict[str, str],
         existing_content: str | None = None,
     ) -> str:
-        """Pure render — no file I/O.
-
-        Merges env_dict into existing_content (if any), preserving keys not
-        present in env_dict and overwriting keys that are.
-
-        Args:
-            env_dict: New key=value pairs to write.
-            existing_content: Existing file content to merge with, or None.
-
-        Returns:
-            Merged content as a string.
-        """
+        """Return merged env content without writing to disk."""
         merged: dict[str, str] = {}
 
         if existing_content:
@@ -67,27 +57,16 @@ def render(
                     key, _, value = stripped.partition("=")
                     merged[key.strip()] = value.strip()
 
-        for k, v in {**merged, **env_dict}.items():
-            _validate_env_pair(k, v)
+        for key, value in {**merged, **env_dict}.items():
+            _validate_env_pair(key, value)
 
         merged.update(env_dict)
 
-        lines = [f"{k}={v}" for k, v in sorted(merged.items())]
+        lines = [f"{key}={value}" for key, value in sorted(merged.items())]
         return "\n".join(lines) + ("\n" if lines else "")
 
     def write(self, env_dict: dict[str, str], target_path: Path) -> str:
-        """Merge env_dict with existing file and atomically write result.
-
-        Preserves keys not in env_dict; overwrites keys that are.
-        Uses tmp-file + rename for atomicity.
-
-        Args:
-            env_dict: New key=value pairs to write.
-            target_path: Destination path.
-
-        Returns:
-            Merged content as a string.
-        """
+        """Merge env_dict with an existing file and atomically write the result."""
         existing_content: str | None = None
         if target_path.exists():
             existing_content = target_path.read_text(encoding="utf-8")
@@ -101,8 +80,8 @@ def write(self, env_dict: dict[str, str], target_path: Path) -> str:
             prefix=f".{target_path.name}.tmp.",
         )
         try:
-            with os.fdopen(fd, "w", encoding="utf-8") as fh:
-                fh.write(content)
+            with os.fdopen(fd, "w", encoding="utf-8") as file_handle:
+                file_handle.write(content)
             Path(tmp_path).replace(target_path)
         except Exception:
             with suppress(OSError):
@@ -112,4 +91,9 @@ def write(self, env_dict: dict[str, str], target_path: Path) -> str:
         return content
 
 
-__all__ = ["ConfigWriter", "ConfigWriterError"]
+def write_env_file(env_dict: dict[str, str], target_path: Path) -> str:
+    """Explicit convenience wrapper for callers that do not need a writer instance."""
+    return ConfigWriter().write(env_dict, target_path)
+
+
+__all__ = ["ConfigWriter", "ConfigWriterError", "write_env_file"]

tests/integration/test_config_writer_integration.py

@@ -13,7 +13,11 @@
 
 import pytest
 
-from omnibase_infra.onboarding.config_writer import ConfigWriter, ConfigWriterError
+from omnibase_infra.onboarding.config_writer import (
+    ConfigWriter,
+    ConfigWriterError,
+    write_env_file,
+)
 
 pytestmark = pytest.mark.integration
 
@@ -85,3 +89,17 @@ def test_rejected_pair_does_not_corrupt_existing_file(
             writer.write({"BAD=KEY": "value\nINJECTED=evil"}, target)
 
         assert target.read_text(encoding="utf-8") == original
+
+    def test_write_env_file_wrapper_merges_on_real_filesystem(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        target = tmp_path / "omnibase.env"
+        target.write_text("KEEP=existing\nMODE=old\n", encoding="utf-8")
+
+        returned = write_env_file({"MODE": "interactive"}, target)
+
+        assert returned == target.read_text(encoding="utf-8")
+        assert "KEEP=existing" in returned
+        assert "MODE=interactive" in returned
+        assert "MODE=old" not in returned

tests/unit/onboarding/test_config_writer.py

@@ -1,19 +1,25 @@
 # SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
 # SPDX-License-Identifier: MIT
 
-"""Tests for ConfigWriter (OMN-10783)."""
+"""Tests for ConfigWriter and write_env_file."""
 
 from __future__ import annotations
 
 from pathlib import Path
 
 import pytest
 
-from omnibase_infra.onboarding.config_writer import ConfigWriter, ConfigWriterError
+from omnibase_infra.onboarding.config_writer import (
+    ConfigWriter,
+    ConfigWriterError,
+    write_env_file,
+)
+
+pytestmark = pytest.mark.unit
 
 
 class TestConfigWriterRender:
-    """Tests for the pure render method (no I/O)."""
+    """Tests for the pure render method."""
 
     def test_render_empty_dict_no_existing(self) -> None:
         writer = ConfigWriter()
@@ -53,23 +59,12 @@ def test_render_skips_comments_in_existing(self) -> None:
         writer = ConfigWriter()
         existing = "# This is a comment\nKEY=value\n"
         result = writer.render({"KEY": "value"}, existing_content=existing)
-        # Comments are not preserved in the merged output (pure key=value format)
         assert "# This is a comment" not in result
         assert "KEY=value" in result
 
-    def test_render_none_existing_content(self) -> None:
-        writer = ConfigWriter()
-        result = writer.render({"A": "1"}, existing_content=None)
-        assert "A=1" in result
-
-    def test_render_returns_string(self) -> None:
-        writer = ConfigWriter()
-        result = writer.render({"X": "y"})
-        assert isinstance(result, str)
-
 
 class TestConfigWriterWrite:
-    """Tests for the write method (file I/O with atomicity guarantees)."""
+    """Tests for file I/O with atomicity guarantees."""
 
     def test_write_creates_file(self, tmp_path: Path) -> None:
         writer = ConfigWriter()
@@ -113,7 +108,6 @@ def test_write_atomic_no_tmp_file_left_on_success(self, tmp_path: Path) -> None:
         writer = ConfigWriter()
         target = tmp_path / "test.env"
         writer.write({"KEY": "value"}, target)
-        # No .tmp. files should remain after successful write
         tmp_files = list(tmp_path.glob(f".{target.name}.tmp.*"))
         assert not tmp_files, f"Leftover tmp files: {tmp_files}"
 
@@ -133,32 +127,42 @@ def test_write_empty_dict_with_existing_preserves_all(self, tmp_path: Path) -> N
         assert "B=2" in content
 
 
+class TestWriteEnvFile:
+    """Tests for the convenience function retained by OMN-10773."""
+
+    def test_write_env_file_uses_config_writer_semantics(self, tmp_path: Path) -> None:
+        target = tmp_path / "test.env"
+        target.write_text("EXISTING=old\nKEY=old\n", encoding="utf-8")
+
+        content = write_env_file({"KEY": "new"}, target)
+
+        assert "EXISTING=old" in content
+        assert "KEY=new" in content
+        assert "old\nKEY=old" not in content
+        assert target.read_text(encoding="utf-8") == content
+
+    def test_write_env_file_rejects_unsafe_pairs(self, tmp_path: Path) -> None:
+        with pytest.raises(ConfigWriterError, match="newline"):
+            write_env_file({"KEY": "value\nINJECTED=evil"}, tmp_path / "test.env")
+
+
 class TestConfigWriterSafety:
-    """Safety assertions — tests must never write under real ~/.omnibase/."""
+    """Safety assertions: tests must never write under real ~/.omnibase/."""
 
     def test_real_home_omnibase_untouched(self, tmp_path: Path) -> None:
         real_omnibase = Path.home() / ".omnibase"
         writer = ConfigWriter()
         target = tmp_path / "safe.env"
         writer.write({"SAFE": "yes"}, target)
 
-        # Assert the real .omnibase dir was not created by this test
-        # (it may exist from prior system state, but we verify no .env was modified)
         if real_omnibase.exists():
-            # If it already exists, verify target was written to tmp_path, not home
             assert target.parent == tmp_path
         else:
-            # If it doesn't exist, confirm we didn't create it
-            assert not real_omnibase.exists(), (
-                "ConfigWriter must never write under ~/.omnibase/; "
-                "all test targets must use tmp_path"
-            )
+            assert not real_omnibase.exists()
 
     def test_tmp_path_is_not_under_home_omnibase(self, tmp_path: Path) -> None:
         real_omnibase = Path.home() / ".omnibase"
-        assert not str(tmp_path).startswith(str(real_omnibase)), (
-            "tmp_path must not resolve under ~/.omnibase/"
-        )
+        assert not str(tmp_path).startswith(str(real_omnibase))
 
 
 class TestConfigWriterValidation: