STDIO spec support

Author: p1c2u

openapi_spec_validator/__init__.py

@@ -2,7 +2,7 @@
 from openapi_spec_validator.shortcuts import (
     validate_spec_factory, validate_spec_url_factory,
 )
-from openapi_spec_validator.handlers import UrlHandler
+from openapi_spec_validator.handlers import UrlHandler, FileObjectHandler
 from openapi_spec_validator.schemas import get_openapi_schema
 from openapi_spec_validator.factories import JSONSpecValidatorFactory
 from openapi_spec_validator.validators import SpecValidator
@@ -19,8 +19,10 @@
     'validate_v2_spec_url', 'validate_v3_spec_url', 'validate_spec_url',
 ]
 
+file_object_handler = FileObjectHandler()
+all_urls_handler = UrlHandler('http', 'https', 'file')
 default_handlers = {
-    '<all_urls>': UrlHandler('http', 'https', 'file'),
+    '<all_urls>': all_urls_handler,
     'http': UrlHandler('http'),
     'https': UrlHandler('https'),
     'file': UrlHandler('file'),

openapi_spec_validator/__main__.py

@@ -7,8 +7,10 @@
     import pathlib2 as pathlib
 import sys
 
-
-from openapi_spec_validator import validate_spec_url, validate_v2_spec_url
+from openapi_spec_validator import (
+    openapi_v2_spec_validator, openapi_v3_spec_validator, all_urls_handler,
+    file_object_handler,
+)
 from openapi_spec_validator.exceptions import ValidationError
 
 logger = logging.getLogger(__name__)
@@ -18,6 +20,19 @@
 )
 
 
+def read_from_stdin(filename):
+    return file_object_handler(sys.stdin), ''
+
+
+def read_from_filename(filename):
+    if not os.path.isfile(filename):
+        raise SystemError("No such file {0}".format(filename))
+
+    filename = os.path.abspath(filename)
+    uri = pathlib.Path(filename).as_uri()
+    return all_urls_handler(uri), uri
+
+
 def main(args=None):
     parser = argparse.ArgumentParser()
     parser.add_argument('filename', help="Absolute or relative path to file")
@@ -29,21 +44,34 @@ def main(args=None):
         default='3.0.0'
     )
     args = parser.parse_args(args)
-    filename = args.filename
-    filename = os.path.abspath(filename)
+
+    # choose source
+    reader = read_from_filename
+    if args.filename == '-':
+        reader = read_from_stdin
+
+    # read source
+    try:
+        spec, spec_url = reader(args.filename)
+    except Exception as exc:
+        print(exc)
+        sys.exit(1)
+
     # choose the validator
-    if args.schema == '2.0':
-        validate_url = validate_v2_spec_url
-    elif args.schema == '3.0.0':
-        validate_url = validate_spec_url
+    validators = {
+        '2.0': openapi_v2_spec_validator,
+        '3.0.0': openapi_v3_spec_validator,
+    }
+    validator = validators[args.schema]
+
     # validate
     try:
-        validate_url(pathlib.Path(filename).as_uri())
-    except ValidationError as e:
-        print(e)
+        validator.validate(spec, spec_url=spec_url)
+    except ValidationError as exc:
+        print(exc)
         sys.exit(1)
-    except Exception as e:
-        print(e)
+    except Exception as exc:
+        print(exc)
         sys.exit(2)
     else:
         print('OK')

openapi_spec_validator/handlers.py

@@ -8,19 +8,30 @@
 from openapi_spec_validator.loaders import ExtendedSafeLoader
 
 
-class UrlHandler:
-    """OpenAPI spec validator URL scheme handler."""
+class FileObjectHandler(object):
+    """OpenAPI spec validator file-like object handler."""
 
-    def __init__(self, *allowed_schemes, **options):
-        self.allowed_schemes = allowed_schemes
+    def __init__(self, **options):
         self.options = options
 
     @property
     def loader(self):
         return self.options.get('loader', ExtendedSafeLoader)
 
+    def __call__(self, f):
+        return load(f, self.loader)
+
+
+class UrlHandler(FileObjectHandler):
+    """OpenAPI spec validator URL scheme handler."""
+
+    def __init__(self, *allowed_schemes, **options):
+        super(UrlHandler, self).__init__(**options)
+        self.allowed_schemes = allowed_schemes
+
     def __call__(self, url, timeout=1):
         assert urlparse(url).scheme in self.allowed_schemes
 
-        with contextlib.closing(urlopen(url, timeout=timeout)) as fh:
-            return load(fh, self.loader)
+        f = urlopen(url, timeout=timeout)
+        with contextlib.closing(f) as fh:
+            return super(UrlHandler, self).__call__(fh)

tests/integration/test_main.py

@@ -1,4 +1,7 @@
+import mock
+
 import pytest
+from six import StringIO
 
 from openapi_spec_validator.__main__ import main
 
@@ -39,8 +42,32 @@ def test_validation_error():
         main(testargs)
 
 
+@mock.patch(
+    'openapi_spec_validator.__main__.openapi_v3_spec_validator.validate',
+    side_effect=Exception,
+)
+def test_unknown_error(m_validate):
+    """SystemExit on running with unknown error."""
+    testargs = ['--schema', '3.0.0',
+                './tests/integration/data/v2.0/petstore.yaml']
+    with pytest.raises(SystemExit):
+        main(testargs)
+
+
 def test_nonexisting_file():
     """Calling with non-existing file should sys.exit."""
     testargs = ['i_dont_exist.yaml']
     with pytest.raises(SystemExit):
         main(testargs)
+
+
+def test_schema_stdin():
+    """Test schema from STDIN"""
+    spes_path = './tests/integration/data/v3.0/petstore.yaml'
+    with open(spes_path, 'r') as spec_file:
+        spec_lines = spec_file.readlines()
+    spec_io = StringIO("".join(spec_lines))
+
+    testargs = ['-']
+    with mock.patch('openapi_spec_validator.__main__.sys.stdin', spec_io):
+        main(testargs)