diff --git a/app/mywhoosh/__init__.py b/app/mywhoosh/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/mywhoosh/models.py b/app/mywhoosh/models.py new file mode 100644 index 0000000..54a2623 --- /dev/null +++ b/app/mywhoosh/models.py @@ -0,0 +1,17 @@ +from dataclasses import dataclass +from datetime import datetime + + +@dataclass(frozen=True) +class MyWhooshToken: + access_token: str + refresh_token: str | None + whoosh_id: str | None + + +@dataclass(frozen=True) +class MyWhooshActivity: + id: str + title: str + activity_file_id: str + started_at: datetime | None diff --git a/app/mywhoosh/tokenstore.py b/app/mywhoosh/tokenstore.py new file mode 100644 index 0000000..5ed3dda --- /dev/null +++ b/app/mywhoosh/tokenstore.py @@ -0,0 +1,42 @@ +import json +import os +from pathlib import Path + +from app.mywhoosh.models import MyWhooshToken + + +class MyWhooshTokenStore: + def __init__(self, path: Path) -> None: + self.path = path + + def load(self) -> MyWhooshToken | None: + try: + raw = json.loads(self.path.read_text("utf-8")) + except FileNotFoundError: + return None + return MyWhooshToken( + access_token=raw["access_token"], + refresh_token=raw.get("refresh_token"), + whoosh_id=raw.get("whoosh_id"), + ) + + def save(self, token: MyWhooshToken) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + tmp = self.path.with_suffix(".tmp") + tmp.write_text( + json.dumps( + { + "access_token": token.access_token, + "refresh_token": token.refresh_token, + "whoosh_id": token.whoosh_id, + }, + indent=2, + ), + "utf-8", + ) + os.chmod(tmp, 0o600) + tmp.replace(self.path) + os.chmod(self.path, 0o600) + + def clear(self) -> None: + self.path.unlink(missing_ok=True) diff --git a/tests/mywhoosh/__init__.py b/tests/mywhoosh/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/mywhoosh/test_tokenstore.py b/tests/mywhoosh/test_tokenstore.py new file mode 100644 index 0000000..2a322aa --- /dev/null +++ b/tests/mywhoosh/test_tokenstore.py @@ -0,0 +1,18 @@ +from pathlib import Path + +from app.mywhoosh.models import MyWhooshToken +from app.mywhoosh.tokenstore import MyWhooshTokenStore + + +def test_tokenstore_round_trip_and_permissions(tmp_path: Path) -> None: + store = MyWhooshTokenStore(tmp_path / "tokens" / "7" / "mywhoosh.json") + token = MyWhooshToken(access_token="access", refresh_token="refresh", whoosh_id="whoosh-7") + store.save(token) + + assert store.load() == token + assert oct(store.path.stat().st_mode & 0o777) == "0o600" + + +def test_missing_token_returns_none(tmp_path: Path) -> None: + store = MyWhooshTokenStore(tmp_path / "missing.json") + assert store.load() is None