pali-lili/tests/tofu/test_tofu_state.py

from typing import AsyncGenerator

import pytest
from httpx import AsyncClient, ASGITransport

from src.auth.service import get_admin
from src.main import app

PASSWORD="password123"

AUTH=("tofu", PASSWORD)

CREATE_INSTANCE_PAYLOAD = {
    "suppress_deployment": True,
    "password": PASSWORD,
    "configuration": {
        "terraform": {"required_providers": {"random": {"source": "hashicorp/random", "version": ">= 3.0.0"}}},
        "provider": {"random": {}},
        "variable": {
            "password_length": {"description": "Length of the random password", "type": "number", "default": 16}
        },
        "resource": {"random_password": {"example": {"length": "${var.password_length}", "special": True}}},
        "output": {"generated_password": {"value": "${random_password.example.result}", "sensitive": True}},
    },
}

INITIAL_STATE_PAYLOAD = {"key": "value"}

UPDATE_STATE_PAYLOAD = {"key": "value"}

STATE_LOCK_PAYLOAD_1 = {"ID": "bd812a7e-2297-4b70-acc9-d51015a9172c",
                        "Operation": "OperationTypeInvalid",
                        "Info": "",
                        "Who": "",
                        "Version": "",
                        "Created": "1990-01-01T12:00:00Z",
                        "Path": ""
                        }

STATE_LOCK_PAYLOAD_2 = {"ID": "ab0eb55f-2f00-4e02-9bf5-e2c6658ab8af",
                        "Operation": "OperationTypeInvalid",
                        "Info": "",
                        "Who": "",
                        "Version": "",
                        "Created": "1990-01-01T12:00:00Z",
                        "Path": ""
                        }


@pytest.fixture(scope="module")
async def client() -> AsyncGenerator[AsyncClient, None]:
    host, port = "127.0.0.1", "9000"

    async def override_get_admin():
        return None

    app.dependency_overrides[get_admin] = override_get_admin
    async with AsyncClient(transport=ASGITransport(app=app, client=(host, port)), base_url="http://test", auth=AUTH) as client:
        yield client


@pytest.fixture
async def instance_id(client: AsyncClient):
    response = await client.post("/api/v1/tofu/instances", json=CREATE_INSTANCE_PAYLOAD)
    assert response.status_code == 202
    return response.json()["id"]


@pytest.mark.anyio
async def test_state_no_locking(client: AsyncClient, instance_id: int):
    # Initially there should be no state
    response = await client.get(f"/api/v1/tofu/instances/{instance_id}/state")
    assert response.status_code == 404
    # Let's create the state
    response = await client.post(f"/api/v1/tofu/instances/{instance_id}/state", json=INITIAL_STATE_PAYLOAD)
    assert response.status_code == 200
    # Now check the state is retrievable
    response = await client.get(f"/api/v1/tofu/instances/{instance_id}/state")
    assert response.status_code == 200
    assert response.json() == INITIAL_STATE_PAYLOAD
    # Now update the state
    response = await client.post(f"/api/v1/tofu/instances/{instance_id}/state", json=UPDATE_STATE_PAYLOAD)
    assert response.status_code == 200
    # Now check the state is retrievable
    response = await client.get(f"/api/v1/tofu/instances/{instance_id}/state")
    assert response.status_code == 200
    assert response.json() == UPDATE_STATE_PAYLOAD
    # Now purge the state
    response = await client.delete(f"/api/v1/tofu/instances/{instance_id}/state")
    assert response.status_code == 200
    # And check it is gone
    response = await client.get(f"/api/v1/tofu/instances/{instance_id}/state")
    assert response.status_code == 404

@pytest.mark.anyio
async def test_state_double_locking(client: AsyncClient, instance_id: int):
    response = await client.request("LOCK", f"/api/v1/tofu/instances/{instance_id}/state", json=STATE_LOCK_PAYLOAD_1)
    assert response.status_code == 200
    response = await client.request("LOCK", f"/api/v1/tofu/instances/{instance_id}/state", json=STATE_LOCK_PAYLOAD_2)
    assert response.status_code == 423
    assert response.json() == STATE_LOCK_PAYLOAD_1
    response = await client.request("UNLOCK",
                                    f"/api/v1/tofu/instances/{instance_id}/state?ID=" + STATE_LOCK_PAYLOAD_2['ID'])
    assert response.status_code == 423
    assert response.json() == STATE_LOCK_PAYLOAD_1
    response = await client.request("UNLOCK",
                                    f"/api/v1/tofu/instances/{instance_id}/state?ID=" + STATE_LOCK_PAYLOAD_1['ID'])
    assert response.status_code == 200

@pytest.mark.anyio
async def test_state_locked_update(client: AsyncClient, instance_id: int):
    response = await client.request("LOCK", f"/api/v1/tofu/instances/{instance_id}/state", json=STATE_LOCK_PAYLOAD_1)
    assert response.status_code == 200
    response = await client.post(f"/api/v1/tofu/instances/{instance_id}/state?ID=" + STATE_LOCK_PAYLOAD_1['ID'],
                                 json=INITIAL_STATE_PAYLOAD)
    assert response.status_code == 200
    response = await client.post(f"/api/v1/tofu/instances/{instance_id}/state?ID=" + STATE_LOCK_PAYLOAD_2['ID'],
                                 json=UPDATE_STATE_PAYLOAD)
    assert response.status_code == 423
    assert response.json() == STATE_LOCK_PAYLOAD_1
    response = await client.post(f"/api/v1/tofu/instances/{instance_id}/state?ID=" + STATE_LOCK_PAYLOAD_1['ID'],
                                 json=UPDATE_STATE_PAYLOAD)
    assert response.status_code == 200
    response = await client.delete(f"/api/v1/tofu/instances/{instance_id}/state?ID=" + STATE_LOCK_PAYLOAD_2['ID'])
    assert response.status_code == 423
    assert response.json() == STATE_LOCK_PAYLOAD_1
    response = await client.delete(f"/api/v1/tofu/instances/{instance_id}/state?ID=" + STATE_LOCK_PAYLOAD_1['ID'])
    assert response.status_code == 200

@pytest.mark.anyio
async def test_state_allow_force_unlock(client: AsyncClient, instance_id: int):
    # force-unlock doesn't include the ID when calling the unlock endpoint
    response = await client.request("LOCK", f"/api/v1/tofu/instances/{instance_id}/state", json=STATE_LOCK_PAYLOAD_1)
    assert response.status_code == 200
    response = await client.request("UNLOCK", f"/api/v1/tofu/instances/{instance_id}/state", json=STATE_LOCK_PAYLOAD_1)
    assert response.status_code == 200