diff schwab_trader/schwab_client_test.py @ 220:eb8b4230fdb9

[schwab-trader] Add guarded trading experiment
author MrJuneJune <me@mrjunejune.com>
date Sun, 02 Aug 2026 08:52:13 -0700
parents
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/schwab_trader/schwab_client_test.py	Sun Aug 02 08:52:13 2026 -0700
@@ -0,0 +1,67 @@
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from urllib.parse import parse_qs, urlparse
+
+from schwab_trader.schwab_client import (
+    SchwabConfig,
+    SchwabError,
+    build_authorization_url,
+    build_equity_order,
+    extract_authorization_code,
+    save_tokens,
+)
+
+
+class SchwabClientTest(unittest.TestCase):
+    def test_authorization_url_contains_oauth_params(self):
+        url = build_authorization_url("app-key", "https://127.0.0.1/callback", "state-1")
+        parsed = urlparse(url)
+        params = parse_qs(parsed.query)
+
+        self.assertEqual("https", parsed.scheme)
+        self.assertEqual("api.schwabapi.com", parsed.netloc)
+        self.assertEqual(["code"], params["response_type"])
+        self.assertEqual(["app-key"], params["client_id"])
+        self.assertEqual(["https://127.0.0.1/callback"], params["redirect_uri"])
+        self.assertEqual(["state-1"], params["state"])
+
+    def test_extract_authorization_code_from_callback_url(self):
+        code = extract_authorization_code("https://127.0.0.1/callback?code=abc%40123&state=x")
+        self.assertEqual("abc@123", code)
+
+    def test_build_market_equity_order(self):
+        order = build_equity_order("buy", "aapl", 1)
+
+        self.assertEqual("MARKET", order["orderType"])
+        self.assertEqual("BUY", order["orderLegCollection"][0]["instruction"])
+        self.assertEqual("AAPL", order["orderLegCollection"][0]["instrument"]["symbol"])
+        self.assertNotIn("price", order)
+
+    def test_limit_order_requires_price(self):
+        with self.assertRaises(SchwabError):
+            build_equity_order("SELL", "MSFT", 2, order_type="LIMIT")
+
+    def test_save_tokens_uses_owner_only_permissions(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            token_file = Path(temp_dir) / "tokens.json"
+            save_tokens(token_file, {"access_token": "x"})
+
+            self.assertEqual(0o600, token_file.stat().st_mode & 0o777)
+
+    def test_config_from_env_requires_credentials(self):
+        old_env = os.environ.copy()
+        try:
+            for key in ("SCHWAB_APP_KEY", "SCHWAB_APP_SECRET", "SCHWAB_REDIRECT_URI"):
+                os.environ.pop(key, None)
+            with self.assertRaises(SchwabError):
+                SchwabConfig.from_env()
+        finally:
+            os.environ.clear()
+            os.environ.update(old_env)
+
+
+if __name__ == "__main__":
+    unittest.main()
+