Mercurial

comparison schwab_trader/schwab_client_test.py @ 220:eb8b4230fdb9

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
[schwab-trader] Add guarded trading experiment
author MrJuneJune <me@mrjunejune.com>
date Sun, 02 Aug 2026 08:52:13 -0700
parents
children
comparison
equal deleted inserted replaced
214:4c725fde6999 220:eb8b4230fdb9
1 import os
2 import tempfile
3 import unittest
4 from pathlib import Path
5 from urllib.parse import parse_qs, urlparse
6
7 from schwab_trader.schwab_client import (
8 SchwabConfig,
9 SchwabError,
10 build_authorization_url,
11 build_equity_order,
12 extract_authorization_code,
13 save_tokens,
14 )
15
16
17 class SchwabClientTest(unittest.TestCase):
18 def test_authorization_url_contains_oauth_params(self):
19 url = build_authorization_url("app-key", "https://127.0.0.1/callback", "state-1")
20 parsed = urlparse(url)
21 params = parse_qs(parsed.query)
22
23 self.assertEqual("https", parsed.scheme)
24 self.assertEqual("api.schwabapi.com", parsed.netloc)
25 self.assertEqual(["code"], params["response_type"])
26 self.assertEqual(["app-key"], params["client_id"])
27 self.assertEqual(["https://127.0.0.1/callback"], params["redirect_uri"])
28 self.assertEqual(["state-1"], params["state"])
29
30 def test_extract_authorization_code_from_callback_url(self):
31 code = extract_authorization_code("https://127.0.0.1/callback?code=abc%40123&state=x")
32 self.assertEqual("abc@123", code)
33
34 def test_build_market_equity_order(self):
35 order = build_equity_order("buy", "aapl", 1)
36
37 self.assertEqual("MARKET", order["orderType"])
38 self.assertEqual("BUY", order["orderLegCollection"][0]["instruction"])
39 self.assertEqual("AAPL", order["orderLegCollection"][0]["instrument"]["symbol"])
40 self.assertNotIn("price", order)
41
42 def test_limit_order_requires_price(self):
43 with self.assertRaises(SchwabError):
44 build_equity_order("SELL", "MSFT", 2, order_type="LIMIT")
45
46 def test_save_tokens_uses_owner_only_permissions(self):
47 with tempfile.TemporaryDirectory() as temp_dir:
48 token_file = Path(temp_dir) / "tokens.json"
49 save_tokens(token_file, {"access_token": "x"})
50
51 self.assertEqual(0o600, token_file.stat().st_mode & 0o777)
52
53 def test_config_from_env_requires_credentials(self):
54 old_env = os.environ.copy()
55 try:
56 for key in ("SCHWAB_APP_KEY", "SCHWAB_APP_SECRET", "SCHWAB_REDIRECT_URI"):
57 os.environ.pop(key, None)
58 with self.assertRaises(SchwabError):
59 SchwabConfig.from_env()
60 finally:
61 os.environ.clear()
62 os.environ.update(old_env)
63
64
65 if __name__ == "__main__":
66 unittest.main()
67