Windfree poller application & jenkins pipeline

tests/test_oauth.py

@@ -0,0 +1,52 @@
+import json
+import os
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+from windfree_poller.oauth_server import State, save_tokens, token_metadata
+from windfree_poller.oauth_config import OAuthConfig
+
+
+def config(path: Path) -> OAuthConfig:
+    return OAuthConfig(
+        "127.0.0.1", 8088, "https://example.com",
+        "https://sso.example/realms/test", "client", "secret", "admin",
+        "https://example.com/oauth/keycloak/callback", "st-client", "st-secret",
+        "https://example.com/oauth/smartthings/callback", ("r:devices:$",),
+        "x" * 32, path,
+    )
+
+
+class OAuthTests(unittest.TestCase):
+    def test_signed_session_cookie_cannot_be_modified(self):
+        with tempfile.TemporaryDirectory() as folder:
+            state = State(config(Path(folder) / "token.json"))
+            session_id, session = state.new_session()
+            self.assertIs(state.verify_cookie(state.cookie_value(session_id))[1], session)
+            self.assertIsNone(state.verify_cookie(state.cookie_value(session_id) + "changed"))
+
+    def test_tokens_are_saved_atomically_without_exposing_them_in_metadata(self):
+        with tempfile.TemporaryDirectory() as folder:
+            path = Path(folder) / "token.json"
+            save_tokens(path, {
+                "access_token": "access", "refresh_token": "refresh",
+                "expires_in": 60, "installed_app_id": "installed", "scope": "r:devices:$",
+            })
+            saved = json.loads(path.read_text(encoding="utf-8"))
+            self.assertEqual(saved["refresh_token"], "refresh")
+            self.assertGreater(saved["expires_at"], time.time())
+            metadata = token_metadata(path)
+            self.assertTrue(metadata["connected"])
+            self.assertNotIn("access_token", metadata)
+            self.assertNotIn("refresh_token", metadata)
+            if os.name != "nt":
+                self.assertEqual(path.stat().st_mode & 0o777, 0o600)
+
+    def test_missing_token_file_is_disconnected(self):
+        self.assertEqual(token_metadata(Path("missing-token-file")), {"connected": False})
+
+
+if __name__ == "__main__":
+    unittest.main()