330 lines
9.8 KiB
Python
330 lines
9.8 KiB
Python
from unittest.mock import patch
|
|
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_is_ringing_returns_false_when_not_running():
|
|
"""Test that isRinging returns False when no wecker process is running."""
|
|
headers = {"X-API-Key": "test-secret"}
|
|
query = """
|
|
query {
|
|
isRinging
|
|
}
|
|
"""
|
|
res = client.post("/graphql", json={"query": query}, headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json()["data"]["isRinging"] is False
|
|
|
|
|
|
@patch("api.schema.is_wecker_ringing", return_value=True)
|
|
def test_is_ringing_returns_true_when_running(mock_is_ringing):
|
|
"""Test that isRinging returns True when a wecker process is running."""
|
|
headers = {"X-API-Key": "test-secret"}
|
|
query = """
|
|
query {
|
|
isRinging
|
|
}
|
|
"""
|
|
res = client.post("/graphql", json={"query": query}, headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json()["data"]["isRinging"] is True
|
|
|
|
|
|
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_bugfix_default_command_uses_append_for_logs():
|
|
"""
|
|
Test for bugfix: Ensure the default command appends (>>) to wecker.log
|
|
instead of overwriting (>) it.
|
|
"""
|
|
from api.schema import Mutation
|
|
|
|
mutation = Mutation()
|
|
# Call the resolver directly without command to trigger default command generation
|
|
alarm = mutation.set_alarm(cron_expression="0 9 * * *")
|
|
|
|
# Verify the generated command string
|
|
command = alarm.command
|
|
assert ">> wecker.log 2>&1" in command, (
|
|
f"Command must use append '>>' syntax. Got: {command}"
|
|
)
|
|
assert "> wecker.log 2>&1" not in command.replace(">> wecker.log", "REPLACED"), (
|
|
"Command must not use overwrite '>'"
|
|
)
|
|
|
|
# Cleanup
|
|
mutation.delete_alarm(id=alarm.id)
|
|
|
|
|
|
def test_set_alarm_default_command_append():
|
|
headers = {"X-API-Key": "test-secret"}
|
|
mutation = """
|
|
mutation {
|
|
setAlarm(cronExpression: "0 9 * * *") {
|
|
id
|
|
command
|
|
}
|
|
}
|
|
"""
|
|
res = client.post("/graphql", json={"query": mutation}, headers=headers)
|
|
assert res.status_code == 200
|
|
data = res.json()["data"]["setAlarm"]
|
|
|
|
# Assert the command contains the correct append syntax (>>) and not just overwrite (>)
|
|
command = data["command"]
|
|
assert ">> wecker.log 2>&1" in command
|
|
assert "> wecker.log 2>&1" not in command.replace(">> wecker.log", "REPLACED")
|
|
|
|
# Cleanup so we don't break subsequent tests
|
|
alarm_id = data["id"]
|
|
mutation_delete = f"""
|
|
mutation {{
|
|
deleteAlarm(id: "{alarm_id}")
|
|
}}
|
|
"""
|
|
client.post("/graphql", json={"query": mutation_delete}, headers=headers)
|
|
|
|
|
|
def test_set_alarm_rejects_custom_command():
|
|
"""Custom command argument must be rejected to prevent crontab injection."""
|
|
headers = {"X-API-Key": "test-secret"}
|
|
mutation = """
|
|
mutation {
|
|
setAlarm(cronExpression: "0 9 * * *", command: "rm -rf /") {
|
|
id
|
|
}
|
|
}
|
|
"""
|
|
res = client.post("/graphql", json={"query": mutation}, headers=headers)
|
|
assert res.status_code == 200
|
|
assert "errors" in res.json()
|
|
assert "command" in str(res.json()["errors"])
|
|
|
|
|
|
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 * * *", 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 "wecker.py" in alarm["command"]
|
|
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 * * *", 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"] == []
|
|
|
|
|
|
def test_start_ringing_starts_process_when_not_ringing():
|
|
"""startRinging returns True and spawns wecker.py when not already ringing."""
|
|
from api.schema import Mutation
|
|
|
|
mutation = Mutation()
|
|
|
|
with patch("api.schema.is_wecker_ringing", return_value=False), \
|
|
patch("subprocess.Popen") as mock_popen:
|
|
mock_proc = mock_popen.return_value
|
|
mock_proc.pid = 9999
|
|
result = mutation.start_ringing()
|
|
assert result is True
|
|
mock_popen.assert_called_once()
|
|
args, kwargs = mock_popen.call_args
|
|
assert "wecker.py" in args[0]
|
|
assert kwargs.get("shell") is True
|
|
|
|
|
|
def test_start_ringing_ignores_when_already_ringing():
|
|
"""startRinging returns False when already ringing."""
|
|
from api.schema import Mutation
|
|
|
|
mutation = Mutation()
|
|
|
|
with patch("api.schema.is_wecker_ringing", return_value=True), \
|
|
patch("subprocess.Popen") as mock_popen:
|
|
result = mutation.start_ringing()
|
|
assert result is False
|
|
mock_popen.assert_not_called()
|
|
|
|
|
|
def test_stop_ringing_kills_process_when_ringing():
|
|
"""stopRinging returns True and kills the process when ringing."""
|
|
from api.schema import Mutation
|
|
|
|
mutation = Mutation()
|
|
|
|
with patch("api.schema.is_wecker_ringing", return_value=True), \
|
|
patch("api.schema.os.kill") as mock_kill, \
|
|
patch("common.PID_FILE", "/tmp/fake_wecker.pid"), \
|
|
patch("builtins.open") as mock_open:
|
|
mock_f = mock_open.return_value.__enter__.return_value
|
|
mock_f.read.return_value = "1234"
|
|
mock_kill.return_value = None
|
|
result = mutation.stop_ringing()
|
|
assert result is True
|
|
mock_kill.assert_called_with(1234, 15)
|
|
|
|
|
|
def test_stop_ringing_does_nothing_when_not_ringing():
|
|
"""stopRinging returns False when not ringing."""
|
|
from api.schema import Mutation
|
|
|
|
mutation = Mutation()
|
|
|
|
with patch("api.schema.is_wecker_ringing", return_value=False), \
|
|
patch("api.schema.os.kill") as mock_kill:
|
|
result = mutation.stop_ringing()
|
|
assert result is False
|
|
mock_kill.assert_not_called()
|
|
|
|
|
|
def test_start_ringing_graphql_endpoint():
|
|
"""GraphQL mutation startRinging works via the API."""
|
|
headers = {"X-API-Key": "test-secret"}
|
|
mutation_str = """
|
|
mutation {
|
|
startRinging
|
|
}
|
|
"""
|
|
|
|
with patch("api.schema.is_wecker_ringing", return_value=False), \
|
|
patch("subprocess.Popen") as mock_popen:
|
|
mock_proc = mock_popen.return_value
|
|
mock_proc.pid = 9999
|
|
res = client.post("/graphql", json={"query": mutation_str}, headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json()["data"]["startRinging"] is True
|
|
|
|
|
|
def test_stop_ringing_graphql_endpoint():
|
|
"""GraphQL mutation stopRinging works via the API."""
|
|
headers = {"X-API-Key": "test-secret"}
|
|
mutation_str = """
|
|
mutation {
|
|
stopRinging
|
|
}
|
|
"""
|
|
|
|
with patch("api.schema.is_wecker_ringing", return_value=True), \
|
|
patch("api.schema.os.kill") as mock_kill, \
|
|
patch("builtins.open") as mock_open:
|
|
mock_f = mock_open.return_value.__enter__.return_value
|
|
mock_f.read.return_value = "1234"
|
|
mock_kill.return_value = None
|
|
res = client.post("/graphql", json={"query": mutation_str}, headers=headers)
|
|
assert res.status_code == 200
|
|
assert res.json()["data"]["stopRinging"] is True
|
|
|
|
|
|
def test_pid_file_same_shared_constant_in_api():
|
|
"""Ensure api.schema uses the same PID_FILE from common, not a redefinition."""
|
|
from api import schema as api_schema
|
|
import common
|
|
|
|
assert api_schema.PID_FILE is common.PID_FILE, (
|
|
"api.schema.PID_FILE must reference common.PID_FILE, not redefine it"
|
|
)
|