Add support for filtering in loaders

docs/changelog.rst

@@ -6,6 +6,11 @@ Changelog
 `Unreleased`_
 -------------
 
+Added
+~~~~~
+
+- Support for filtering in loaders `#75`_
+
 `0.7.1`_ - 2019-09-27
 ---------------------
 
@@ -142,6 +147,7 @@ Fixed
 .. _0.3.0: https://github.com/kiwicom/schemathesis/compare/v0.2.0...v0.3.0
 .. _0.2.0: https://github.com/kiwicom/schemathesis/compare/v0.1.0...v0.2.0
 
+.. _#75: https://github.com/kiwicom/schemathesis/issues/75
 .. _#69: https://github.com/kiwicom/schemathesis/issues/69
 .. _#58: https://github.com/kiwicom/schemathesis/issues/58
 .. _#55: https://github.com/kiwicom/schemathesis/issues/55

src/schemathesis/loaders.py

@@ -1,42 +1,47 @@
-from typing import IO, Any, Dict, Union
+from typing import IO, Any, Dict, Optional, Union
 from urllib.request import urlopen
 
 import yaml
 
 from .lazy import LazySchema
 from .schemas import BaseSchema, OpenApi30, SwaggerV20
-from .types import PathLike
+from .types import Filter, PathLike
 from .utils import deprecated
 
 
-def from_path(path: PathLike) -> BaseSchema:
+def from_path(path: PathLike, method: Optional[Filter] = None, endpoint: Optional[Filter] = None) -> BaseSchema:
     """Load a file from OS path and parse to schema instance.."""
     with open(path) as fd:
-        return from_file(fd)
+        return from_file(fd, method=method, endpoint=endpoint)
 
 
-def from_uri(uri: str) -> BaseSchema:
+def from_uri(uri: str, method: Optional[Filter] = None, endpoint: Optional[Filter] = None) -> BaseSchema:
     """Load a remote resource and parse to schema instance."""
     response = urlopen(uri)
     data = response.read()
-    return from_file(data)
+    return from_file(data, method=method, endpoint=endpoint)
 
 
-def from_file(file: Union[IO[str], str]) -> BaseSchema:
+def from_file(
+    file: Union[IO[str], str], method: Optional[Filter] = None, endpoint: Optional[Filter] = None
+) -> BaseSchema:
     """Load a file content and parse to schema instance.
 
     `file` could be a file descriptor, string or bytes.
     """
     raw = yaml.safe_load(file)
-    return from_dict(raw)
+    return from_dict(raw, method=method, endpoint=endpoint)
 
 
-def from_dict(raw_schema: Dict[str, Any]) -> BaseSchema:
+def from_dict(
+    raw_schema: Dict[str, Any], method: Optional[Filter] = None, endpoint: Optional[Filter] = None
+) -> BaseSchema:
     """Get a proper abstraction for the given raw schema."""
     if "swagger" in raw_schema:
-        return SwaggerV20(raw_schema)
+        return SwaggerV20(raw_schema, method=method, endpoint=endpoint)
+
     if "openapi" in raw_schema:
-        return OpenApi30(raw_schema)
+        return OpenApi30(raw_schema, method=method, endpoint=endpoint)
     raise ValueError("Unsupported schema type")
 
 

src/schemathesis/schemas.py

@@ -19,6 +19,7 @@
 from .filters import should_skip_endpoint, should_skip_method
 from .models import Endpoint
 from .types import Filter
+from .utils import NOT_SET
 
 
 @attr.s(slots=True)
@@ -44,8 +45,12 @@ def get_all_tests(
         for endpoint in self.get_all_endpoints():
             yield endpoint, create_test(endpoint, func, settings)
 
-    def parametrize(self, method: Optional[Filter] = None, endpoint: Optional[Filter] = None) -> Callable:
+    def parametrize(self, method: Optional[Filter] = NOT_SET, endpoint: Optional[Filter] = NOT_SET) -> Callable:
         """Mark a test function as a parametrized one."""
+        if method is NOT_SET:
+            method = self.method
+        if endpoint is NOT_SET:
+            endpoint = self.endpoint
 
         def wrapper(func: Callable) -> Callable:
             func._schemathesis_test = self.__class__(self.raw_schema, method=method, endpoint=endpoint)  # type: ignore

src/schemathesis/types.py

@@ -12,4 +12,4 @@
 FormData = Dict[str, Any]
 
 # A filter for endpoint / method
-Filter = Union[str, List[str], Tuple[str], Set[str]]
+Filter = Union[str, List[str], Tuple[str], Set[str], object]

src/schemathesis/utils.py

@@ -4,6 +4,8 @@
 
 from .types import Filter
 
+NOT_SET = object()
+
 
 def deprecated(func: Callable, message: str) -> Callable:
     """Emit a warning if the given function is used."""

test/conftest.py

@@ -48,7 +48,7 @@ def maker(**kwargs):
 
 @pytest.fixture()
 def testdir(testdir):
-    def maker(content, **kwargs):
+    def maker(content, method=None, endpoint=None, **kwargs):
         schema = make_schema(**kwargs)
         preparation = dedent(
             """
@@ -57,9 +57,9 @@ def maker(content, **kwargs):
         from test.utils import *
         from hypothesis import settings
         raw_schema = {schema}
-        schema = schemathesis.from_dict(raw_schema)
+        schema = schemathesis.from_dict(raw_schema, method={method}, endpoint={endpoint})
         """.format(
-                schema=schema
+                schema=schema, method=repr(method), endpoint=repr(endpoint)
             )
         )
         testdir.makepyfile(preparation, content)

test/test_filters.py

@@ -49,3 +49,39 @@ def test_(request, case):
     result.stdout.re_match_lines(
         [r"test_method_filter.py::test_\[GET:/v1/foo\] PASSED", r"test_method_filter.py::test_\[GET:/v1/users\] PASSED"]
     )
+
+
+def test_loader_filter(testdir):
+    testdir.make_test(
+        """
+@schema.parametrize()
+def test_(request, case):
+    request.config.HYPOTHESIS_CASES += 1
+    assert case.path == "/v1/foo"
+    assert case.method == "POST"
+""",
+        paths={"/foo": {"post": {"parameters": [integer(name="id", required=True)]}}},
+        method="POST",
+        endpoint="/v1/foo",
+    )
+    result = testdir.runpytest("-v", "-s")
+    result.assert_outcomes(passed=1)
+    result.stdout.re_match_lines([r"Hypothesis calls: 1"])
+
+
+def test_override_filter(testdir):
+    testdir.make_test(
+        """
+@schema.parametrize(method=None, endpoint="/v1/users")
+def test_(request, case):
+    request.config.HYPOTHESIS_CASES += 1
+    assert case.path == "/v1/users"
+    assert case.method == "GET"
+""",
+        paths={"/foo": {"post": {"parameters": [integer(name="id", required=True)]}}},
+        method="POST",
+        endpoint="/v1/foo",
+    )
+    result = testdir.runpytest("-v", "-s")
+    result.assert_outcomes(passed=1)
+    result.stdout.re_match_lines([r"Hypothesis calls: 1"])