diff --git a/main.py b/main.py index 0c24471..2ea5ba7 100644 --- a/main.py +++ b/main.py @@ -1,89 +1,94 @@ import os -from unittest.mock import MagicMock, patch +import random +import time -import pytest -from fastapi.testclient import TestClient +import psutil +from fastapi import FastAPI +from fastapi.responses import RedirectResponse +from pydantic import BaseModel -from main import SystemMetrics, app +app = FastAPI() -client = TestClient(app) +# Modelo de dados -def test_metrics_endpoint_status_code(): - """Testa se o endpoint /metrics retorna status code 200""" - os.environ.pop("SIMULATE_HEAVY_LOAD", None) - os.environ.pop("SIMULATE_UNHEALTHY_BEHAVIOR", None) - response = client.get("/metrics") - assert response.status_code == 200 +class SystemMetrics(BaseModel): + memoria: float # Uso de memória em % + cpu: float # Uso de CPU em % + disk: float # Uso de disco em % -def test_metrics_endpoint_response_model(): - """Testa se a resposta do endpoint /metrics está de acordo com o SystemMetrics model""" - os.environ.pop("SIMULATE_HEAVY_LOAD", None) - os.environ.pop("SIMULATE_UNHEALTHY_BEHAVIOR", None) +# Endpoint principal - response = client.get("/metrics") - assert response.status_code == 200 +@app.get("/metrics", response_model=SystemMetrics, tags=["main"]) +def get_system_metrics(): - data = response.json() + simulate_heavy_load_randomly() - assert "memoria" in data - assert "cpu" in data - assert "disk" in data + memory = psutil.virtual_memory().percent + cpu = psutil.cpu_percent(interval=1) + disk = psutil.disk_usage("/").percent + return SystemMetrics(memoria=memory, cpu=cpu, disk=disk) - assert isinstance(data["memoria"], (int, float)) - assert isinstance(data["cpu"], (int, float)) - assert isinstance(data["disk"], (int, float)) - assert 0 <= data["memoria"] <= 100 - assert 0 <= data["cpu"] <= 100 - assert 0 <= data["disk"] <= 100 +# Endpoints para o Kubernetes - try: - metrics = SystemMetrics(**data) - assert metrics.memoria == data["memoria"] - assert metrics.cpu == data["cpu"] - assert metrics.disk == data["disk"] - except Exception as e: - pytest.fail(f"Resposta não está de acordo com o modelo SystemMetrics: {e}") +@app.get("/health", tags=["k8s"]) +@app.get("/healthz", tags=["k8s"]) +@app.get("/ready", tags=["k8s"]) +@app.get("/readyz", tags=["k8s"]) +def health_check(): + """Para de enviar trafego se estiver em um estado inadequado.""" + simulate_unhealthy_behavior_randomly() + return {"status": "ok"} -@patch("psutil.virtual_memory") -@patch("psutil.cpu_percent") -@patch("psutil.disk_usage") -def test_metrics_endpoint_with_mocked_values(mock_disk, mock_cpu, mock_memory): - """Testa o endpoint /metrics com valores mockados para garantir consistência""" - os.environ.pop("SIMULATE_HEAVY_LOAD", None) - os.environ.pop("SIMULATE_UNHEALTHY_BEHAVIOR", None) - mock_memory_obj = MagicMock() - mock_memory_obj.percent = 45.5 - mock_memory.return_value = mock_memory_obj +@app.get("/liveness", tags=["k8s"]) +def liveness_check(): + """Reinicia o Pod se estiver em um estado inadequado.""" + simulate_delay_randomly() + return {"status": "ok"} - mock_cpu.return_value = 23.8 - mock_disk_obj = MagicMock() - mock_disk_obj.percent = 67.2 - mock_disk.return_value = mock_disk_obj +@app.get("/startup", tags=["k8s"]) +def startup_check(): + """Verifica se o Pod está pronto para receber trafego.""" + simulate_delay_randomly() + return {"status": "ok"} - response = client.get("/metrics") - assert response.status_code == 200 +# Redirect `/` para `/docs` +@app.get("/", tags=["util"]) +def redirect_to_docs(): + return RedirectResponse(url="/docs") - data = response.json() - assert data["memoria"] == 45.5 - assert data["cpu"] == 23.8 - assert data["disk"] == 67.2 +# Funções de simulação -def test_metrics_endpoint_content_type(): - """Testa se o endpoint /metrics retorna o content-type correto""" - os.environ.pop("SIMULATE_HEAVY_LOAD", None) - os.environ.pop("SIMULATE_UNHEALTHY_BEHAVIOR", None) - response = client.get("/metrics") +def simulate_heavy_load_randomly(): + """Simula um uso pesado aleatoriamente.""" + if not os.getenv("SIMULATE_HEAVY_LOAD"): + return + if random.random() < 0.5: + time.sleep(random.uniform(0.1, 10)) + else: + simulate_unhealthy_behavior_randomly() - assert response.status_code == 200 - assert "application/json" in response.headers["content-type"] + +def simulate_unhealthy_behavior_randomly(): + """Simula um comportamento inadequado aleatoriamente.""" + if not os.getenv("SIMULATE_UNHEALTHY_BEHAVIOR"): + return + if random.random() < 0.5: + raise Exception("Simulação de comportamento inadequado") + + +def simulate_delay_randomly(): + """Simula um atraso aleatório.""" + if not os.getenv("SIMULATE_DELAY"): + return + time.sleep(random.uniform(0.1, 10))