sovereign_ai/SovereignAI-Starter/tests/test_workspace_changes.py

"""Tests for preview-only workspace edits and explicit one-time application."""

from __future__ import annotations

import tempfile
import unittest
import os
import json
from pathlib import Path
from unittest.mock import patch

from app import auth, workspace


class WorkspaceChangeTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temporary = tempfile.TemporaryDirectory()
        self.root = Path(self.temporary.name).resolve()
        (self.root / "src").mkdir()

    def tearDown(self) -> None:
        self.temporary.cleanup()

    def test_create_preview_does_not_write_until_apply_and_is_one_time(self) -> None:
        preview = workspace.create_change_preview(
            self.root,
            "src/new.py",
            "create",
            "print('hello')\n",
        )
        target = self.root / "src" / "new.py"

        self.assertFalse(target.exists())
        self.assertIn("+print('hello')", str(preview["diff"]))

        result = workspace.apply_change_preview(str(preview["token"]))
        self.assertEqual(result["status"], "applied")
        self.assertEqual(target.read_text(encoding="utf-8"), "print('hello')\n")
        with self.assertRaisesRegex(ValueError, "انتهت صلاحية"):
            workspace.apply_change_preview(str(preview["token"]))

    def test_preview_can_only_be_applied_by_its_owner(self) -> None:
        target = self.root / "src" / "owned.py"
        target.write_text("before = True\n", encoding="utf-8")
        preview = workspace.create_change_preview(
            self.root,
            "src/owned.py",
            "update",
            "after = True\n",
            user_id="account-one",
        )

        with self.assertRaisesRegex(ValueError, "لا تخص جلسة"):
            workspace.apply_change_preview(
                str(preview["token"]), user_id="account-two"
            )
        self.assertEqual(target.read_text(encoding="utf-8"), "before = True\n")
        workspace.apply_change_preview(str(preview["token"]), user_id="account-one")
        self.assertEqual(target.read_text(encoding="utf-8"), "after = True\n")

    def test_update_preview_detects_external_changes_before_apply(self) -> None:
        target = self.root / "src" / "existing.py"
        target.write_text("old = 1\n", encoding="utf-8")
        preview = workspace.create_change_preview(
            self.root,
            "src/existing.py",
            "update",
            "new = 2\n",
        )
        self.assertIn("-old = 1", str(preview["diff"]))
        self.assertIn("+new = 2", str(preview["diff"]))

        target.write_text("external = 3\n", encoding="utf-8")
        with self.assertRaisesRegex(ValueError, "تغير الملف"):
            workspace.apply_change_preview(str(preview["token"]))
        self.assertEqual(target.read_text(encoding="utf-8"), "external = 3\n")

    def test_rejects_escape_hidden_unsupported_and_missing_parent_paths(self) -> None:
        invalid = [
            ("../outside.py", "create", "escape"),
            (".env", "create", "hidden"),
            ("src/image.png", "create", "unsupported"),
            ("missing/new.py", "create", "parent does not exist"),
        ]
        for relative_path, operation, _reason in invalid:
            with self.subTest(path=relative_path):
                with self.assertRaises(ValueError):
                    workspace.create_change_preview(
                        self.root, relative_path, operation, "content\n"
                    )

    def test_selected_workspace_must_be_under_server_allowlisted_roots(self) -> None:
        allowed = self.root / "allowed"
        nested = allowed / "project"
        nested.mkdir(parents=True)
        outside = self.root / "outside"
        outside.mkdir()
        with patch.dict(
            os.environ,
            {"SOVEREIGNAI_ALLOWED_WORKSPACES": str(allowed)},
            clear=False,
        ):
            with self.assertRaisesRegex(ValueError, "خارج مساحة العمل"):
                workspace.selected_root(str(outside))
            self.assertEqual(workspace.selected_root(str(nested)), nested.resolve())

    def test_account_workspaces_are_separate_and_admin_config_must_not_overlap(self) -> None:
        allowed = self.root / "accounts"
        first_root = allowed / "first"
        second_root = allowed / "second"
        first_root.mkdir(parents=True)
        second_root.mkdir()
        config = {
            "first@example.test": [str(first_root)],
            "second@example.test": [str(second_root)],
        }
        with (
            patch.dict(
                os.environ,
                {
                    "SOVEREIGNAI_ALLOWED_WORKSPACES": str(allowed),
                    "SOVEREIGNAI_USER_WORKSPACES": json.dumps(config),
                },
                clear=False,
            ),
            patch.object(
                auth,
                "account_email",
                side_effect=lambda user_id: {
                    "user-first": "first@example.test",
                    "user-second": "second@example.test",
                }.get(user_id),
            ),
        ):
            self.assertEqual(
                workspace.selected_root(str(first_root), user_id="user-first"),
                first_root.resolve(),
            )
            with self.assertRaisesRegex(ValueError, "خارج مساحة العمل"):
                workspace.selected_root(str(second_root), user_id="user-first")
            config["second@example.test"] = [str(allowed)]
            os.environ["SOVEREIGNAI_USER_WORKSPACES"] = json.dumps(config)
            with self.assertRaisesRegex(ValueError, "متداخلة"):
                workspace.selected_root(str(first_root), user_id="user-first")

    def test_rejects_create_overwrite_and_update_of_missing_file(self) -> None:
        target = self.root / "src" / "existing.py"
        target.write_text("value = 1\n", encoding="utf-8")
        with self.assertRaisesRegex(ValueError, "موجود بالفعل"):
            workspace.create_change_preview(
                self.root, "src/existing.py", "create", "value = 2\n"
            )
        with self.assertRaisesRegex(ValueError, "غير موجود"):
            workspace.create_change_preview(
                self.root, "src/missing.py", "update", "value = 2\n"
            )


if __name__ == "__main__":
    unittest.main()