wecker/tests/test_api.py

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 any("wecker.py" in str(arg) for arg in args[0])
        assert kwargs.get("shell") is not True
        assert kwargs.get("cwd") is not None
        assert kwargs.get("start_new_session") 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"
    )