Merge pull request #9 from BITNP:response-types

feat: better error handling and type defs

README.md

@@ -29,10 +29,9 @@ Usage: bitsrun login/logout [OPTIONS]
   Log into or out of the BIT network.
 
 Options:
-  -u, --username TEXT  Username.
-  -p, --password TEXT  Password.
+  -u, --username TEXT  Your username.
+  -p, --password TEXT  Your password.
   -v, --verbose        Verbosely echo API response.
-  -s, --silent         Silent, no output to stdout.
   --help               Show this message and exit.
 ```
 

bitsrun/cli.py

@@ -1,5 +1,6 @@
 import sys
 from getpass import getpass
+from pprint import pprint
 
 import click
 
@@ -9,10 +10,9 @@
 # A hacky way to specify shared options for multiple click commands:
 # https://stackoverflow.com/questions/40182157/shared-options-and-flags-between-commands
 _options = [
-    click.option("-u", "--username", help="Username.", required=False),
-    click.option("-p", "--password", help="Password.", required=False),
+    click.option("-u", "--username", help="Your username.", required=False),
+    click.option("-p", "--password", help="Your password.", required=False),
     click.option("-v", "--verbose", is_flag=True, help="Verbosely echo API response."),
-    click.option("-s", "--silent", is_flag=True, help="Silent, no output to stdout."),
 ]
 
 
@@ -41,55 +41,70 @@ def config_paths():
 
 @cli.command()
 @add_options(_options)
-def login(username, password, verbose, silent):
+def login(username, password, verbose):
     """Log into the BIT network."""
-    do_action("login", username, password, verbose, silent)
+    do_action("login", username, password, verbose)
 
 
 @cli.command()
 @add_options(_options)
-def logout(username, password, verbose, silent):
+def logout(username, password, verbose):
     """Log out of the BIT network."""
-    do_action("logout", username, password, verbose, silent)
+    do_action("logout", username, password, verbose)
 
 
-def do_action(action, username, password, verbose, silent):
-    """Log in/out the BIT network."""
+def do_action(action, username, password, verbose):
+    # Support reading password from stdin when not passed via `--password`
     if username and not password:
         password = getpass(prompt="Please enter your password: ")
+
+    # Try to read username and password from args provided. If none, look for config
+    # files in possible paths. If none, fail and prompt user to provide one.
     if username and password:
         user = User(username, password)
     elif conf := read_config():
-        user = User(*conf)
+        user = User(**conf[0])
+        if verbose:
+            click.echo(
+                click.style("bitsrun: ", fg="blue")
+                + "Reading config from "
+                + click.style(conf[1], fg="yellow", underline=True)
+            )
     else:
         ctx = click.get_current_context()
-        ctx.fail("No username/password provided.")
+        ctx.fail("No username or password provided")
 
     try:
         if action == "login":
-            res = user.login()
-
-            # Output login result by default if not silent
-            if not silent:
-                click.echo(f"{res.get('username')} ({res.get('online_ip')}) logged in")
-
+            resp = user.login()
+            message = f"{user.username} ({resp['online_ip']}) logged in"
         elif action == "logout":
-            res = user.logout()
-
-            # Output logout result by default if not silent
-            if not silent:
-                click.echo(f"{res.get('online_ip')} logged out")
-
+            resp = user.logout()
+            message = f"{resp['online_ip']} logged out"
         else:
             # Should not reach here, but just in case
-            raise ValueError(f"unknown action `{action}`")
+            raise ValueError(f"Unknown action `{action}`")
 
-        # Output direct result of response if verbose
+        # Output direct result of the API response if verbose
         if verbose:
-            click.echo(f"{click.style('info:', fg='blue')} {res}")
+            click.echo(f"{click.style('bitsrun:', fg='cyan')} Response from API:")
+            pprint(resp)
+
+        # Handle error from API response. When field `error` is not `ok`, then the
+        # login/logout action has likely failed. Hints are provided in the `error_msg`.
+        if resp["error"] != "ok":
+            raise Exception(
+                resp["error_msg"]
+                if resp["error_msg"]
+                else "Action failed, use --verbose for more info"
+            )
+
+        # Print success message
+        click.echo(f"{click.style('bitsrun:', fg='green')} {message}")
 
     except Exception as e:
-        click.echo(f"{click.style('error:', fg='red')} {e}")
+        # Exception is caught and printed to stderr
+        click.echo(f"{click.style('error:', fg='red')} {e}", err=True)
         # Throw with error code 1 for scripts to pick up error state
         sys.exit(1)
 

bitsrun/config.py

@@ -2,7 +2,7 @@
 from os import getenv
 from pathlib import Path
 from sys import platform
-from typing import Optional, Tuple
+from typing import Optional, Tuple, TypedDict
 
 from platformdirs import site_config_path, user_config_path
 
@@ -49,7 +49,12 @@ def get_config_paths() -> map:
     return map(lambda path: path / "bit-user.json", paths)
 
 
-def read_config() -> Optional[Tuple[str, str]]:
+class ConfigType(TypedDict):
+    username: str
+    password: str
+
+
+def read_config() -> Optional[Tuple[ConfigType, str]]:
     """Read config from the first available config file with name `bit-user.json`.
 
     The config file should be a JSON file with the following structure:
@@ -59,15 +64,15 @@ def read_config() -> Optional[Tuple[str, str]]:
     ```
 
     Returns:
