118 lines
3.2 KiB
Python
118 lines
3.2 KiB
Python
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"] == []
|