CLI - Add auth, headers & timeout flags (#13)

* split on `:` or `=` for `auth` and `headers`
* update version and Readme

Author: Kilo59

README.md

@@ -29,9 +29,9 @@ Check CLI usage
 
 ```
 ❯ csv2http --help
-usage: csv2http [-h] [-c CONCURRENCY] [--method {POST,PATCH,PUT}] [-d] [-n] file url
+usage: csv2http [-h] [-c CONCURRENCY] [--method {POST,PATCH,PUT}] [-a AUTH] [-H [HEADER ...]] [-d] [-n] [-t TIMEOUT] file url
 
-HTTP request for every row of a CSV file - v0.0.2a
+HTTP request for every row of a CSV file - v0.0.3a
 
 positional arguments:
   file                  payload csv file
@@ -43,8 +43,13 @@ options:
                         Maximum number of concurrent requests (default: 25)
   --method {POST,PATCH,PUT}
                         HTTP method/verb (default: POST)
+  -a AUTH, --auth AUTH  Basic Authentication enter <USERNAME>:<PASSWORD>. If password is blank you will be prompted for input
+  -H [HEADER ...], --header [HEADER ...]
+                        Header `key:value` pairs
   -d, --form-data       Send payload as form encoded data instead of JSON (default: false)
   -n, --no-save         Do not save results to log file (default: false)
+  -t TIMEOUT, --timeout TIMEOUT
+                        Connection timeout of the request in seconds (default: 5)
 ```
 
 ### Mockbin Example
@@ -75,6 +80,28 @@ Use the returned bin id from before.
 Check the bin log from.
 https://mockbin.org/bin/9e95289e-d048-4515-9a61-07f2c74810f5/log
 
+### Set Auth and Headers
+
+Header key, value pairs can be set with the `-H` or `-header` flag.
+
+Key value pairs should be separated with either a `:` or `=`.
+
+```
+csv2http my_file.csv httpbin.org/post -H user-agent:csv2http-cli x-custom-header=foobar
+```
+
+To provide basic auth pass a username and password with `-a` or `--auth`.
+
+If the password is omitted you will be prompted to provide it.
+
+```
+--auth my_username:my_password
+```
+
+```
+--auth my_username
+```
+
 ## Roadmap
 
 - [x] As Library - Alpha

csv2http/_version.py

@@ -1,2 +1,2 @@
 """_version"""
-__version__ = "0.0.2a1"
+__version__ = "0.0.3a"

csv2http/cli.py

@@ -4,15 +4,22 @@
 """
 import argparse
 import pathlib
-from typing import Literal, NamedTuple, Union
+import re
+from typing import Literal, NamedTuple, Optional, Union
 
-from httpx import URL
+from httpx import URL, Headers
 
 from csv2http._version import __version__
 
 SUPPORTED_METHODS = ["POST", "PATCH", "PUT"]
-
 CONCURRENCY_DEFAULT = 25
+TIMEOUT_DEFAULT = 5
+
+_SPLIT_REGEX = r"[:=]"
+
+
+def _get_input(prompt: str):
+    return input(prompt)
 
 
 def _normalize_url(value: Union[str, URL]) -> URL:
@@ -23,20 +30,22 @@ def _normalize_url(value: Union[str, URL]) -> URL:
     return url
 
 
-class Args(NamedTuple):
-    """Expected user Args."""
+def _resolve_auth(value: str) -> Union[tuple[str, str], tuple[None, None]]:
+    """
+    Parse username & password. Prompt for password if not provided.
+    """
+    username, *extras = re.split(_SPLIT_REGEX, value, maxsplit=1)
+    password = " ".join(extras) if extras else _get_input("password:")
+    return username, password
 
-    file: pathlib.Path
-    url: Union[URL, str]
-    concurrency: int
-    method: Literal["POST", "PATCH", "PUT"]
-    form_data: bool = False
-    save_log: bool = True
-    # verbose: bool = True
 
+def _parse_header(value: str) -> tuple[str, str]:
+    """Splits string on `:` or `=` and returns a tuple of key, value."""
+    key, value = re.split(_SPLIT_REGEX, value, maxsplit=1)
+    return key, value
 
-def get_args() -> Args:
-    """Get user args from the command line."""
+
+def _bootstrap_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(
         description=f"HTTP request for every row of a CSV file - v{__version__}"
     )
@@ -59,6 +68,16 @@ def get_args() -> Args:
         default="POST",
         choices=SUPPORTED_METHODS,
     )
+    parser.add_argument(
+        "-a",
+        "--auth",
+        help="Basic Authentication enter <USERNAME>:<PASSWORD>."
+        " If password is blank you will be prompted for input",
+        type=_resolve_auth,
+    )
+    parser.add_argument(
+        "-H", "--header", help="Header `key:value` pairs", nargs="*", type=_parse_header
+    )
     parser.add_argument(
         "-d",
         "--form-data",
@@ -71,18 +90,53 @@ def get_args() -> Args:
         help="Do not save results to log file (default: false)",
         action="store_true",
     )
+    parser.add_argument(
+        "-t",
+        "--timeout",
+        help=f"Connection timeout of each request in seconds (default: {TIMEOUT_DEFAULT})",
+        default=TIMEOUT_DEFAULT,
+        type=int,
+    )
     # parser.add_argument(
-    #     "-v", "--verbose", help="verbose stdout logging", default=False, type=bool
+    #     "-v",
+    #     "--verbose",
+    #     help="verbose stdout logging",
+    #     action="store_true",
     # )
+    return parser
+
+
+class Args(NamedTuple):
+    """Expected user Args."""
 
-    args = parser.parse_args()
+    file: pathlib.Path
+    url: Union[URL, str]
+    concurrency: int
+    method: Literal["POST", "PATCH", "PUT"]
+    auth: Optional[tuple[str, str]] = None
+    headers: Optional[Headers] = None
+    form_data: bool = False
+    save_log: bool = True
+    timeout: int = TIMEOUT_DEFAULT
+    # verbose: bool = False
+
+
+_PARSER = _bootstrap_parser()
+
+
+def get_args() -> Args:
+    """Get user args from the command line."""
+    args = _PARSER.parse_args()
     return Args(
         args.file,
         args.url,
         args.concurrency,
         args.method,
+        args.auth,
+        Headers(args.header),
         args.form_data,
-        not args.no_save
+        not args.no_save,
+        args.timeout,
         # args.verbose,
     )
 

csv2http/core.py

@@ -92,7 +92,9 @@ async def execute(args: cli.Args) -> int:
     content = "data" if args.form_data else "json"
     log_file = _add_timestamp_and_suffix(file_input, "log")
 
-    async with httpx.AsyncClient() as client_session:
+    async with httpx.AsyncClient(
+        auth=args.auth, headers=args.headers, timeout=args.timeout
+    ) as client_session:
 
         print(f" {args.method} {args.url}")
 

tests/test_cli.py

@@ -1,3 +1,6 @@
+import base64
+import pathlib
+
 import pytest
 from httpx import URL
 
@@ -16,5 +19,98 @@ def test_normalize_ulr(url, expected):
     assert URL(expected) == cli._normalize_url(url)
 
 
+@pytest.mark.parametrize(
+    "auth_input,expected",
+    [
+        ("foo:bar", ("foo", "bar")),
+        ("foo=bar", ("foo", "bar")),
+        ("fizz", ("fizz", "fake_input")),
+        ("nopassword:", ("nopassword", "")),
+        ("nopassword=", ("nopassword", "")),
+        # also not valid
+        ("", ("", "fake_input")),
+    ],
+)
+def test_resolve_auth(monkeypatch, auth_input, expected):
+    monkeypatch.setattr(cli, "_get_input", lambda x: "fake_input", raising=True)
+
+    assert expected == cli._resolve_auth(auth_input)
+
+
+@pytest.mark.parametrize(
+    "headers_input,expected",
+    [
+        ("foo:bar", ("foo", "bar")),
+        ("foo=bar", ("foo", "bar")),
+        (
+            "Authorization=Bearer MY.JWT.TOKEN",
+            ("Authorization", "Bearer MY.JWT.TOKEN"),
+        ),
+        (
+            # ensure base64 padding is not removed
+            f"foo={base64.standard_b64encode(b'bar64').decode('utf-8')}",
+            ("foo", base64.standard_b64encode(b"bar64").decode("utf-8")),
+        ),
+    ],
+)
+def test_pase_header(headers_input, expected):
+    assert expected == cli._parse_header(headers_input)
+
+
+@pytest.mark.parametrize(
+    "args,expected",
+    [
+        # minimal example
+        (
+            ["myfile.csv", "example.com"],
+            {
+                "auth": None,
+                "concurrency": cli.CONCURRENCY_DEFAULT,
+                "file": pathlib.Path("myfile.csv"),
+                "form_data": False,
+                "header": None,
+                "method": "POST",
+                "no_save": False,
+                "url": URL("https://example.com"),
+                "timeout": cli.TIMEOUT_DEFAULT,
+            },
+        ),
+        (
+            [
+                "myfile.csv",
+                "example.com",
+                "-H",
+                "x-foo:bar",
+                "fizz=buzz",
+                "-c",
+                "23",
+                "--auth",
+                "bigetti:password",
+                "--timeout",
+                "3",
+            ],
+            {
+                "auth": ("bigetti", "password"),
+                "concurrency": 23,
+                "file": pathlib.Path("myfile.csv"),
+                "form_data": False,
+                "header": [
+                    ("x-foo", "bar"),
+                    ("fizz", "buzz"),
+                ],
+                "method": "POST",
+                "no_save": False,
+                "url": URL("https://example.com"),
+                "timeout": 3,
+            },
+        ),
+    ],
+)
+def test_get_args(args, expected):
+    parser = cli._bootstrap_parser()
+
+    assert expected == vars(parser.parse_args(args))
+
+
 if __name__ == "__main__":
     pytest.main(["-vv"])

tests/test_csv2http.py

@@ -2,4 +2,4 @@
 
 
 def test_version():
-    assert __version__ == "0.0.2a1"
+    assert __version__ == "0.0.3a"

tests/test_utils.py

@@ -1,3 +1,5 @@
+import ast
+
 import httpx
 import pytest
 from respx import MockResponse
@@ -19,9 +21,15 @@
 )
 def test_summarize_responses(responses):
     summary = utils.summarize_responses(responses)
-    print(summary)
     assert summary
 
+    _, status_codes = summary.split(" - ")
+    status_codes_dict = ast.literal_eval(status_codes)
+    print(status_codes_dict)
+
+    for r in responses:
+        assert getattr(r, "status_code", r.__class__.__name__) in status_codes_dict
+
 
 if __name__ == "__main__":
     pytest.main(["-vv"])