feat: add MyWhoosh token persistence
This commit is contained in:
0
app/mywhoosh/__init__.py
Normal file
0
app/mywhoosh/__init__.py
Normal file
17
app/mywhoosh/models.py
Normal file
17
app/mywhoosh/models.py
Normal file
@@ -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
|
||||||
42
app/mywhoosh/tokenstore.py
Normal file
42
app/mywhoosh/tokenstore.py
Normal file
@@ -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)
|
||||||
0
tests/mywhoosh/__init__.py
Normal file
0
tests/mywhoosh/__init__.py
Normal file
18
tests/mywhoosh/test_tokenstore.py
Normal file
18
tests/mywhoosh/test_tokenstore.py
Normal file
@@ -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
|
||||||
Reference in New Issue
Block a user