diff --git a/src/omnibase_infra/onboarding/config_writer.py b/src/omnibase_infra/onboarding/config_writer.py index 06dffcad77..9a3aac97c6 100644 --- a/src/omnibase_infra/onboarding/config_writer.py +++ b/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,7 +37,7 @@ 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( @@ -44,18 +45,7 @@ def render( 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"] diff --git a/tests/integration/test_config_writer_integration.py b/tests/integration/test_config_writer_integration.py index 6ab5343aea..8b59a2c13b 100644 --- a/tests/integration/test_config_writer_integration.py +++ b/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 diff --git a/tests/unit/onboarding/test_config_writer.py b/tests/unit/onboarding/test_config_writer.py index 474cebbf37..07eb0898f7 100644 --- a/tests/unit/onboarding/test_config_writer.py +++ b/tests/unit/onboarding/test_config_writer.py @@ -1,7 +1,7 @@ # 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 @@ -9,11 +9,17 @@ 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,8 +127,27 @@ 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" @@ -142,23 +155,14 @@ def test_real_home_omnibase_untouched(self, tmp_path: Path) -> None: 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: