Add authorization & custom headers to runner & CLI. Resolves #96

README.rst

@@ -139,6 +139,9 @@ The ``schemathesis`` command can be used to perform Schemathesis test cases:
 
     schemathesis run https://example.com/api/swagger.json
 
+If your application requires authorization then you can use ``--auth`` option for Basic Auth and ``--header`` to specify
+custom headers to be sent with each request.
+
 For the full list of options, run:
 
 .. code:: bash

src/schemathesis/commands.py

@@ -1,4 +1,5 @@
-from typing import Iterable
+from contextlib import contextmanager
+from typing import Dict, Generator, Iterable, Optional, Tuple
 from urllib.parse import urlparse
 
 import click
@@ -16,6 +17,33 @@ def main() -> None:
     """Command line tool for testing your web application built with Open API / Swagger specifications."""
 
 
+def validate_auth(
+    ctx: click.core.Context, param: click.core.Option, raw_value: Optional[str]
+) -> Optional[Tuple[str, str]]:
+    if raw_value is not None:
+        with reraise_format_error(raw_value):
+            user, password = tuple(raw_value.split(":"))
+        return user, password
+    return None
+
+
+def validate_headers(ctx: click.core.Context, param: click.core.Option, raw_value: Tuple[str, ...]) -> Dict[str, str]:
+    headers = {}
+    for header in raw_value:
+        with reraise_format_error(header):
+            key, value = header.split(":")
+        headers[key] = value.lstrip()
+    return headers
+
+
+@contextmanager
+def reraise_format_error(raw_value: str) -> Generator:
+    try:
+        yield
+    except ValueError:
+        raise click.BadParameter(f"Should be in KEY:VALUE format. Got: {raw_value}")
+
+
 @main.command(short_help="Perform schemathesis test.")
 @click.argument("schema", type=str)
 @click.option(
@@ -26,7 +54,25 @@ def main() -> None:
     type=click.Choice(DEFAULT_CHECKS_NAMES),
     default=DEFAULT_CHECKS_NAMES,
 )
