import unittest import hashlib import os import base64 import sys from pathlib import Path from unittest.mock import patch, AsyncMock from unittest.mock import MagicMock EMMA_DIR = Path(__file__).resolve().parents[1] / "emma" if str(EMMA_DIR) not in sys.path: sys.path.insert(0, str(EMMA_DIR)) import server # Følger eksisterende mønster KMS_KEY_NAME = "projects/p/locations/l/keyRings/k/cryptoKeys/k" class TestProposeGiteaChange(unittest.IsolatedAsyncioTestCase): def setUp(self): """Set up valid parameters for tests.""" self.valid_params = { "repo": server.GITEA_REPO, "branch": "feature/new-idea", "path": "docs/new-file.md", "new_content": "This is new content.", "base_sha": "a" * 40, "commit_message": "A valid commit message.", } def tearDown(self): patch.stopall() @patch('server.create_gitea_change_plan', new_callable=AsyncMock) @patch('server._get_gitea_file_details', new_callable=AsyncMock) @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock) async def test_rejects_main_branch_before_io(self, mock_resolve_sha, mock_get_details, mock_create_plan): params = self.valid_params | {"branch": "main"} with self.assertRaisesRegex(ValueError, "Direct writes to protected branch 'main' are not allowed."): await server.propose_gitea_change(params) mock_resolve_sha.assert_not_awaited() mock_get_details.assert_not_awaited() mock_create_plan.assert_not_awaited() @patch('server.create_gitea_change_plan', new_callable=AsyncMock) @patch('server._get_gitea_file_details', new_callable=AsyncMock) @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock) async def test_rejects_master_branch_before_io(self, mock_resolve_sha, mock_get_details, mock_create_plan): params = self.valid_params | {"branch": "master"} with self.assertRaisesRegex(ValueError, "Direct writes to protected branch 'master' are not allowed."): await server.propose_gitea_change(params) mock_resolve_sha.assert_not_awaited() mock_get_details.assert_not_awaited() mock_create_plan.assert_not_awaited() @patch("server.create_gitea_change_plan", new_callable=AsyncMock) @patch("server._get_gitea_file_details", new_callable=AsyncMock) @patch("server.resolve_branch_to_commit_sha", new_callable=AsyncMock) @patch("server._validate_admin_gitea_path") async def test_rejects_dot_git_path_via_new_write_policy( self, mock_admin_path_validator, mock_resolve_sha, mock_get_details, mock_create_plan, ): params = self.valid_params | {"path": "some/dir/.git/config"} with self.assertRaisesRegex( ValueError, r"Changes within a '.git' directory are not allowed.", ): await server.propose_gitea_change(params) mock_admin_path_validator.assert_called_once_with(params["path"]) mock_resolve_sha.assert_not_awaited() mock_get_details.assert_not_awaited() mock_create_plan.assert_not_awaited() @patch('server.create_gitea_change_plan', new_callable=AsyncMock) @patch('server._get_gitea_file_details', new_callable=AsyncMock) @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock) async def test_rejects_base_sha_mismatch(self, mock_resolve_sha, mock_get_details, mock_create_plan): mock_resolve_sha.return_value = "c" * 40 # Mismatched SHA with self.assertRaisesRegex(ValueError, "Branch head does not match the supplied base_sha"): await server.propose_gitea_change(self.valid_params) mock_resolve_sha.assert_awaited_once() mock_get_details.assert_not_awaited() mock_create_plan.assert_not_awaited() @patch('server.create_gitea_change_plan', new_callable=AsyncMock) @patch('server._get_gitea_file_details', new_callable=AsyncMock) @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock) async def test_rejects_no_op_diff(self, mock_resolve_sha, mock_get_details, mock_create_plan): mock_resolve_sha.return_value = self.valid_params["base_sha"] mock_get_details.return_value = (self.valid_params["new_content"], "b" * 40) with self.assertRaisesRegex(ValueError, "Proposed content produces no file change."): await server.propose_gitea_change(self.valid_params) mock_resolve_sha.assert_awaited_once() mock_get_details.assert_awaited_once() mock_create_plan.assert_not_awaited() @patch('server.create_gitea_change_plan', new_callable=AsyncMock) @patch("server.kms_v1.KeyManagementServiceAsyncClient") @patch('server._get_gitea_file_details', new_callable=AsyncMock, return_value=("old content", "b" * 40)) @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock) async def test_happy_path_creates_pending_plan(self, mock_resolve_sha, mock_get_details, mock_kms_constructor, mock_create_plan): mock_kms_client = AsyncMock() mock_kms_client.encrypt.return_value = MagicMock(ciphertext=b"encrypted-data") mock_kms_constructor.return_value.__aenter__ = AsyncMock(return_value=mock_kms_client) mock_kms_constructor.return_value.__aexit__ = AsyncMock(return_value=False) mock_resolve_sha.return_value = self.valid_params["base_sha"] with patch.dict(os.environ, {"GITEA_PLAN_KMS_KEY_NAME": KMS_KEY_NAME}): result = await server.propose_gitea_change(self.valid_params) self.assertEqual(result["status"], "PENDING") self.assertEqual(result["repo"], self.valid_params["repo"]) self.assertEqual(result["branch"], self.valid_params["branch"]) self.assertEqual(result["path"], self.valid_params["path"]) self.assertIn("approval_subject_hash", result) expected_hash = hashlib.sha256(self.valid_params["new_content"].encode("utf-8")).hexdigest() self.assertEqual(result["content_hash"], expected_hash) mock_kms_client.encrypt.assert_awaited_once() mock_create_plan.assert_awaited_once() # Verify the object passed to create_gitea_change_plan is the real Pydantic model call_args = mock_create_plan.call_args[0][0] self.assertIsInstance(call_args, server.GiteaChangePlan) self.assertEqual(call_args.repo, self.valid_params["repo"]) self.assertEqual(call_args.branch, self.valid_params["branch"]) self.assertEqual(call_args.path, self.valid_params["path"]) self.assertEqual(call_args.base_sha, self.valid_params["base_sha"]) self.assertEqual(call_args.existing_file_sha, "b" * 40) @patch('server.create_gitea_change_plan', new_callable=AsyncMock) @patch("server.kms_v1.KeyManagementServiceAsyncClient") @patch('server._get_gitea_file_details', new_callable=AsyncMock) @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock) async def test_patch2a_missing_kms_key_fails_after_gitea_reads( self, mock_resolve_sha, mock_get_details, mock_kms_constructor, mock_create_plan ): mock_resolve_sha.return_value = self.valid_params["base_sha"] mock_get_details.return_value = ("old content", "b" * 40) with patch.dict(os.environ, {"GITEA_PLAN_KMS_KEY_NAME": ""}): with self.assertRaisesRegex(RuntimeError, "KMS key is not configured"): await server.propose_gitea_change(self.valid_params) mock_resolve_sha.assert_awaited_once() mock_get_details.assert_awaited_once() mock_kms_constructor.assert_not_called() mock_create_plan.assert_not_awaited() @patch('server.create_gitea_change_plan', new_callable=AsyncMock) @patch("server.kms_v1.KeyManagementServiceAsyncClient") @patch('server._get_gitea_file_details', new_callable=AsyncMock) @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock) async def test_patch2a_invalid_kms_key_fails_before_kms_construction( self, mock_resolve_sha, mock_get_details, mock_kms_constructor, mock_create_plan ): mock_resolve_sha.return_value = self.valid_params["base_sha"] mock_get_details.return_value = ("old content", "b" * 40) with patch.dict(os.environ, {"GITEA_PLAN_KMS_KEY_NAME": "invalid-key-format"}): with self.assertRaisesRegex(RuntimeError, "invalid format"): await server.propose_gitea_change(self.valid_params) mock_resolve_sha.assert_awaited_once() mock_get_details.assert_awaited_once() mock_kms_constructor.assert_not_called() mock_create_plan.assert_not_awaited() @patch('server.create_gitea_change_plan', new_callable=AsyncMock) @patch("server.kms_v1.KeyManagementServiceAsyncClient") @patch('server._get_gitea_file_details', new_callable=AsyncMock, return_value=("old content", "b" * 40)) @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock) async def test_patch2a_empty_kms_ciphertext_is_rejected( self, mock_resolve_sha, mock_get_details, mock_kms_constructor, mock_create_plan ): mock_resolve_sha.return_value = self.valid_params["base_sha"] mock_kms_client = AsyncMock() mock_kms_client.encrypt.return_value = MagicMock(ciphertext=b"") mock_kms_constructor.return_value.__aenter__ = AsyncMock(return_value=mock_kms_client) mock_kms_constructor.return_value.__aexit__ = AsyncMock(return_value=False) with patch.dict(os.environ, {"GITEA_PLAN_KMS_KEY_NAME": KMS_KEY_NAME}): with self.assertRaisesRegex(RuntimeError, "invalid empty ciphertext"): await server.propose_gitea_change(self.valid_params) mock_kms_client.encrypt.assert_awaited_once_with( request={ "name": KMS_KEY_NAME, "plaintext": self.valid_params["new_content"].encode("utf-8"), } ) mock_create_plan.assert_not_awaited() @patch('server.create_gitea_change_plan', new_callable=AsyncMock) @patch("server.kms_v1.KeyManagementServiceAsyncClient") @patch('server._get_gitea_file_details', new_callable=AsyncMock, return_value=("old content", "b" * 40)) @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock) async def test_patch2a_non_bytes_kms_ciphertext_is_rejected( self, mock_resolve_sha, mock_get_details, mock_kms_constructor, mock_create_plan ): mock_resolve_sha.return_value = self.valid_params["base_sha"] mock_kms_client = AsyncMock() mock_kms_client.encrypt.return_value = MagicMock(ciphertext=None) mock_kms_constructor.return_value.__aenter__ = AsyncMock(return_value=mock_kms_client) mock_kms_constructor.return_value.__aexit__ = AsyncMock(return_value=False) with patch.dict(os.environ, {"GITEA_PLAN_KMS_KEY_NAME": KMS_KEY_NAME}): with self.assertRaisesRegex(RuntimeError, "invalid empty ciphertext"): await server.propose_gitea_change(self.valid_params) mock_kms_client.encrypt.assert_awaited_once_with( request={ "name": KMS_KEY_NAME, "plaintext": self.valid_params["new_content"].encode("utf-8"), } ) mock_create_plan.assert_not_awaited() @patch('server.create_gitea_change_plan', new_callable=AsyncMock) @patch("server.kms_v1.KeyManagementServiceAsyncClient") @patch('server._get_gitea_file_details', new_callable=AsyncMock, return_value=("old content", "b" * 40)) @patch('server.resolve_branch_to_commit_sha', new_callable=AsyncMock) async def test_patch2a_successful_encrypted_proposal( self, mock_resolve_sha, mock_get_details, mock_kms_constructor, mock_create_plan ): mock_resolve_sha.return_value = self.valid_params["base_sha"] mock_kms_client = AsyncMock() mock_kms_client.encrypt.return_value = MagicMock(ciphertext=b"encrypted-data") mock_kms_constructor.return_value.__aenter__ = AsyncMock(return_value=mock_kms_client) mock_kms_constructor.return_value.__aexit__ = AsyncMock(return_value=False) with patch.dict(os.environ, {"GITEA_PLAN_KMS_KEY_NAME": KMS_KEY_NAME}): result = await server.propose_gitea_change(self.valid_params) mock_resolve_sha.assert_awaited_once() mock_get_details.assert_awaited_once() mock_kms_client.encrypt.assert_awaited_once_with( request={ "name": KMS_KEY_NAME, "plaintext": self.valid_params["new_content"].encode("utf-8"), } ) mock_create_plan.assert_awaited_once() persisted_plan = mock_create_plan.call_args.args[0] self.assertIsInstance(persisted_plan, server.GiteaChangePlan) self.assertIsInstance( persisted_plan.encrypted_payload, server.EncryptedPayloadV1, ) self.assertEqual(persisted_plan.encrypted_payload.payload_version, "1") self.assertEqual(persisted_plan.encrypted_payload.kms_key_name, KMS_KEY_NAME) self.assertEqual( persisted_plan.encrypted_payload.ciphertext, base64.b64encode(b"encrypted-data").decode("ascii"), ) self.assertNotIn("new_content", persisted_plan.model_dump(mode="json")) expected_content_hash = hashlib.sha256(self.valid_params["new_content"].encode("utf-8")).hexdigest() self.assertEqual(persisted_plan.content_hash, expected_content_hash) self.assertEqual(result["content_hash"], expected_content_hash) self.assertEqual( persisted_plan.approval_subject_hash, server._calculate_approval_subject_hash(persisted_plan), ) self.assertEqual(result["approval_subject_hash"], persisted_plan.approval_subject_hash) for field in ("new_content", "encrypted_payload", "ciphertext", "kms_key_name"): self.assertNotIn(field, result) def test_legacy_plan_parses_without_patch2a_fields(self): current_plan = server.GiteaChangePlan( repo="o/r", branch="b", path="p", base_sha="a" * 40, content_hash="c" * 64, commit_message="m", unified_diff="d" ) legacy_dict = current_plan.model_dump(mode="json") legacy_dict.pop("encrypted_payload", None) legacy_dict.pop("approval_subject_hash", None) legacy_plan = server.GiteaChangePlan(**legacy_dict) self.assertIsNone(legacy_plan.encrypted_payload) self.assertIsNone(legacy_plan.approval_subject_hash) def test_malformed_base64_in_hash_raises_error(self): plan_with_bad_payload = server.GiteaChangePlan( plan_id="plan-123", repo="o/r", branch="b", path="p", base_sha="a"*40, content_hash="c"*64, commit_message="m", unified_diff="d", encrypted_payload=server.EncryptedPayloadV1( kms_key_name=KMS_KEY_NAME, ciphertext="not-valid-base64!" ) ) with self.assertRaisesRegex(ValueError, "Encrypted payload ciphertext is invalid"): server._calculate_approval_subject_hash(plan_with_bad_payload) def test_approval_subject_hash_sensitivity(self): base_plan = server.GiteaChangePlan( plan_id="plan-123", repo="o/r", branch="b", path="p", base_sha="a" * 40, existing_file_sha="b" * 40, content_hash="c" * 64, commit_message="m", unified_diff="d", encrypted_payload=server.EncryptedPayloadV1( kms_key_name=KMS_KEY_NAME, ciphertext=base64.b64encode(b"secret").decode("ascii"), ), ) base_hash = server._calculate_approval_subject_hash(base_plan) fields_to_test = { "plan_id": "plan-456", "repo": "o/r2", "branch": "b2", "path": "p2", "base_sha": "A" * 40, "existing_file_sha": "B" * 40, "content_hash": "C" * 64, "commit_message": "m2", "unified_diff": "d2", } for field, new_value in fields_to_test.items(): with self.subTest(field=field): modified_plan = base_plan.model_copy(update={field: new_value}) new_hash = server._calculate_approval_subject_hash(modified_plan) self.assertNotEqual(base_hash, new_hash) with self.subTest(field="kms_key_name"): modified_plan_kms = base_plan.model_copy(deep=True) modified_plan_kms.encrypted_payload.kms_key_name = "projects/p/locations/l/keyRings/k/cryptoKeys/k2" self.assertNotEqual(base_hash, server._calculate_approval_subject_hash(modified_plan_kms)) with self.subTest(field="ciphertext"): modified_plan_cipher = base_plan.model_copy(deep=True) modified_plan_cipher.encrypted_payload.ciphertext = base64.b64encode(b"secret2").decode("ascii") self.assertNotEqual(base_hash, server._calculate_approval_subject_hash(modified_plan_cipher)) if __name__ == "__main__": unittest.main()