diff --git a/README.md b/README.md index 8e16e43..b2d711a 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,7 @@ __Table of Contents__ * [Usage](#usage) * [Authentication](#authentication) * [Web login](#web-login) + * [Other accounts of your login](#other-accounts-of-your-login) * [Development](#development) * [Setting Up a Development Environment](#setting-up-a-development-environment) * [Linting and Code Formatting](#linting-and-code-formatting) @@ -61,16 +62,18 @@ If you want to use the cutting-edge version, use this command instead: ```console usage: pytr [-h] [-V] [-v {warning,info,debug}] [--debug-logfile DEBUG_LOGFILE] [--debug-log-filter DEBUG_LOG_FILTER] - {help,login,portfolio,rates,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,get_savings_plans,completion} ... + {help,login,accounts,portfolio,rates,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,get_savings_plans,completion} ... Use "pytr command_name --help" to get detailed help to a specific command Commands: - {help,login,portfolio,rates,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,get_savings_plans,completion} + {help,login,accounts,portfolio,rates,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,get_savings_plans,completion} Desired action to perform help Print this help message login Check if credentials file exists. If not create it and ask for input. Try to login. Ask for device reset if needed + accounts List accounts of this login, e.g. a company accounts or child accounts. Use them + with --account portfolio Show current portfolio rates Fetch current prices for a list of ISINs given as direct list or CSV input details Get details for an ISIN @@ -116,6 +119,22 @@ Web login uses the public web-login endpoints at `api.traderepublic.com`. Two va Both variants keep you logged in on your primary device, but you may need to re-authenticate every so often when running `pytr`. +### Other accounts of your login + +A Trade Republic login can hold more than its own account, e.g. company accounts or child accounts. Trade Republic +treats each of them as a customer of its own, and the web app switches between them after the login. Pass `--account` +to do the same, on any subcommand that performs a login: + +```sh +pytr portfolio --account LEGAL_ENTITY +pytr dl_docs --account "Mustermann Holding GmbH" ./docs +``` + +The value is the account's type, its name or its customer id, as listed by `pytr accounts`. Without `--account`, +pytr uses your own account. The choice is not remembered between runs. + +Unfortunately, it seems that this only works with company accounts, not with child accounts at the moment. + ## Development ### Setting Up a Development Environment diff --git a/pytr/account.py b/pytr/account.py index 07ab2e3..b5203fe 100644 --- a/pytr/account.py +++ b/pytr/account.py @@ -4,9 +4,10 @@ import time from getpass import getpass +import requests from pygments import formatters, highlight, lexers -from .api import BASE_DIR, CREDENTIALS_FILE, TradeRepublicApi +from .api import BASE_DIR, CREDENTIALS_FILE, TradeRepublicApi, account_kind, account_name from .utils import get_logger @@ -19,10 +20,34 @@ def get_settings(tr): return formatted_json -def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwright", v2=False): +def print_accounts(tr): + """List the accounts of this login, one per line: type, name, customer id and state. + + `--account` takes the type, the name or, where those are not unique, the customer id. + """ + rows = [ + ( + account_kind(r), + account_name(r), + str(r.get("customerId") or ""), + "your own account" if r.get("relationshipType") == "SELF" else str(r.get("accountState") or ""), + ) + for r in tr.relationships() + ] + if not rows: + return "" + heading = ("TYPE", "NAME", "CUSTOMER ID", "STATE") + widths = [max(len(heading[i]), max(len(row[i]) for row in rows)) for i in range(4)] + all_rows = [heading] + rows + return "\n".join(" ".join(f"{cell:<{width}}" for cell, width in zip(row, widths)).rstrip() for row in all_rows) + + +def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwright", v2=False, account=None): """ Handle credentials parameters and store to credentials file if requested. If no parameters are set but are needed then ask for input + + `account` selects another account of this login, e.g. a company account. """ log = get_logger(__name__) save_cookies = True @@ -89,4 +114,15 @@ def login(phone_no=None, pin=None, store_credentials=False, waf_token="playwrigh log.info("Logged in.") log.debug(get_settings(tr)) + + if account is not None: + try: + tr.switch_account(account) + except ValueError as e: + log.fatal(str(e)) + sys.exit(1) + except requests.exceptions.HTTPError as e: + status = e.response.status_code if e.response is not None else "unknown" + log.fatal(f"Trade Republic refused to switch to account {account!r} (status {status}).") + sys.exit(1) return tr diff --git a/pytr/api.py b/pytr/api.py index 3a42f79..3fb5ec1 100644 --- a/pytr/api.py +++ b/pytr/api.py @@ -32,6 +32,7 @@ import time import urllib.parse import uuid +from collections import deque from datetime import datetime from http.cookiejar import Cookie, MozillaCookieJar from typing import Any, Dict @@ -87,6 +88,38 @@ WEB_PLATFORM = "web-pro" +# Seconds after which the web session is refreshed. +SESSION_REFRESH_INTERVAL = 290 + + +def account_kind(relationship): + """The kind of account behind an entry of `relationships()`, e.g. `ADULT` or `LEGAL_ENTITY`.""" + return str(relationship.get("accountType") or relationship.get("relationshipType") or "UNKNOWN") + + +def account_name(relationship): + """The name of the account behind an entry of `relationships()`: the account's, else the person's.""" + person = [str(part) for part in (relationship.get("firstName"), relationship.get("lastName")) if part] + return str(relationship.get("accountName") or " ".join(person)) + + +def _account_selectors(relationship): + """Everything an entry of `relationships()` can be selected by, lower-cased.""" + selectors = [ + relationship.get("customerId"), + relationship.get("relationshipType"), + relationship.get("accountType"), + relationship.get("firstName"), + account_name(relationship), + ] + return {str(selector).strip().casefold() for selector in selectors if selector and str(selector).strip()} + + +def _describe_account(relationship): + name = account_name(relationship) + return f"{account_kind(relationship)} ({name})" if name else account_kind(relationship) + + class TradeRepublicApi: _default_headers = { "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36" @@ -148,6 +181,10 @@ def __init__( if self._save_cookies: self._websession.cookies = MozillaCookieJar(self._cookies_file) self._sec_acc_no: str | None = None + # Customer id of the account the session acts for, once it is not the login's own. + self._subject_id: str | None = None + # Messages read off the websocket on behalf of a caller that has not asked for them yet. + self._held_back: deque = deque() def _fetch_waf_token_awswaf(self): """ @@ -516,11 +553,81 @@ def resume_websession(self): def _web_request(self, url_path, payload=None, method="GET"): if self._session_expires_at < time.time(): - r = self._websession.get(f"{self._host}/api/v1/auth/web/session") + if self._subject_id is None: + # Refreshing without a subject returns the session to the login's own account. + # A new process starts here, so it never inherits the account an earlier run chose. + r = self._websession.get(f"{self._host}/api/v1/auth/web/session") + else: + r = self._post_session(self._subject_id) r.raise_for_status() - self._session_expires_at = time.time() + 290 + self._session_expires_at = time.time() + SESSION_REFRESH_INTERVAL return self._websession.request(method=method, url=f"{self._host}{url_path}", data=payload) + def _post_session(self, subject_id): + """Ask for session cookies that act for `subject_id`, the way the web frontend does. + + The frontend uses this call both to switch the account and to refresh the session. + """ + return self._websession.post( + f"{self._host}/api/v2/auth/web/session", + json={"subjectId": subject_id}, + headers=self._login_headers(), + ) + + def relationships(self): + """List the accounts this login can act for. + + Besides the login's own account (`relationshipType` `SELF`) this is e.g. a + company account (`LEGAL_ENTITY_ACTOR`). + """ + r = self._web_request("/api/v1/customer/relationships/detailed") + r.raise_for_status() + return r.json().get("relationships") or [] + + def _find_relationship(self, account): + """Pick one relationship by customer id, type or name. `None` picks the login's own account.""" + relationships = self.relationships() + if account is None: + matches = [r for r in relationships if r.get("relationshipType") == "SELF"] + wanted = "Your own account" + else: + matches = [r for r in relationships if account.strip().casefold() in _account_selectors(r)] + wanted = repr(account) + if len(matches) == 1 and matches[0].get("customerId"): + return matches[0] + known = ", ".join(_describe_account(r) for r in relationships) or "none" + if len(matches) > 1: + raise ValueError(f"{wanted} matches more than one account, use its customer id. Accounts: {known}.") + raise ValueError(f"{wanted} was not found among the accounts of this login. Accounts: {known}.") + + def switch_account(self, account=None): + """Let the session act for another account of this login, e.g. a company account. + + `account` is the account's type (`LEGAL_ENTITY`), its name or its customer id, as listed + by `relationships()`. `None` switches back to the login's own account. Returns the chosen + relationship. + + Trade Republic does not address these accounts by a second account number. It hands + out new session cookies that act for the other customer, so every call made afterwards, + the websocket included, sees that account only. + """ + if self._ws is not None and self._ws.close_code is None: + raise ValueError("Close the websocket connection before switching the account.") + + relationship = self._find_relationship(account) + r = self._post_session(relationship["customerId"]) + r.raise_for_status() + + is_self = relationship.get("relationshipType") == "SELF" + self._subject_id = None if is_self else relationship["customerId"] + self._session_expires_at = time.time() + SESSION_REFRESH_INTERVAL + # The securities account belongs to the previous account. + self._sec_acc_no = None + # The switch replaced the session cookies; keep the saved ones usable for the next run. + self.save_websession() + self.log.info(f"Using account: {_describe_account(relationship)}") + return relationship + async def _get_ws(self): if self._ws and self._ws.close_code is None: return self._ws @@ -591,6 +698,11 @@ async def unsubscribe(self, subscription_id): self._previous_responses.pop(subscription_id, None) async def recv(self): + if self._held_back: + return self._held_back.popleft() + return await self._recv_ws() + + async def _recv_ws(self): ws = await self._get_ws() while True: response = await ws.recv() @@ -669,11 +781,42 @@ async def portfolio_status(self): async def compact_portfolio(self): if self._sec_acc_no is None: - self.settings() + if self._subject_id is None: + self.settings() + else: + # /api/v2/auth/account answers 400 INVALID_AUTH_ACCOUNT_STATE for any account but + # the login's own, so the number has to come from the websocket instead. + self._sec_acc_no = await self._sec_acc_no_from_account_pairs() if self._sec_acc_no is None: - raise ValueError("Could not retrieve securities account number from account settings.") + source = "account settings" if self._subject_id is None else "account pairs" + raise ValueError(f"Could not retrieve securities account number from {source}.") return await self.subscribe({"type": "compactPortfolioByType", "secAccNo": self._sec_acc_no}) + async def account_pairs(self): + return await self.subscribe({"type": "accountPairs"}) + + async def _sec_acc_no_from_account_pairs(self, timeout=5.0): + """Read the securities account number of the account the session acts for.""" + subscription_id = await self.account_pairs() + try: + response = await asyncio.wait_for(self._recv_holding_back_others(subscription_id), timeout) + except asyncio.TimeoutError: + return None + finally: + await self.unsubscribe(subscription_id) + accounts = [a for a in response.get("accounts") or [] if a.get("securitiesAccountNumber")] + # Prefer the pair of the default product if the account has more than one. + accounts.sort(key=lambda a: a.get("productType") != "DEFAULT") + return accounts[0]["securitiesAccountNumber"] if accounts else None + + async def _recv_holding_back_others(self, subscription_id): + """Wait for one subscription's answer; keep the answers to others for the next `recv()`.""" + while True: + message = await self._recv_ws() + if message[0] == subscription_id: + return message[2] + self._held_back.append(message) + async def watchlist(self): return await self.subscribe({"type": "watchlist"}) diff --git a/pytr/main.py b/pytr/main.py index ecc4a87..d994f56 100644 --- a/pytr/main.py +++ b/pytr/main.py @@ -12,7 +12,7 @@ import shtab -from pytr.account import login +from pytr.account import login, print_accounts from pytr.alarms import Alarms from pytr.details import Details from pytr.dl import DL @@ -100,6 +100,15 @@ def formatter(prog): const=None, dest="waf_token", ) + parser_login_args.add_argument( + "--account", + help=( + "Use another account of this login, e.g. a company account. Give its type " + '(e.g. "LEGAL_ENTITY"), name or customer id as listed by "pytr accounts". ' + "Defaults to your own account." + ), + default=None, + ) parser_login_args.add_argument( "--store_credentials", help="Store credentials (Phone number, pin, cookies) for next usage", @@ -166,6 +175,16 @@ def formatter(prog): description=info, ) + # accounts + info = "List accounts of this login, e.g. a company accounts or child accounts. Use them with --account" + parser_cmd.add_parser( + "accounts", + formatter_class=formatter, + parents=[parser_login_args], + help=info, + description=info, + ) + # portfolio info = "Show current portfolio" parser_portfolio = parser_cmd.add_parser( @@ -560,6 +579,20 @@ def main(): store_credentials=args.store_credentials, waf_token=args.waf_token, v2=args.v2, + account=args.account, + ) + elif args.command == "accounts": + print( + print_accounts( + login( + phone_no=args.phone_no, + pin=args.pin, + store_credentials=args.store_credentials, + waf_token=args.waf_token, + v2=args.v2, + account=args.account, + ) + ) ) elif args.command == "portfolio": Portfolio( @@ -569,6 +602,7 @@ def main(): store_credentials=args.store_credentials, waf_token=args.waf_token, v2=args.v2, + account=args.account, ), args.include_watchlist, instruments_to_ignore=re.split(r"[,;]", args.ignore) if args.ignore else [], @@ -590,6 +624,7 @@ def main(): store_credentials=args.store_credentials, waf_token=args.waf_token, v2=args.v2, + account=args.account, ), isins, output=args.output, @@ -606,6 +641,7 @@ def main(): store_credentials=args.store_credentials, waf_token=args.waf_token, v2=args.v2, + account=args.account, ), args.isin, ).get() @@ -619,6 +655,7 @@ def main(): store_credentials=args.store_credentials, waf_token=args.waf_token, v2=args.v2, + account=args.account, ), args.output, args.format, @@ -653,6 +690,7 @@ def main(): store_credentials=args.store_credentials, waf_token=args.waf_token, v2=args.v2, + account=args.account, ), args.outputdir, not_before, @@ -688,6 +726,8 @@ def main(): pin=args.pin, store_credentials=args.store_credentials, waf_token=args.waf_token, + v2=args.v2, + account=args.account, ), args.input, args.outputfile, @@ -703,6 +743,8 @@ def main(): pin=args.pin, store_credentials=args.store_credentials, waf_token=args.waf_token, + v2=args.v2, + account=args.account, ), args.input, args.inputfile, @@ -719,6 +761,7 @@ def main(): store_credentials=args.store_credentials, waf_token=args.waf_token, v2=args.v2, + account=args.account, ), args.outputfile, decimal_localization=args.decimal_localization, diff --git a/tests/test_accounts.py b/tests/test_accounts.py new file mode 100644 index 0000000..5c843f4 --- /dev/null +++ b/tests/test_accounts.py @@ -0,0 +1,511 @@ +"""Pin how pytr reaches an account of the login other than its own, e.g. a company account.""" + +import asyncio +import json as jsonlib +from typing import Any + +import pytest +import requests + +import pytr.account as account_module +from pytr.api import TradeRepublicApi +from pytr.main import get_main_parser + +HOST = "https://api.traderepublic.com" +SESSION_V1 = f"{HOST}/api/v1/auth/web/session" +SESSION_V2 = f"{HOST}/api/v2/auth/web/session" +RELATIONSHIPS = f"{HOST}/api/v1/customer/relationships/detailed" +ACCOUNT = f"{HOST}/api/v2/auth/account" + +SELF = { + "customerId": "id-self", + "firstName": "Erika", + "lastName": "Mustermann", + "relationshipType": "SELF", + "accountState": "ACTIVE", + "accountType": "ADULT", +} +COMPANY = { + "customerId": "id-company", + "relationshipType": "LEGAL_ENTITY_ACTOR", + "accountState": "ACTIVE", + "accountType": "LEGAL_ENTITY", + "accountName": "Mustermann Holding GmbH", +} +SECOND_COMPANY = {**COMPANY, "customerId": "id-second", "accountName": "Mustermann Immobilien UG"} +BOTH = {"relationships": [SELF, COMPANY]} + + +class _Response: + def __init__(self, payload: Any = None, status_code: int = 200): + self.status_code = status_code + self._payload = {} if payload is None else payload + + def __bool__(self) -> bool: + # Like requests.Response: falsy for 4xx and 5xx. + return self.status_code < 400 + + def raise_for_status(self) -> None: + if self.status_code >= 400: + raise requests.HTTPError(f"{self.status_code} Client Error", response=self) # type: ignore[arg-type] + + def json(self) -> Any: + return self._payload + + +class _Session: + """Records requests instead of sending them. Replies come from a queue; an unexpected request fails.""" + + def __init__(self, replies: list[Any]): + self.calls: list[dict[str, Any]] = [] + self.replies = list(replies) + self.headers = dict(TradeRepublicApi._default_headers) + + def _record(self, method: str, url: str, json: Any = None, headers: Any = None) -> _Response: + self.calls.append({"method": method, "url": url, "json": json, "headers": headers or {}}) + assert self.replies, f"unexpected request: {method} {url}" + reply = self.replies.pop(0) + status, payload = reply if isinstance(reply, tuple) else (200, reply) + return _Response(payload, status) + + def post(self, url, json=None, headers=None): + return self._record("POST", url, json, headers) + + def get(self, url, headers=None): + return self._record("GET", url, None, headers) + + def request(self, method, url, data=None): + return self._record(method, url, data, None) + + +class _Websocket: + """Answers each subscription with the messages listed for its topic: (code, topic answered, payload).""" + + close_code = None + + def __init__(self, answers: dict[str, list[tuple[str, str, Any]]]): + self.sent: list[str] = [] + self._answers = answers + self._ids: dict[str, str] = {} + self._inbox: asyncio.Queue[str] = asyncio.Queue() + + async def send(self, message: str) -> None: + self.sent.append(message) + command, subscription_id, *payload = message.split(" ", 2) + if command != "sub": + return + topic = jsonlib.loads(payload[0])["type"] + self._ids[topic] = subscription_id + for code, answered_topic, body in self._answers.get(topic, []): + await self._inbox.put(f"{self._ids[answered_topic]} {code} {jsonlib.dumps(body)}") + + async def recv(self) -> str: + return await self._inbox.get() + + def subscriptions(self) -> list[dict[str, Any]]: + return [jsonlib.loads(m.split(" ", 2)[2]) for m in self.sent if m.startswith("sub ")] + + +def _api(replies, session_is_fresh=True): + tr = TradeRepublicApi(phone_no="+490000000000", pin="0000", waf_token=None, save_cookies=False) + tr._websession = _Session(replies) + # These are class attributes; keep what one test subscribes to out of the next. + tr.subscriptions = {} + tr._previous_responses = {} + tr._subscription_id_counter = 1 + if session_is_fresh: + tr._session_expires_at = float("inf") + return tr + + +def _switched(extra_replies=()): + """An API whose session acts for the company account.""" + tr = _api([BOTH, {}, *extra_replies]) + tr.switch_account("LEGAL_ENTITY") + tr._websession.calls.clear() + return tr + + +def _with_websocket(tr, answers): + ws = _Websocket(answers) + + async def get_ws(): + return ws + + tr._get_ws = get_ws + return ws + + +def _calls(tr): + return [(c["method"], c["url"]) for c in tr._websession.calls] + + +# --- listing the accounts ------------------------------------------------------------ + + +def test_relationships_lists_the_accounts_of_the_login(): + tr = _api([BOTH]) + + assert tr.relationships() == [SELF, COMPANY] + assert _calls(tr) == [("GET", RELATIONSHIPS)] + + +@pytest.mark.parametrize("payload", [{}, {"relationships": None}]) +def test_relationships_of_a_login_without_any_is_empty(payload): + assert _api([payload]).relationships() == [] + + +# --- switching ----------------------------------------------------------------------- + + +def test_switch_account_asks_for_a_session_acting_for_the_other_customer(): + tr = _api([BOTH, {}]) + tr._sec_acc_no = "own-depot" + + chosen = tr.switch_account("LEGAL_ENTITY") + + assert chosen == COMPANY + assert _calls(tr) == [("GET", RELATIONSHIPS), ("POST", SESSION_V2)] + assert tr._websession.calls[1]["json"] == {"subjectId": "id-company"} + assert tr._subject_id == "id-company" + # The securities account of the own account must not leak into the other one. + assert tr._sec_acc_no is None + + +def test_switch_account_sends_the_headers_of_the_v2_login_calls(): + tr = _api([BOTH, {}]) + + tr.switch_account("LEGAL_ENTITY") + + for header in ("X-TR-Device-Info", "X-TR-App-Version", "X-Tr-Platform"): + assert tr._websession.calls[1]["headers"].get(header) + + +def test_switch_account_saves_the_new_cookies(): + tr = _api([BOTH, {}]) + saved = [] + tr.save_websession = lambda: saved.append(tr._subject_id) + + tr.switch_account("LEGAL_ENTITY") + + assert saved == ["id-company"] + + +@pytest.mark.parametrize( + "account", + ["id-company", "legal_entity", "LEGAL_ENTITY_ACTOR", "mustermann holding gmbh", " Mustermann Holding GmbH "], +) +def test_switch_account_finds_the_account_by_id_type_or_name(account): + tr = _api([BOTH, {}]) + + assert tr.switch_account(account) == COMPANY + + +@pytest.mark.parametrize("account", ["Erika", "erika mustermann", "ADULT", "self"]) +def test_switch_account_finds_a_person_by_first_or_full_name(account): + tr = _api([BOTH, {}]) + + assert tr.switch_account(account) == SELF + + +@pytest.mark.parametrize("account", ["Mustermann", "Holding", "LEGAL", "id-", ""]) +def test_switch_account_does_not_match_on_parts_of_a_name(account): + tr = _api([BOTH]) + + with pytest.raises(ValueError, match="was not found"): + tr.switch_account(account) + + +def test_switch_account_without_a_name_returns_to_the_own_account(): + tr = _switched([BOTH, {}]) + + assert tr.switch_account() == SELF + assert tr._websession.calls[1]["json"] == {"subjectId": "id-self"} + assert tr._subject_id is None + + +def test_switch_account_reports_a_login_without_an_own_account(): + tr = _api([{"relationships": [COMPANY]}]) + + with pytest.raises(ValueError, match="Your own account was not found"): + tr.switch_account() + + +def test_switch_account_rejects_an_unknown_account_and_names_the_known_ones(): + tr = _api([BOTH]) + + with pytest.raises(ValueError, match="'nope' was not found.*LEGAL_ENTITY \\(Mustermann Holding GmbH\\)"): + tr.switch_account("nope") + + assert _calls(tr) == [("GET", RELATIONSHIPS)] + + +def test_switch_account_rejects_an_ambiguous_account_until_given_its_id(): + """Two companies share their type, so the type alone does not say which one is meant.""" + accounts = {"relationships": [SELF, COMPANY, SECOND_COMPANY]} + tr = _api([accounts, accounts, {}]) + + with pytest.raises(ValueError, match="more than one account, use its customer id"): + tr.switch_account("LEGAL_ENTITY") + + assert tr.switch_account("id-second") == SECOND_COMPANY + + +def test_switch_account_rejects_an_account_without_a_customer_id(): + tr = _api([{"relationships": [SELF, {"accountType": "LEGAL_ENTITY", "accountName": "Nameless"}]}]) + + with pytest.raises(ValueError, match="was not found"): + tr.switch_account("LEGAL_ENTITY") + + +def test_switch_account_copes_with_fields_that_are_not_text(): + odd = {"customerId": 42, "firstName": 1, "lastName": None, "relationshipType": "LEGAL_ENTITY_ACTOR"} + tr = _api([{"relationships": [SELF, odd]}, {}]) + + assert tr.switch_account("42") == odd + + +def test_a_refused_switch_leaves_the_session_as_it_was(): + tr = _api([BOTH, (403, {})]) + tr._sec_acc_no = "own-depot" + tr._session_expires_at = expires_at = 10**10 + tr.save_websession = lambda: pytest.fail("must not save the cookies of a refused switch") + + with pytest.raises(requests.HTTPError): + tr.switch_account("LEGAL_ENTITY") + + assert (tr._subject_id, tr._sec_acc_no, tr._session_expires_at) == (None, "own-depot", expires_at) + + +def test_switch_account_refuses_while_the_websocket_is_connected(): + """An open websocket keeps answering for the account it was authenticated with.""" + tr = _api([]) + tr._ws = _Websocket({}) + + with pytest.raises(ValueError, match="websocket"): + tr.switch_account("LEGAL_ENTITY") + + assert tr._websession.calls == [] + + +# --- keeping the session on the chosen account --------------------------------------- + + +def test_the_switch_counts_as_a_session_refresh(): + tr = _api([BOTH, {}, {}]) + tr._session_expires_at = 10**10 # far enough ahead that only the switch can have moved it + + tr.switch_account("LEGAL_ENTITY") + tr._web_request("/api/v1/whatever") + + assert tr._session_expires_at < 10**10 + assert _calls(tr) == [("GET", RELATIONSHIPS), ("POST", SESSION_V2), ("GET", f"{HOST}/api/v1/whatever")] + + +def test_session_refresh_stays_on_the_chosen_account(): + """A refresh without the subject would silently return the session to the own account.""" + tr = _switched([{}, {}]) + tr._session_expires_at = 0 + + tr._web_request("/api/v1/whatever") + + assert _calls(tr) == [("POST", SESSION_V2), ("GET", f"{HOST}/api/v1/whatever")] + assert tr._websession.calls[0]["json"] == {"subjectId": "id-company"} + + +def test_a_new_process_starts_on_the_own_account(): + """Its first request refreshes without a subject, whatever account the saved cookies act for.""" + tr = _api([{}, {}], session_is_fresh=False) + + tr.settings() + + assert _calls(tr) == [("GET", SESSION_V1), ("GET", ACCOUNT)] + + +# --- portfolio ----------------------------------------------------------------------- + +PAIRS = { + "authAccountId": "id-company", + "accounts": [ + {"securitiesAccountNumber": "depot-other", "cashAccountNumber": "cash-other", "productType": "OTHER"}, + {"securitiesAccountNumber": "depot-company", "cashAccountNumber": "cash-company", "productType": "DEFAULT"}, + ], +} + + +def test_compact_portfolio_reads_the_securities_account_from_account_pairs_after_a_switch(): + tr = _switched() + ws = _with_websocket(tr, {"accountPairs": [("A", "accountPairs", PAIRS)]}) + + asyncio.run(tr.compact_portfolio()) + + assert ws.subscriptions() == [ + {"type": "accountPairs"}, + {"type": "compactPortfolioByType", "secAccNo": "depot-company"}, + ] + # The account settings are refused for this account, so they are not asked for. + assert _calls(tr) == [] + assert "unsub 1" in ws.sent + + +def test_compact_portfolio_takes_the_first_pair_when_none_is_the_default_product(): + tr = _switched() + pairs = {"accounts": [{"cashAccountNumber": "c"}, {"securitiesAccountNumber": "depot-a", "productType": "X"}]} + ws = _with_websocket(tr, {"accountPairs": [("A", "accountPairs", pairs)]}) + + asyncio.run(tr.compact_portfolio()) + + assert ws.subscriptions()[-1]["secAccNo"] == "depot-a" + + +def test_compact_portfolio_keeps_the_answers_to_other_subscriptions(): + """Looking up the securities account must not swallow what the caller subscribed to before.""" + tr = _switched() + _with_websocket( + tr, + { + "cash": [], + "accountPairs": [("A", "cash", [{"amount": 1}]), ("A", "accountPairs", PAIRS)], + "compactPortfolioByType": [("A", "compactPortfolioByType", {"categories": []})], + }, + ) + + async def cash_then_portfolio(): + await tr.cash() + await tr.compact_portfolio() + return [await tr.recv(), await tr.recv()] + + received = asyncio.run(asyncio.wait_for(cash_then_portfolio(), 5)) + + assert [(subscription["type"], payload) for _, subscription, payload in received] == [ + ("cash", [{"amount": 1}]), + ("compactPortfolioByType", {"categories": []}), + ] + + +@pytest.mark.parametrize( + "answers", + [ + [], # no answer at all + [("A", "accountPairs", {"authAccountId": "id-company", "accounts": []})], + [("A", "accountPairs", {"authAccountId": "id-company", "accounts": None})], + ], +) +def test_compact_portfolio_reports_a_missing_securities_account_after_a_switch(answers, monkeypatch): + tr = _switched() + ws = _with_websocket(tr, {"accountPairs": answers}) + lookup = tr._sec_acc_no_from_account_pairs + monkeypatch.setattr(tr, "_sec_acc_no_from_account_pairs", lambda: lookup(timeout=0.05)) + + with pytest.raises(ValueError, match="securities account number from account pairs"): + asyncio.run(tr.compact_portfolio()) + + assert "unsub 1" in ws.sent + + +def test_compact_portfolio_is_unchanged_for_the_own_account(): + tr = _api([{"securitiesAccountNumber": "own-depot"}]) + ws = _with_websocket(tr, {}) + + asyncio.run(tr.compact_portfolio()) + + assert _calls(tr) == [("GET", ACCOUNT)] + assert ws.subscriptions() == [{"type": "compactPortfolioByType", "secAccNo": "own-depot"}] + + +def test_compact_portfolio_still_reports_account_settings_without_a_securities_account(): + tr = _api([{}]) + ws = _with_websocket(tr, {}) + + with pytest.raises(ValueError, match="securities account number from account settings"): + asyncio.run(tr.compact_portfolio()) + + assert ws.sent == [] + + +# --- pytr accounts ------------------------------------------------------------------- + + +def test_print_accounts_prints_what_account_accepts(): + tr = _api([{"relationships": [SELF, COMPANY, {"customerId": "id-bare"}]}]) + + assert account_module.print_accounts(tr).splitlines() == [ + "TYPE NAME CUSTOMER ID STATE", + "ADULT Erika Mustermann id-self your own account", + "LEGAL_ENTITY Mustermann Holding GmbH id-company ACTIVE", + "UNKNOWN id-bare", + ] + + +def test_print_accounts_of_a_login_without_relationships_is_empty(): + assert account_module.print_accounts(_api([{}])) == "" + + +def test_the_accounts_command_exists_and_takes_the_login_arguments(): + args = get_main_parser().parse_args(["accounts", "--v2"]) + + assert (args.command, args.v2, args.account) == ("accounts", True, None) + + +def test_every_command_that_logs_in_takes_an_account(): + parser = get_main_parser() + + for command in (["login"], ["portfolio"], ["details", "DE0000000000"], ["dl_docs", "out"], ["get_price_alarms"]): + assert parser.parse_args([*command, "--account", "LEGAL_ENTITY"]).account == "LEGAL_ENTITY" + + +# --- login() ------------------------------------------------------------------------- + + +class _LoggedIn: + """Stands in for a TradeRepublicApi whose saved session could be resumed.""" + + def __init__(self, switch_error=None): + self._switch_error = switch_error + self.events: list[Any] = [] + + def resume_websession(self): + return True + + def settings(self): + self.events.append("settings") + return {} + + def switch_account(self, account=None): + self.events.append(("switch", account)) + if self._switch_error: + raise self._switch_error + + +def _login(monkeypatch, tmp_path, tr, **kwargs): + # login() creates the directory it stores credentials in; keep it off the real one. + monkeypatch.setattr(account_module, "BASE_DIR", tmp_path / "pytr") + monkeypatch.setattr(account_module, "CREDENTIALS_FILE", tmp_path / "pytr" / "credentials") + monkeypatch.setattr(account_module, "TradeRepublicApi", lambda **_: tr) + return account_module.login(phone_no="+490000000000", pin="0000", **kwargs) + + +def test_login_does_not_switch_by_default(monkeypatch, tmp_path): + tr = _LoggedIn() + + assert _login(monkeypatch, tmp_path, tr) is tr + assert tr.events == ["settings"] + + +def test_login_switches_to_the_requested_account(monkeypatch, tmp_path): + tr = _LoggedIn() + + assert _login(monkeypatch, tmp_path, tr, account="LEGAL_ENTITY") is tr + # The settings are read first: they are refused once the session acts for the other account. + assert tr.events == ["settings", ("switch", "LEGAL_ENTITY")] + + +@pytest.mark.parametrize( + "error", + [ValueError("'nope' was not found"), requests.HTTPError("403", response=_Response({}, 403))], # type: ignore[arg-type] +) +def test_login_exits_when_the_account_cannot_be_used(monkeypatch, tmp_path, error): + with pytest.raises(SystemExit) as excinfo: + _login(monkeypatch, tmp_path, _LoggedIn(switch_error=error), account="nope") + + assert excinfo.value.code == 1