-def run(schema: str, checks: Iterable[str] = DEFAULT_CHECKS_NAMES) -> None:
+@click.option(  # type: ignore
+    "--auth",
+    "-a",
+    help="Server user and password. Example: USER:PASSWORD",
+    type=str,
+    callback=validate_auth,  # type: ignore
+)
+@click.option(  # type: ignore
+    "--header",
+    "-H",
+    "headers",
+    help=r"Custom header in a that will be used in all requests to the server. Example: Authorization: Bearer\ 123",
+    multiple=True,
+    type=str,
+    callback=validate_headers,  # type: ignore
+)
+def run(
+    schema: str, auth: Optional[Tuple[str, str]], headers: Dict[str, str], checks: Iterable[str] = DEFAULT_CHECKS_NAMES
+) -> None:
     """Perform schemathesis test against an API specified by SCHEMA.
 
     SCHEMA must be a valid URL pointing to an Open API / Swagger specification.
@@ -38,6 +84,6 @@ def run(schema: str, checks: Iterable[str] = DEFAULT_CHECKS_NAMES) -> None:
 
     click.echo("Running schemathesis test cases ...")
 
-    runner.execute(schema, checks=selected_checks)
+    runner.execute(schema, checks=selected_checks, auth=auth, headers=headers)
 
     click.echo("Done.")

src/schemathesis/runner.py

@@ -1,13 +1,16 @@
 from contextlib import suppress
-from typing import Callable, Iterable
+from typing import Any, Callable, Dict, Iterable, Optional, Tuple, Union
 from urllib.parse import urlsplit, urlunsplit
 
 import requests
+from requests.auth import AuthBase
 
 from .loaders import from_uri
 from .models import Case
 from .schemas import BaseSchema
 
+Auth = Union[Tuple[str, str], AuthBase]
+
 
 def not_a_server_error(response: requests.Response) -> None:
     """A check to verify that the response is not a server-side error."""
@@ -17,17 +20,33 @@ def not_a_server_error(response: requests.Response) -> None:
 DEFAULT_CHECKS = (not_a_server_error,)
 
 
-def _execute_all_tests(schema: BaseSchema, base_url: str, checks: Iterable[Callable]) -> None:
+def _execute_all_tests(
+    schema: BaseSchema,
+    base_url: str,
+    checks: Iterable[Callable],
+    auth: Optional[Auth] = None,
+    headers: Optional[Dict[str, Any]] = None,
+) -> None:
     with requests.Session() as session, suppress(AssertionError):
+        if auth is not None:
+            session.auth = auth
+        if headers is not None:
+            session.headers.update(**headers)
         for _, test in schema.get_all_tests(single_test):
             test(session, base_url, checks)
 
 
-def execute(schema_uri: str, base_url: str = "", checks: Iterable[Callable] = DEFAULT_CHECKS) -> None:
+def execute(
+    schema_uri: str,
+    base_url: str = "",
+    checks: Iterable[Callable] = DEFAULT_CHECKS,
+    auth: Optional[Auth] = None,
+    headers: Optional[Dict[str, Any]] = None,
+) -> None:
     """Generate and run test cases against the given API definition."""
     schema = from_uri(schema_uri)
     base_url = base_url or get_base_url(schema_uri)
-    _execute_all_tests(schema, base_url, checks)
+    _execute_all_tests(schema, base_url, checks, auth, headers)
 
 
 def get_base_url(uri: str) -> str:

test/test_commands.py

@@ -35,6 +35,14 @@ def test_commands_version(schemathesis_cmd):
     (
         (("run",), 'Error: Missing argument "SCHEMA".'),
         (("run", "not-url"), "Error: Invalid SCHEMA, must be a valid URL."),
+        (
+            ("run", "http://127.0.0.1", "--auth=123"),
+            'Error: Invalid value for "--auth" / "-a": Should be in KEY:VALUE format. Got: 123',
+        ),
+        (
+            ("run", "http://127.0.0.1", "--header=123"),
+            'Error: Invalid value for "--header" / "-H": Should be in KEY:VALUE format. Got: 123',
+        ),
     ),
 )
 def test_commands_run_errors(schemathesis_cmd, args, error):
@@ -59,17 +67,40 @@ def test_commands_run_help(schemathesis_cmd):
         "Options:",
         "  -c, --checks [not_a_server_error]",
         "                                  List of checks to run.",
+        "  -a, --auth TEXT                 Server user and password. Example:",
+        "                                  USER:PASSWORD",
+        "  -H, --header TEXT               Custom header in a that will be used in all",
+        r"                                  requests to the server. Example:",
+        r"                                  Authorization: Bearer\ 123",
         "  -h, --help                      Show this message and exit.",
     ]
 
 
-def test_commands_run_schema_uri(mocker):
+SCHEMA_URI = "https://example.com/swagger.json"
+
+
+@pytest.mark.parametrize(
+    "args, expected",
+    (
+        ([SCHEMA_URI], {"checks": runner.DEFAULT_CHECKS, "auth": None, "headers": {}}),
+        ([SCHEMA_URI, "--auth=test:test"], {"checks": runner.DEFAULT_CHECKS, "auth": ("test", "test"), "headers": {}}),
+        (
+            [SCHEMA_URI, "--header=Authorization:Bearer 123"],
+            {"checks": runner.DEFAULT_CHECKS, "auth": None, "headers": {"Authorization": "Bearer 123"}},
+        ),
+        (
+            [SCHEMA_URI, "--header=Authorization:  Bearer 123 "],
+            {"checks": runner.DEFAULT_CHECKS, "auth": None, "headers": {"Authorization": "Bearer 123 "}},
+        ),
+    ),
+)
+def test_commands_run(mocker, args, expected):
     m_execute = mocker.patch("schemathesis.runner.execute")
     cli = CliRunner()
 
     schema_uri = "https://example.com/swagger.json"
-    result = cli.invoke(commands.run, [schema_uri])
+    result = cli.invoke(commands.run, args)
 
     assert result.exit_code == 0
-    m_execute.assert_called_once_with(schema_uri, checks=runner.DEFAULT_CHECKS)
+    m_execute.assert_called_once_with(schema_uri, **expected)
     assert result.stdout.split("\n")[:-1] == ["Running schemathesis test cases ...", "Done."]

test/test_runner.py

@@ -58,16 +58,29 @@ def server(app, aiohttp_unused_port):
     yield {"port": port}
 
 
-def assert_request(app, idx, method, path):
-    assert app["saved_requests"][idx].method == method
-    assert app["saved_requests"][idx].path == path
+def assert_request(app, idx, method, path, headers=None):
+    request = app["saved_requests"][idx]
+    assert request.method == method
+    assert request.path == path
+    if headers:
+        for key, value in headers.items():
+            assert request.headers.get(key) == value
 
 
 def test_execute(server, app):
-    execute(f"http://127.0.0.1:{server['port']}/swagger.yaml")
+    headers = {"Authorization": "Bearer 123"}
+    execute(f"http://127.0.0.1:{server['port']}/swagger.yaml", headers=headers)
     assert len(app["saved_requests"]) == 2
-    assert_request(app, 0, "GET", "/v1/pets")
-    assert_request(app, 1, "GET", "/v1/users")
+    assert_request(app, 0, "GET", "/v1/pets", headers)
+    assert_request(app, 1, "GET", "/v1/users", headers)
+
+
+def test_auth(server, app):
+    execute(f"http://127.0.0.1:{server['port']}/swagger.yaml", auth=("test", "test"))
+    assert len(app["saved_requests"]) == 2
+    headers = {"Authorization": "Basic dGVzdDp0ZXN0"}
+    assert_request(app, 0, "GET", "/v1/pets", headers)
+    assert_request(app, 1, "GET", "/v1/users", headers)
 
 
 def test_server_error(server, app):