opsxsql/tests/test_crypto.py

"""Tests for AES encryption (Phase 3.1)."""
import pytest
from app.core.crypto import encrypt_password, decrypt_password
pytestmark = pytest.mark.skip(reason="Skip: awaiting Go backend alignment")


class TestEncryption:
    def test_encrypt_decrypt_roundtrip(self):
        """Encrypt then decrypt returns original."""
        plain = "my_secret_password_123"
        encrypted = encrypt_password(plain)
        assert encrypted != plain
        assert decrypt_password(encrypted) == plain

    def test_encrypt_produces_different_ciphertext(self):
        """Each encryption uses a random nonce, so ciphertext differs."""
        plain = "same_password"
        e1 = encrypt_password(plain)
        e2 = encrypt_password(plain)
        assert e1 != e2  # different nonces
        assert decrypt_password(e1) == plain
        assert decrypt_password(e2) == plain

    def test_encrypt_empty_string(self):
        """Can encrypt and decrypt empty string."""
        encrypted = encrypt_password("")
        assert decrypt_password(encrypted) == ""

    def test_encrypt_special_characters(self):
        """Handles special characters in passwords."""
        plain = 'p@ssw0rd!#$%^&*()_+-=[]{}|;\':",./<>?'
        encrypted = encrypt_password(plain)
        assert decrypt_password(encrypted) == plain

    def test_encrypt_unicode(self):
        """Handles unicode characters."""
        plain = "密码测试🔑"
        encrypted = encrypt_password(plain)
        assert decrypt_password(encrypted) == plain

    def test_decrypt_invalid_data_returns_as_is(self):
        """Invalid ciphertext returns original string (backward compat)."""
        result = decrypt_password("not_valid_base64!!!")
        assert result == "not_valid_base64!!!"


class TestInstanceEncryption:
    """Test that instance passwords are stored encrypted via API."""

    @pytest.mark.asyncio
    async def test_create_instance_encrypts_password(self, client, admin_token):
        """POST /instances stores password as encrypted ciphertext."""
        from tests.conftest import auth_header
        resp = await client.post(
            "/api/v1/instances",
            json={
                "name": "加密测试",
                "db_type": "mysql",
                "host": "10.0.0.1",
                "port": 3306,
                "username": "root",
                "password": "plain_text_pw",
                "database": "test",
                "env": "development",
            },
            headers=auth_header(admin_token),
        )
        assert resp.status_code == 201
        data = resp.json()
        # Response should NOT contain password at all (InstanceResponse excludes it)
        assert "password" not in data

    @pytest.mark.asyncio
    async def test_password_not_stored_as_plaintext(self, client, db, admin_token):
        """Verify the database actually stores encrypted, not plain text."""
        from tests.conftest import auth_header
        from app.models.instance import Instance
        from sqlalchemy import select

        plain_pw = "super_secret_123"
        resp = await client.post(
            "/api/v1/instances",
            json={
                "name": "明文检查",
                "db_type": "mysql",
                "host": "10.0.0.2",
                "port": 3306,
                "username": "root",
                "password": plain_pw,
                "database": "test",
                "env": "development",
            },
            headers=auth_header(admin_token),
        )
        assert resp.status_code == 201
        instance_id = resp.json()["id"]

        # Read raw from database
        result = await db.execute(select(Instance).where(Instance.id == instance_id))
        instance = result.scalar_one()
        assert instance.password != plain_pw  # Not stored as plaintext
        assert len(instance.password) > 20  # Encrypted is longer

    @pytest.mark.asyncio
    async def test_update_instance_re_encrypts_password(self, client, db, admin_token):
        """PUT /instances/:id re-encrypts password when updated."""
        from tests.conftest import auth_header
        from app.models.instance import Instance
        from sqlalchemy import select


        # Create
        resp = await client.post(
            "/api/v1/instances",
            json={
                "name": "更新加密",
                "db_type": "mysql",
                "host": "10.0.0.3",
                "port": 3306,
                "username": "root",
                "password": "old_password",
                "database": "test",
            },
            headers=auth_header(admin_token),
        )
        iid = resp.json()["id"]

        # Update password
        resp = await client.put(
            f"/api/v1/instances/{iid}",
            json={"password": "new_password"},
            headers=auth_header(admin_token),
        )
        assert resp.status_code == 200

        # Verify encrypted
        result = await db.execute(select(Instance).where(Instance.id == iid))
        inst = result.scalar_one()
        assert inst.password != "new_password"
        assert inst.password != "old_password"