hermes-webui/tests/test_issue1164_env_file_corruption.py

"""
Regression tests for #1164 — .env file corruption by WebUI.

The WebUI's onboarding flow had a duplicate _write_env_file() without
_ENV_LOCK protection. Concurrent writes (e.g. Telegram bot + WebUI) could
corrupt the shared .env file. Additionally, both _write_env_file copies
rewrote the entire file from a parsed dict, stripping comments and
reordering keys alphabetically.

Fix:
- onboarding.py now imports _write_env_file from providers.py (which holds
  _ENV_LOCK from api.streaming for the entire load→modify→write cycle).
- _write_env_file in providers.py now preserves comments, blank lines, and
  original key order instead of rebuilding from a sorted dict.

Sprint/commit: v0.50.227+
"""
import os
import tempfile
import textwrap
import unittest
from pathlib import Path


class TestEnvFileCommentPreservation(unittest.TestCase):
    """Verify _write_env_file preserves comments, blank lines, and key order."""

    def setUp(self):
        self.tmpdir = tempfile.mkdtemp()
        self.env_path = Path(self.tmpdir) / ".env"
        # Must import AFTER setting up, as the module has top-level code
        from api.providers import _write_env_file
        self._write_env_file = _write_env_file

    def tearDown(self):
        # Clean os.environ entries set during tests
        for key in ("OPENROUTER_API_KEY", "OPENAI_API_KEY", "NEW_KEY"):
            os.environ.pop(key, None)
        import shutil
        shutil.rmtree(self.tmpdir, ignore_errors=True)

    def _read(self) -> str:
        return self.env_path.read_text(encoding="utf-8")

    # ── Comment preservation ──────────────────────────────────────────

    def test_comments_preserved_on_update(self):
        """Comments in .env must survive a key value update."""
        self.env_path.write_text(textwrap.dedent("""\
            # Hermes API keys
            OPENROUTER_API_KEY=sk-or-old
            # Another comment
            OPENAI_API_KEY=sk-oai-old
        """).strip() + "\n", encoding="utf-8")

        self._write_env_file(self.env_path, {"OPENROUTER_API_KEY": "sk-or-new"})

        content = self._read()
        self.assertIn("# Hermes API keys", content,
                      "Leading comment must be preserved")
        self.assertIn("# Another comment", content,
                      "Inline comment must be preserved")
        self.assertIn("sk-or-new", content)

    def test_blank_lines_preserved(self):
        """Blank lines between key blocks must be preserved."""
        self.env_path.write_text(
            "KEY_A=val_a\n\nKEY_B=val_b\n", encoding="utf-8")

        self._write_env_file(self.env_path, {"KEY_A": "updated"})

        content = self._read()
        self.assertEqual(content.count("\n\n"), 1,
                         "Blank line between keys must be preserved")

    def test_key_order_preserved(self):
        """Original key order must not be sorted alphabetically."""
        self.env_path.write_text(
            "ZZZ_KEY=last\nAAA_KEY=first\nBBB_KEY=middle\n",
            encoding="utf-8")

        self._write_env_file(self.env_path, {"AAA_KEY": "updated"})

        content = self._read()
        zzz_pos = content.find("ZZZ_KEY")
        aaa_pos = content.find("AAA_KEY")
        bbb_pos = content.find("BBB_KEY")
        # Original order: ZZZ, AAA, BBB
        self.assertLess(zzz_pos, aaa_pos,
                        "ZZZ_KEY must still come before AAA_KEY (original order)")
        self.assertLess(aaa_pos, bbb_pos,
                        "AAA_KEY must still come before BBB_KEY (original order)")

    def test_new_key_appended_with_separator(self):
        """New keys are appended at the end with a blank-line separator."""
        self.env_path.write_text(
            "EXISTING_KEY=value\n", encoding="utf-8")

        self._write_env_file(self.env_path, {"NEW_KEY": "new_value"})

        content = self._read()
        self.assertIn("NEW_KEY=new_value", content)
        # New key should appear after the existing one
        self.assertGreater(content.find("NEW_KEY"), content.find("EXISTING_KEY"))

    def test_key_removal_preserves_others(self):
        """Removing a key leaves other keys and comments intact."""
        self.env_path.write_text(textwrap.dedent("""\
            # Comment A
            KEY_A=val_a
            # Comment B
            KEY_B=val_b
        """).strip() + "\n", encoding="utf-8")

        self._write_env_file(self.env_path, {"KEY_B": None})

        content = self._read()
        self.assertIn("KEY_A=val_a", content)
        self.assertIn("# Comment A", content)
        self.assertNotIn("KEY_B", content)
        # Comment B stays (it's just a comment, not tied to KEY_B structurally)
        self.assertIn("# Comment B", content)

    def test_empty_file_handled_gracefully(self):
        """Writing to a non-existent .env file works."""
        self.assertFalse(self.env_path.exists())
        self._write_env_file(self.env_path, {"NEW_KEY": "value"})
        self.assertTrue(self.env_path.exists())
        self.assertEqual(self._read().strip(), "NEW_KEY=value")


class TestOnboardingUsesProviderWriteEnv(unittest.TestCase):
    """Verify that onboarding.py delegates to providers._write_env_file
    (which holds _ENV_LOCK), eliminating the duplicate unprotected path."""

    def test_onboarding_imports_write_env_from_providers(self):
        """api.onboarding._write_env_file must be the same object as
        api.providers._write_env_file (shared implementation with lock)."""
        from api import onboarding, providers
        self.assertIs(
            onboarding._write_env_file,
            providers._write_env_file,
            "onboarding must use providers._write_env_file for thread safety (#1164)"
        )

    def test_providers_write_env_holds_env_lock(self):
        """providers._write_env_file must acquire _ENV_LOCK from api.streaming."""
        import inspect
        from api.providers import _write_env_file
        source = inspect.getsource(_write_env_file)
        self.assertIn("_ENV_LOCK", source,
                      "_write_env_file must use _ENV_LOCK for concurrency safety")
        self.assertIn("from api.streaming import _ENV_LOCK", source,
                      "_ENV_LOCK must be imported from api.streaming")

    def test_providers_write_env_uses_atomic_rename(self):
        """providers._write_env_file must write atomically via tempfile +
        os.replace so cross-process readers (Telegram, CLI) never observe
        a truncated half-written file (#1164 cross-process leg)."""
        import inspect
        from api.providers import _write_env_file
        source = inspect.getsource(_write_env_file)
        self.assertIn("tempfile", source,
                      "_write_env_file must stage writes through a tempfile")
        self.assertIn("os.replace(", source,
                      "_write_env_file must atomically rename via os.replace")
        # The original O_TRUNC pattern must NOT remain — it is the source of
        # the cross-process race the PR is closing.
        self.assertNotIn("O_TRUNC", source,
                         "_write_env_file must not truncate-in-place (#1164)")