wecker/tests/test_api.py

from fastapi.testclient import TestClient
import tempfile
import os

# We need to set the environment variable before importing the app
os.environ["API_KEY"] = "test-secret"
dummy_tab = tempfile.mktemp()
with open(dummy_tab, "w") as f:
    f.write("")
os.environ["TABFILE"] = dummy_tab

from api.main import app  # noqa: E402

client = TestClient(app)

def test_auth_missing():
    query = """
    query {
      getAlarms {
        id
      }
    }
    """
    response = client.post("/graphql", json={"query": query})
    assert response.status_code == 401

def test_auth_invalid():
    query = """
    query {
      getAlarms {
        id
      }
    }
    """
    response = client.post("/graphql", json={"query": query}, headers={"X-API-Key": "wrong"})
    assert response.status_code == 401

def test_graphql_workflow():
    headers = {"X-API-Key": "test-secret"}
    
    # 1. Get empty alarms
    query_get = """
    query {
      getAlarms {
        id
      }
    }
    """
    res = client.post("/graphql", json={"query": query_get}, headers=headers)
    assert res.status_code == 200
    assert res.json()["data"]["getAlarms"] == []
    
    # 2. Set alarm
    mutation_set = """
    mutation {
      setAlarm(cronExpression: "30 7 * * *", command: "python wecker.py", isEnabled: true) {
        id
        cronExpression
        command
        isEnabled
      }
    }
    """
    res = client.post("/graphql", json={"query": mutation_set}, headers=headers)
    assert res.status_code == 200
    alarm = res.json()["data"]["setAlarm"]
    assert alarm["cronExpression"] == "30 7 * * *"
    assert alarm["command"] == "python wecker.py"
    assert alarm["isEnabled"] is True
    alarm_id = alarm["id"]
    
    # 3. Get alarms lists it
    res = client.post("/graphql", json={"query": query_get}, headers=headers)
    assert len(res.json()["data"]["getAlarms"]) == 1
    assert res.json()["data"]["getAlarms"][0]["id"] == alarm_id
    
    # 4. Get specific alarm
    query_one = f"""
    query {{
      getAlarm(id: "{alarm_id}") {{
        id
        cronExpression
      }}
    }}
    """
    res = client.post("/graphql", json={"query": query_one}, headers=headers)
    assert res.json()["data"]["getAlarm"]["id"] == alarm_id
    assert res.json()["data"]["getAlarm"]["cronExpression"] == "30 7 * * *"
    
    # 5. Update alarm
    mutation_update = f"""
    mutation {{
      setAlarm(id: "{alarm_id}", cronExpression: "0 8 * * *", command: "python wecker.py", isEnabled: false) {{
        id
        cronExpression
        isEnabled
      }}
    }}
    """
    res = client.post("/graphql", json={"query": mutation_update}, headers=headers)
    alarm_updated = res.json()["data"]["setAlarm"]
    assert alarm_updated["id"] == alarm_id
    assert alarm_updated["cronExpression"] == "0 8 * * *"
    assert alarm_updated["isEnabled"] is False
    
    # 6. Delete alarm
    mutation_delete = f"""
    mutation {{
      deleteAlarm(id: "{alarm_id}")
    }}
    """
    res = client.post("/graphql", json={"query": mutation_delete}, headers=headers)
    assert res.json()["data"]["deleteAlarm"] is True
    
    # 7. List again is empty
    res = client.post("/graphql", json={"query": query_get}, headers=headers)
    assert res.json()["data"]["getAlarms"] == []