383 lines
14 KiB
Python
383 lines
14 KiB
Python
import json
|
|
import sqlite3
|
|
from urllib import error
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.auth_db import reset_auth_db_caches
|
|
from app.config import Settings, get_settings
|
|
from app.integrations.ticktick import (
|
|
AUTH_SCOPE,
|
|
TICKTICK_AUTH_URL,
|
|
TickTickClient,
|
|
TickTickTask,
|
|
default_auth_state_store,
|
|
)
|
|
from app.main import create_app
|
|
|
|
|
|
class _FakeJsonResponse:
|
|
def __init__(self, status_code: int, payload):
|
|
self.status_code = status_code
|
|
self.payload = payload
|
|
|
|
def getcode(self) -> int:
|
|
return self.status_code
|
|
|
|
def read(self) -> bytes:
|
|
return json.dumps(self.payload).encode("utf-8")
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
|
|
def _configured_settings(**overrides) -> Settings:
|
|
payload = {
|
|
"ticktick_client_id": "ticktick-client-id",
|
|
"ticktick_client_secret": "ticktick-client-secret",
|
|
"ticktick_redirect_uri": "http://localhost:8000/ticktick/auth/code",
|
|
"ticktick_token": "ticktick-access-token",
|
|
"home_assistant_action_task_project_id": "project-123",
|
|
}
|
|
payload.update(overrides)
|
|
return Settings(_env_file=None, **payload)
|
|
|
|
|
|
def _extract_csrf_token(html: str) -> str:
|
|
import re
|
|
|
|
match = re.search(r'name="csrf_token" value="([^"]+)"', html)
|
|
assert match is not None
|
|
return match.group(1)
|
|
|
|
|
|
def test_build_authorization_url_contains_expected_query(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client = TickTickClient(settings=_configured_settings())
|
|
monkeypatch.setattr("app.integrations.ticktick.secrets.token_hex", lambda _: "state-123")
|
|
|
|
authorization_url = client.build_authorization_url()
|
|
parsed = urlparse(authorization_url)
|
|
query = parse_qs(parsed.query)
|
|
|
|
assert f"{parsed.scheme}://{parsed.netloc}{parsed.path}" == TICKTICK_AUTH_URL
|
|
assert query["client_id"] == ["ticktick-client-id"]
|
|
assert query["response_type"] == ["code"]
|
|
assert query["redirect_uri"] == ["http://localhost:8000/ticktick/auth/code"]
|
|
assert query["state"] == ["state-123"]
|
|
assert query["scope"] == [AUTH_SCOPE]
|
|
|
|
|
|
def test_exchange_authorization_code_posts_expected_request(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
captured = {}
|
|
client = TickTickClient(settings=_configured_settings())
|
|
default_auth_state_store.pending_state = "expected-state"
|
|
|
|
def fake_urlopen(req, timeout):
|
|
captured["url"] = req.full_url
|
|
captured["timeout"] = timeout
|
|
captured["authorization"] = req.headers["Authorization"]
|
|
captured["content_type"] = req.headers["Content-type"]
|
|
captured["body"] = req.data.decode("utf-8")
|
|
return _FakeJsonResponse(200, {"access_token": "new-token"})
|
|
|
|
monkeypatch.setattr("app.integrations.ticktick.request.urlopen", fake_urlopen)
|
|
|
|
token = client.exchange_authorization_code(code="oauth-code", state="expected-state")
|
|
|
|
assert token == "new-token"
|
|
assert captured["url"] == "https://ticktick.com/oauth/token"
|
|
assert captured["timeout"] == pytest.approx(10.0)
|
|
assert captured["content_type"] == "application/x-www-form-urlencoded"
|
|
assert captured["authorization"].startswith("Basic ")
|
|
assert "code=oauth-code" in captured["body"]
|
|
assert "grant_type=authorization_code" in captured["body"]
|
|
assert "scope=tasks%3Aread+tasks%3Awrite" in captured["body"]
|
|
assert "client_id=" not in captured["body"]
|
|
assert "client_secret=" not in captured["body"]
|
|
|
|
|
|
def test_exchange_authorization_code_trims_ticktick_config_values(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
captured = {}
|
|
client = TickTickClient(
|
|
settings=_configured_settings(
|
|
ticktick_client_id=" ticktick-client-id ",
|
|
ticktick_client_secret=" ticktick-client-secret ",
|
|
ticktick_redirect_uri=" http://localhost:8000/ticktick/auth/code ",
|
|
)
|
|
)
|
|
default_auth_state_store.pending_state = "trimmed-state"
|
|
|
|
def fake_urlopen(req, timeout):
|
|
captured["authorization"] = req.headers["Authorization"]
|
|
captured["body"] = req.data.decode("utf-8")
|
|
return _FakeJsonResponse(200, {"access_token": "trimmed-token"})
|
|
|
|
monkeypatch.setattr("app.integrations.ticktick.request.urlopen", fake_urlopen)
|
|
|
|
token = client.exchange_authorization_code(code="oauth-code", state="trimmed-state")
|
|
|
|
assert token == "trimmed-token"
|
|
assert captured["authorization"].startswith("Basic ")
|
|
assert "redirect_uri=http%3A%2F%2Flocalhost%3A8000%2Fticktick%2Fauth%2Fcode" in captured["body"]
|
|
|
|
|
|
def test_create_task_skips_duplicate_titles(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client = TickTickClient(settings=_configured_settings())
|
|
|
|
def fake_urlopen(req, timeout):
|
|
assert req.full_url.endswith("/project/project-123/data")
|
|
return _FakeJsonResponse(
|
|
200,
|
|
{
|
|
"tasks": [
|
|
{
|
|
"id": "task-1",
|
|
"projectId": "project-123",
|
|
"title": "wash dishes",
|
|
"columnId": "column-7",
|
|
}
|
|
]
|
|
},
|
|
)
|
|
|
|
monkeypatch.setattr("app.integrations.ticktick.request.urlopen", fake_urlopen)
|
|
|
|
client.create_task(TickTickTask(projectId="project-123", title="wash dishes"))
|
|
|
|
|
|
def test_get_projects_ignores_unknown_fields(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client = TickTickClient(settings=_configured_settings())
|
|
|
|
def fake_urlopen(req, timeout):
|
|
assert req.full_url.endswith("/project/")
|
|
return _FakeJsonResponse(
|
|
200,
|
|
[
|
|
{
|
|
"id": "project-123",
|
|
"name": "Inbox",
|
|
"etag": "project-etag",
|
|
}
|
|
],
|
|
)
|
|
|
|
monkeypatch.setattr("app.integrations.ticktick.request.urlopen", fake_urlopen)
|
|
|
|
projects = client.get_projects()
|
|
|
|
assert len(projects) == 1
|
|
assert projects[0].id == "project-123"
|
|
assert projects[0].name == "Inbox"
|
|
|
|
|
|
def test_create_task_posts_expected_payload(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
captured = {"calls": []}
|
|
client = TickTickClient(settings=_configured_settings())
|
|
|
|
def fake_urlopen(req, timeout):
|
|
captured["calls"].append(req.full_url)
|
|
if req.full_url.endswith("/project/project-123/data"):
|
|
return _FakeJsonResponse(200, {"tasks": []})
|
|
captured["authorization"] = req.headers["Authorization"]
|
|
captured["content_type"] = req.headers["Content-type"]
|
|
captured["body"] = json.loads(req.data.decode("utf-8"))
|
|
return _FakeJsonResponse(200, {"id": "task-99"})
|
|
|
|
monkeypatch.setattr("app.integrations.ticktick.request.urlopen", fake_urlopen)
|
|
|
|
client.create_task(
|
|
TickTickTask(projectId="project-123", title="wash dishes", dueDate="2026-04-21T00:00:00+0000")
|
|
)
|
|
|
|
assert captured["calls"] == [
|
|
"https://api.ticktick.com/open/v1/project/project-123/data",
|
|
"https://api.ticktick.com/open/v1/task",
|
|
]
|
|
assert captured["authorization"] == "Bearer ticktick-access-token"
|
|
assert captured["content_type"] == "application/json"
|
|
assert captured["body"] == {
|
|
"projectId": "project-123",
|
|
"title": "wash dishes",
|
|
"dueDate": "2026-04-21T00:00:00+0000",
|
|
}
|
|
|
|
|
|
def test_homeassistant_publish_creates_ticktick_action_task(
|
|
test_database_urls,
|
|
ready_location_database,
|
|
ready_poo_database,
|
|
auth_database,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setenv("TICKTICK_CLIENT_ID", "ticktick-client-id")
|
|
monkeypatch.setenv("TICKTICK_CLIENT_SECRET", "ticktick-client-secret")
|
|
monkeypatch.setenv("TICKTICK_REDIRECT_URI", "http://localhost:8000/ticktick/auth/code")
|
|
monkeypatch.setenv("TICKTICK_TOKEN", "ticktick-access-token")
|
|
monkeypatch.setenv("HOME_ASSISTANT_ACTION_TASK_PROJECT_ID", "project-123")
|
|
get_settings.cache_clear()
|
|
reset_auth_db_caches()
|
|
|
|
captured = {"calls": []}
|
|
|
|
def fake_urlopen(req, timeout):
|
|
captured["calls"].append(req.full_url)
|
|
if req.full_url.endswith("/project/project-123/data"):
|
|
return _FakeJsonResponse(200, {"tasks": []})
|
|
captured["body"] = json.loads(req.data.decode("utf-8"))
|
|
return _FakeJsonResponse(200, {"id": "task-1"})
|
|
|
|
monkeypatch.setattr("app.integrations.ticktick.request.urlopen", fake_urlopen)
|
|
|
|
with TestClient(create_app()) as client:
|
|
response = client.post(
|
|
"/homeassistant/publish",
|
|
json={
|
|
"target": "ticktick",
|
|
"action": "create_action_task",
|
|
"content": "{'title': 'ignored', 'action': 'take out trash', 'due_hour': 6}",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert captured["calls"] == [
|
|
"https://api.ticktick.com/open/v1/project/project-123/data",
|
|
"https://api.ticktick.com/open/v1/task",
|
|
]
|
|
assert captured["body"]["projectId"] == "project-123"
|
|
assert captured["body"]["title"] == "take out trash"
|
|
assert captured["body"]["dueDate"].endswith("+0000")
|
|
|
|
|
|
def test_ticktick_auth_start_redirects_authenticated_user(
|
|
test_database_urls,
|
|
ready_location_database,
|
|
ready_poo_database,
|
|
auth_database,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setenv("TICKTICK_CLIENT_ID", "ticktick-client-id")
|
|
monkeypatch.setenv("TICKTICK_CLIENT_SECRET", "ticktick-client-secret")
|
|
monkeypatch.setenv("TICKTICK_REDIRECT_URI", "http://localhost:8000/ticktick/auth/code")
|
|
get_settings.cache_clear()
|
|
reset_auth_db_caches()
|
|
monkeypatch.setattr("app.integrations.ticktick.secrets.token_hex", lambda _: "state-redirect")
|
|
|
|
with TestClient(create_app()) as client:
|
|
login_page = client.get("/login")
|
|
csrf_token = _extract_csrf_token(login_page.text)
|
|
client.post(
|
|
"/login",
|
|
data={
|
|
"username": "admin",
|
|
"password": "test-password",
|
|
"csrf_token": csrf_token,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
response = client.get("/ticktick/auth/start", follow_redirects=False)
|
|
|
|
assert response.status_code == 303
|
|
parsed = urlparse(response.headers["location"])
|
|
query = parse_qs(parsed.query)
|
|
assert f"{parsed.scheme}://{parsed.netloc}{parsed.path}" == TICKTICK_AUTH_URL
|
|
assert query["state"] == ["state-redirect"]
|
|
|
|
|
|
def test_ticktick_auth_callback_persists_token(
|
|
test_database_urls,
|
|
ready_location_database,
|
|
ready_poo_database,
|
|
auth_database,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setenv("TICKTICK_CLIENT_ID", "ticktick-client-id")
|
|
monkeypatch.setenv("TICKTICK_CLIENT_SECRET", "ticktick-client-secret")
|
|
monkeypatch.setenv("TICKTICK_REDIRECT_URI", "http://localhost:8000/ticktick/auth/code")
|
|
get_settings.cache_clear()
|
|
reset_auth_db_caches()
|
|
default_auth_state_store.pending_state = "callback-state"
|
|
|
|
def fake_urlopen(req, timeout):
|
|
return _FakeJsonResponse(200, {"access_token": "persisted-token"})
|
|
|
|
monkeypatch.setattr("app.integrations.ticktick.request.urlopen", fake_urlopen)
|
|
|
|
with TestClient(create_app()) as client:
|
|
response = client.get(
|
|
"/ticktick/auth/code?state=callback-state&code=oauth-code",
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/config?ticktick_oauth=success"
|
|
|
|
conn = sqlite3.connect(test_database_urls["app_path"])
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT value FROM app_config WHERE key = ?",
|
|
("TICKTICK_TOKEN",),
|
|
).fetchone()
|
|
finally:
|
|
conn.close()
|
|
|
|
assert row is not None
|
|
assert row[0] == "persisted-token"
|
|
|
|
|
|
def test_ticktick_auth_callback_redirects_on_invalid_state(
|
|
test_database_urls,
|
|
ready_location_database,
|
|
ready_poo_database,
|
|
auth_database,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setenv("TICKTICK_CLIENT_ID", "ticktick-client-id")
|
|
monkeypatch.setenv("TICKTICK_CLIENT_SECRET", "ticktick-client-secret")
|
|
monkeypatch.setenv("TICKTICK_REDIRECT_URI", "http://localhost:8000/ticktick/auth/code")
|
|
get_settings.cache_clear()
|
|
reset_auth_db_caches()
|
|
default_auth_state_store.pending_state = "expected-state"
|
|
|
|
with TestClient(create_app()) as client:
|
|
response = client.get(
|
|
"/ticktick/auth/code?state=wrong-state&code=oauth-code",
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/config?ticktick_oauth=invalid-state"
|
|
|
|
|
|
def test_ticktick_auth_callback_redirects_when_token_exchange_fails(
|
|
test_database_urls,
|
|
ready_location_database,
|
|
ready_poo_database,
|
|
auth_database,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setenv("TICKTICK_CLIENT_ID", "ticktick-client-id")
|
|
monkeypatch.setenv("TICKTICK_CLIENT_SECRET", "ticktick-client-secret")
|
|
monkeypatch.setenv("TICKTICK_REDIRECT_URI", "http://localhost:8000/ticktick/auth/code")
|
|
get_settings.cache_clear()
|
|
reset_auth_db_caches()
|
|
default_auth_state_store.pending_state = "callback-state"
|
|
|
|
def fake_urlopen(req, timeout):
|
|
raise error.HTTPError(req.full_url, 401, "Unauthorized", hdrs=None, fp=None)
|
|
|
|
monkeypatch.setattr("app.integrations.ticktick.request.urlopen", fake_urlopen)
|
|
|
|
with TestClient(create_app()) as client:
|
|
response = client.get(
|
|
"/ticktick/auth/code?state=callback-state&code=oauth-code",
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/config?ticktick_oauth=failed" |