-        A tuple of (username, password) if the config file is found.
+        A tuple of (config, path to config file) if the config file is found.
     """
 
     paths = get_config_paths()
     for path in paths:
         try:
             with open(path) as f:
                 data = json.loads(f.read())
-                return data["username"], data["password"]
+                return data, str(path)
         except Exception:
             continue
     return None

bitsrun/user.py

@@ -2,31 +2,42 @@
 import json
 from enum import Enum
 from hashlib import sha1
-from typing import Dict, Optional, Union
+from typing import Dict, Literal, Optional, TypedDict, Union
 
 from requests import Session
 
 from bitsrun.utils import fkbase64, parse_homepage, xencode
 
-API_BASE = "http://10.0.0.55"
-TYPE_CONST = 1
-N_CONST = 200
+_API_BASE = "http://10.0.0.55"
+_TYPE_CONST = 1
+_N_CONST = 200
 
 
 class Action(Enum):
     LOGIN = "login"
     LOGOUT = "logout"
 
 
+class UserResponseType(TypedDict):
+    client_ip: str
+    online_ip: str
+    # Field `error` is also `login_error` when logout action fails
+    error: Union[Literal["login_error"], Literal["ok"]]
+    error_msg: str
+    res: Union[Literal["login_error"], Literal["ok"]]
+    # Field `username` is not present on login fails and all logout scenarios
+    username: Optional[str]
+
+
 class User:
     def __init__(self, username: str, password: str):
         self.username = username
         self.password = password
 
-        self.ip, self.acid = parse_homepage(api_base=API_BASE)
+        self.ip, self.acid = parse_homepage(api_base=_API_BASE)
         self.session = Session()
 
-    def login(self) -> Dict[str, Union[str, int]]:
+    def login(self) -> UserResponseType:
         logged_in_user = self._user_validate()
 
         # Raise exception if device is already logged in
@@ -35,7 +46,7 @@ def login(self) -> Dict[str, Union[str, int]]:
 
         return self._do_action(Action.LOGIN)
 
-    def logout(self) -> Dict[str, Union[str, int]]:
+    def logout(self) -> UserResponseType:
         logged_in_user = self._user_validate()
 
         # Raise exception if device is not logged in
@@ -44,9 +55,9 @@ def logout(self) -> Dict[str, Union[str, int]]:
 
         return self._do_action(Action.LOGOUT)
 
-    def _do_action(self, action: Action) -> Dict[str, Union[str, int]]:
+    def _do_action(self, action: Action) -> UserResponseType:
         params = self._make_params(action)
-        response = self.session.get(API_BASE + "/cgi-bin/srun_portal", params=params)
+        response = self.session.get(_API_BASE + "/cgi-bin/srun_portal", params=params)
         return json.loads(response.text[6:-1])
 
     def _get_user_info(self) -> Optional[str]:
@@ -56,7 +67,7 @@ def _get_user_info(self) -> Optional[str]:
             The username of the current logged in user if exists.
         """
 
-        resp = self.session.get(API_BASE + "/cgi-bin/rad_user_info")
+        resp = self.session.get(_API_BASE + "/cgi-bin/rad_user_info")
         data = resp.text
 
         if data == "not_online_error":
@@ -88,7 +99,7 @@ def _user_validate(self) -> Optional[str]:
 
     def _get_token(self) -> str:
         params = {"callback": "jsonp", "username": self.username, "ip": self.ip}
-        response = self.session.get(API_BASE + "/cgi-bin/get_challenge", params=params)
+        response = self.session.get(_API_BASE + "/cgi-bin/get_challenge", params=params)
         result = json.loads(response.text[6:-1])
         return result["challenge"]
 
@@ -101,8 +112,8 @@ def _make_params(self, action: Action) -> Dict[str, Union[int, str]]:
             "action": action.value,
             "ac_id": self.acid,
             "ip": self.ip,
-            "type": TYPE_CONST,
-            "n": N_CONST,
+            "type": _TYPE_CONST,
+            "n": _N_CONST,
         }
 
         data = {
@@ -123,8 +134,8 @@ def _make_params(self, action: Action) -> Dict[str, Union[int, str]]:
                 hmd5,
                 self.acid,
                 self.ip,
-                N_CONST,
-                TYPE_CONST,
+                _N_CONST,
+                _TYPE_CONST,
                 info,
             ).encode()
         ).hexdigest()

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "bitsrun"
-version = "3.2.4"
+version = "3.3.0"
 description = "A headless login / logout script for 10.0.0.55"
 authors = ["spencerwooo <[email protected]>"]
 license = "MIT"