diff --git a/core/common/swagger.py b/core/common/swagger.py index 75e6172a..d43579e4 100644 --- a/core/common/swagger.py +++ b/core/common/swagger.py @@ -1,7 +1,13 @@ """Swagger helpers for OCL-specific OpenAPI generation.""" +from copy import deepcopy + +from django.conf import settings from drf_yasg.generators import EndpointEnumerator, OpenAPISchemaGenerator +from core.fhir.base import FhirViewMixin, is_fhir_path +from core.fhir.swagger import FhirSwaggerSchemaGenerator, fhir_paths, fhir_swagger_url, on_fhir_host + class OCLSwaggerEndpointEnumerator(EndpointEnumerator): """Preserve literal dollar signs when deriving OpenAPI paths from Django URL regexes.""" @@ -18,3 +24,26 @@ class OCLSwaggerSchemaGenerator(OpenAPISchemaGenerator): """Use the OCL endpoint enumerator so Swagger paths match the published URLs.""" endpoint_enumerator_class = OCLSwaggerEndpointEnumerator + + def get_endpoints(self, request): + """Replace every FHIR view entry, including malformed paths from no-slash includes.""" + return {path: endpoint for path, endpoint in super().get_endpoints(request).items() + if not is_fhir_path(path) and not issubclass(endpoint[0], FhirViewMixin)} + + def get_paths(self, endpoints, components, request, public): + """Keep the other API operations intact, adding FHIR only when both share a host.""" + paths, prefix = super().get_paths(endpoints, components, request, public) + if not settings.FHIR_SUBDOMAIN: + paths.update(fhir_paths(components)) + return paths, prefix + + def get_schema(self, request=None, public=False): + """Select the host's schema and link the API documentation to the separate FHIR service.""" + if on_fhir_host(request): + return FhirSwaggerSchemaGenerator(self.info).get_schema(request, public) + schema = super().get_schema(request, public) + if settings.FHIR_SUBDOMAIN: + schema.info = deepcopy(schema.info) + link = fhir_swagger_url(request, self.url or settings.API_BASE_URL) + schema.info.description = schema.info.get('description', '') + f'\n\n[FHIR terminology Swagger]({link}).' + return schema diff --git a/core/fhir/swagger.py b/core/fhir/swagger.py new file mode 100644 index 00000000..282601a6 --- /dev/null +++ b/core/fhir/swagger.py @@ -0,0 +1,304 @@ +"""Curated Swagger 2.0 descriptions of the FHIR terminology reads and operations.""" +from copy import deepcopy +from urllib.parse import urlsplit, urlunsplit + +from django.conf import settings +from drf_yasg import openapi +from drf_yasg.generators import OpenAPISchemaGenerator + +from core.fhir.base import FHIR_JSON, FHIR_XML +from core.fhir.ops.expand import MAX_COUNT as EXPANSION_MAX_COUNT +from core.fhir.search import MAX_COUNT, PARAMETERS, STRING_FIELDS, STRING_LOOKUPS + +BASES = ('/fhir', '/orgs/{org}', '/users/{user}') +TAGS = ('CodeSystem', 'ValueSet', 'ConceptMap', 'Capabilities') +VERSION_SELECTION = ( + 'The default is the latest released version visible at this base; HEAD is used only when explicitly named. ' + 'An instance id at /fhir names a released version; at an owner base it is the repository mnemonic.') +SEARCH_DESCRIPTIONS = { + '_id': 'Resource id: an opaque release id at /fhir, or the repository mnemonic at an owner base.', + '_lastUpdated': 'FHIR date or dateTime, optionally prefixed by eq, ne, gt, lt, ge, le, sa or eb.', + 'url': 'Canonical URL of the resource.', + 'version': 'Business version token. Searches include released versions only, never HEAD.', + 'status': 'Publication status token: active or retired.', + 'identifier': 'Identifier token: value, system|value, |value or system|.', + 'system': 'Canonical URL of the code system (an alias of url).', + 'code': 'A code defined by the code system; code or system|code.', + 'language': 'Language token for the code system (urn:ietf:bcp:47).', + 'reference': 'Canonical code system URL referenced by the value set\'s stored default expansion.', + 'source-system': 'Canonical URL of a source system in a ConceptMap group.', + 'source-uri': 'Alias of source-system.', + 'target-system': 'Canonical URL of a target system in a ConceptMap group.', + 'target-uri': 'Alias of target-system.', +} + + +def on_fhir_host(request): + """Use the same host boundary for the schema and the FHIR middleware.""" + return bool(request and settings.FHIR_SUBDOMAIN and + request.get_host().startswith(settings.FHIR_SUBDOMAIN + '.')) + + +def fhir_swagger_url(request, api_url): + """The FHIR sibling of the API host, keeping the request's scheme and port.""" + url = urlsplit(request.build_absolute_uri('/') if request else api_url) + host = settings.FHIR_SUBDOMAIN + '.' + (url.netloc.partition('.')[2] or url.netloc) + return urlunsplit((url.scheme, host, '/swagger/', '', '')) + + +def query(name, description, type_=openapi.TYPE_STRING, **kwargs): + """A typed query parameter; complex FHIR values are described in the POST body instead.""" + return openapi.Parameter(name, openapi.IN_QUERY, type=type_, description=description, **kwargs) + + +NOT_WITH_CODING = ' Not read when coding or codeableConcept is given.' +FORMAT = query('_format', 'Response format; overrides Accept.', enum=['json', 'xml']) +CODE = query('code', 'Code to look up, validate or translate; alternatively supply coding.' + NOT_WITH_CODING) +SYSTEM = query('system', 'Canonical URL or OCL path of the code system the code is from.' + NOT_WITH_CODING) +VERSION = query('version', 'Business version of the code system; defaults to the latest released version.') +LOOKUP_SYSTEM = query('system', 'Canonical URL or OCL path of the code system the code is from. At an instance it ' + 'must name that code system. With coding at the type level, the coding\'s system ' + 'is used instead.') +URL = query('url', 'Canonical URL or OCL path of the target resource; may include |version.') +VALUE_SET_VERSION = query('valueSetVersion', 'Business version of the value set.') +LANGUAGE = query('displayLanguage', 'Preferred language for the returned display, when a name in it exists.') +DISPLAY = query('display', 'Display to validate against the concept\'s names, ignoring case.' + NOT_WITH_CODING) + + +def coding(replaces, applies=''): + """The coding input of one operation: a system|code token here, a full Coding in a POSTed body.""" + return query('coding', f'A system|code token. It replaces {replaces}.{applies} A GET token carries no version ' + 'or display: send those inside a POSTed valueCoding.') + + +OPERATION_PARAMETERS = { + ('CodeSystem', '$lookup'): [ + LOOKUP_SYSTEM, VERSION, CODE, + coding('code and system', ' version still selects the code system version; at an instance, a separate ' + 'system must still name it.'), + LANGUAGE], + ('CodeSystem', '$validate-code'): [ + URL, VERSION, CODE, + coding('code and display', ' url and version still name the code system validated against.'), + DISPLAY, LANGUAGE], + ('ValueSet', '$validate-code'): [ + URL, VALUE_SET_VERSION, SYSTEM, + query('systemVersion', 'Business version of the input code system.' + NOT_WITH_CODING), + CODE, coding('code, system, systemVersion and display'), DISPLAY, LANGUAGE], + ('ValueSet', '$expand'): [ + URL, VALUE_SET_VERSION, + query('filter', 'Each word must prefix a word in the code or one of its names, ignoring case.'), LANGUAGE, + query('includeDesignations', 'Include every name as a designation.', openapi.TYPE_BOOLEAN), + query('activeOnly', 'Exclude retired concepts.', openapi.TYPE_BOOLEAN), + query('offset', 'Number of matching members to skip.', openapi.TYPE_INTEGER, minimum=0, default=0), + query('count', 'Page size, capped at 1,000; zero returns only the total.', openapi.TYPE_INTEGER, + minimum=0, maximum=EXPANSION_MAX_COUNT, default=EXPANSION_MAX_COUNT), + query('includeDefinition', 'Only false is supported; definitions are not included.', openapi.TYPE_BOOLEAN, + enum=[False]), + query('excludePostCoordinated', 'Only false is supported; stored expansions do not record this distinction.', + openapi.TYPE_BOOLEAN, enum=[False]), + query('excludeNested', 'Either boolean is accepted; stored expansions are flat.', openapi.TYPE_BOOLEAN), + query('excludeNotForUI', 'Either boolean is accepted; stored expansions have no abstract codes.', + openapi.TYPE_BOOLEAN)], + ('ConceptMap', '$translate'): [ + URL, query('conceptMapVersion', 'Business version of the map; without url, selects the source\'s map version.'), + CODE, SYSTEM, + query('version', 'Input code system version; omitted means no mapping-side version filter. ' + 'Without url in forward translation, also selects the map version unless conceptMapVersion ' + 'is given.' + NOT_WITH_CODING), + coding('code, system and version'), + query('source', 'Input value set scope for forward translation; ignored when reverse=true.'), + query('target', 'Input value set scope for reverse translation; ignored in forward translation.'), + query('targetsystem', 'Restrict matches to this code system; in reverse, the system of the codes sought.'), + query('reverse', 'Translate from the target side of the mappings.', openapi.TYPE_BOOLEAN, default=False)], +} +OPERATION_DESCRIPTIONS = { + ('CodeSystem', '$lookup'): ( + 'Look up a code', 'Returns the code\'s display, names and inactive flag, including retired concepts. ' + 'A code and, at type level, its system are required (or supply coding).', (400, 404, 422)), + ('CodeSystem', '$validate-code'): ( + 'Validate a code in a code system', + 'Checks the code and optional display in url (or coding.system). A code and a target system are required. ' + 'An unknown code or system returns Parameters with result=false; retired codes are valid with a warning.', + (400,)), + ('ValueSet', '$validate-code'): ( + 'Validate a code in a value set', + 'Checks membership in the stored default expansion and an optional display. A code and its system are ' + 'required (or supply coding). Without url at type level, checks the code system alone. ' + 'A failed validation returns Parameters with result=false.', (400, 422)), + ('ValueSet', '$expand'): ( + 'Read a stored value set expansion', + 'Returns a page of the stored default expansion; never computes or saves an expansion. ' + 'At type level url is required. count is capped at 1,000. Missing, unfinished or incomplete stored ' + 'expansions return 422. Inline value sets, contexts, dates, selected designations and code system version ' + 'overrides are not supported.', (400, 422)), + ('ConceptMap', '$translate'): ( + 'Translate a code', + 'Returns live mappings from the named ConceptMap. Without url, uses the input system\'s own mappings ' + 'in forward translation, or targetsystem\'s mappings in reverse; without targetsystem, reverse has no match. ' + 'A code and system are required (or supply coding). Supports reverse translation and a target system filter. ' + 'No match returns Parameters with result=false. A POSTed codeableConcept uses its first coding.', + (400, 404, 422)), +} + + +def resource_schema(*resource_types): + """Identify the FHIR response type without duplicating the complete FHIR resource definitions.""" + return openapi.Schema(type=openapi.TYPE_OBJECT, required=['resourceType'], properties={ + 'resourceType': openapi.Schema(type=openapi.TYPE_STRING, enum=list(resource_types)), + }) + + +def parameters_schema(components): + """One shared Parameters body, including the structured alternatives to query-string coding.""" + definitions = components.with_scope(openapi.SCHEMA_DEFINITIONS) + if not definitions.has('FhirParameters'): + schema = resource_schema('Parameters') + schema.properties['parameter'] = openapi.Schema(type=openapi.TYPE_ARRAY, items=openapi.Schema( + type=openapi.TYPE_OBJECT, required=['name'], properties={ + 'name': openapi.Schema(type=openapi.TYPE_STRING), + 'valueString': openapi.Schema(type=openapi.TYPE_STRING), + 'valueCode': openapi.Schema(type=openapi.TYPE_STRING), + 'valueUri': openapi.Schema(type=openapi.TYPE_STRING), + 'valueBoolean': openapi.Schema(type=openapi.TYPE_BOOLEAN), + 'valueInteger': openapi.Schema(type=openapi.TYPE_INTEGER), + 'valueCoding': openapi.Schema(type=openapi.TYPE_OBJECT, properties={ + key: openapi.Schema(type=openapi.TYPE_STRING) for key in ('system', 'code', 'version', 'display')}), + 'valueCodeableConcept': openapi.Schema(type=openapi.TYPE_OBJECT, properties={ + 'coding': openapi.Schema(type=openapi.TYPE_ARRAY, items=openapi.Schema(type=openapi.TYPE_OBJECT))}), + })) + definitions.set('FhirParameters', schema) + return openapi.SchemaRef(definitions, 'FhirParameters') + + +def operation(summary, description, parameters, response_type, errors=(400, 422)): + """A read with FHIR response media types and OperationOutcome errors; a token shows private content.""" + responses = {'200': openapi.Response(response_type, schema=resource_schema(response_type))} + errors_by_status = { + 400: 'Invalid or unsupported input.', 404: 'Resource or code not found or not visible at this base.', + 422: 'The request cannot be processed or exceeds the operation\'s processing bound.', + } + responses.update({str(code): openapi.Response(errors_by_status[code], schema=resource_schema('OperationOutcome')) + for code in errors}) + return openapi.Operation( + operation_id='', summary=summary, description=description, parameters=[*parameters, FORMAT], + responses=openapi.Responses(responses), produces=[FHIR_JSON, FHIR_XML]) + + +def search_operation(resource_type): + """Take the supported names and modifiers from the search parser itself.""" + parameters = [query(name, SEARCH_DESCRIPTIONS[name]) for name in PARAMETERS[resource_type]] + modifiers = {'': 'Starts with, ignoring case and accents.', 'exact': 'Exact, including case and accents.', + 'contains': 'Contains, ignoring case and accents.'} + parameters.extend(query(f'{name}:{modifier}' if modifier else name, f'{name}: {modifiers[modifier]}') + for name in STRING_FIELDS for modifier in STRING_LOOKUPS) + parameters.extend([ + query('_count', 'Page size, capped at 100; zero returns only the total.', openapi.TYPE_INTEGER, + minimum=0, maximum=MAX_COUNT, default=MAX_COUNT), + query('page', 'Page number, starting at 1.', openapi.TYPE_INTEGER, minimum=1, default=1), + ]) + return operation(f'Search {resource_type} resources', + 'Returns a searchset Bundle of visible released versions, never HEAD. Entries are summaries ' + 'tagged SUBSETTED. Commas separate alternatives; repeated parameters must all match. ' + 'Unsupported parameters or modifiers return 400.', parameters, 'Bundle') + + +def resource_operations(resource_type, base): + """The type's search, read and supported terminology operations, once per canonical suffix.""" + read_parameters = [] if base == '/fhir' else [query('version', 'Released business version; never HEAD.')] + read_description = ( + 'The opaque id selects one released version.' if base == '/fhir' else + 'The id is the repository mnemonic. Returns the latest visible release unless version is named.') + paths = { + f'/{resource_type}': {'get': search_operation(resource_type)}, + f'/{resource_type}/{{id}}': {'get': operation( + f'Read a {resource_type}', read_description + ' HEAD is never returned. A cut read is tagged SUBSETTED.', + read_parameters, resource_type, (400, 404, 422))}, + } + for (resource, name), (summary, description, errors) in OPERATION_DESCRIPTIONS.items(): + if resource != resource_type: + continue + for instance in (False, True) if resource != 'ConceptMap' else (False,): + suffix = f'/{resource}/' + ('{id}/' if instance else '') + name + operation_errors = tuple(sorted(set(errors) | ({404} if instance else set()))) + documented = operation(summary, description + ' ' + VERSION_SELECTION, + OPERATION_PARAMETERS[(resource, name)], + 'ValueSet' if name == '$expand' else 'Parameters', operation_errors) + paths[suffix] = {'get': documented, 'post': deepcopy(documented)} + return paths + + +def capability_operations(): + """The metadata variants share a route; $versions only supports GET.""" + metadata = operation('Read service capabilities', + 'Returns CapabilityStatement, or TerminologyCapabilities for mode=terminology.', + [query('mode', 'Use terminology to list the code systems and versions visible at this base.', + enum=['terminology'])], 'CapabilityStatement') + metadata.responses['200'] = openapi.Response( + 'CapabilityStatement; TerminologyCapabilities when mode=terminology.', + schema=resource_schema('CapabilityStatement', 'TerminologyCapabilities')) + return { + '/metadata': {'get': metadata}, + '/$versions': {'get': operation('List supported FHIR versions', + 'Returns the supported FHIR versions and the default (4.0 for R4).', + [], 'Parameters', (400,))}, + } + + +def path_parameters(path): + """Document the FHIR spelling of the route variables, independent of Django's lookup kwarg names.""" + descriptions = { + 'org': 'Organization identifier.', 'user': 'User identifier.', + 'id': 'Opaque release id at /fhir; repository mnemonic at an owner base.', + } + return [openapi.Parameter(name, openapi.IN_PATH, type=openapi.TYPE_STRING, required=True, description=description) + for name, description in descriptions.items() if '{' + name + '}' in path] + + +def fhir_paths(components): + """One source of paths for the dedicated FHIR schema and the combined API schema.""" + paths = {} + body = parameters_schema(components) + for tag in TAGS: + for base in BASES: + templates = capability_operations() if tag == 'Capabilities' else resource_operations(tag, base) + for suffix, methods in templates.items(): + path = base + suffix + for method, documented in methods.items(): + documented.operation_id = 'fhir_' + '_'.join( + path.strip('/').replace('{', '').replace('}', '').replace('$', '').replace('-', '_').split('/') + ) + '_' + method + documented.tags = [tag] + documented.parameters = deepcopy(documented.parameters) + path_parameters(path) + if method == 'post': + documented.consumes = [FHIR_JSON] + documented.parameters.append(openapi.Parameter( + 'parameters', openapi.IN_BODY, schema=body, + description='FHIR Parameters: use the query input names with one value[x] per parameter. ' + 'coding may use valueCoding; codeableConcept may use valueCodeableConcept ' + '(one coding for lookup/validation). Body values take precedence over query ' + 'values. _format stays in the query string.')) + paths[path] = openapi.PathItem(**methods) + return openapi.Paths(paths) + + +class FhirSwaggerSchemaGenerator(OpenAPISchemaGenerator): + """Generate the curated service schema without inspecting inherited resource write methods.""" + + def get_schema(self, request=None, public=False): # pylint: disable=unused-argument + components = self.reference_resolver_class(openapi.SCHEMA_DEFINITIONS, force_init=True) + paths = fhir_paths(components) + api_url = settings.API_BASE_URL.rstrip('/') + '/swagger/' + info = openapi.Info( + title='OCL FHIR terminology service (R4)', default_version='4.0.1', + description='FHIR R4 terminology in JSON and XML. Serves the SVCM transactions Query Value Set (ITI-95), ' + 'Query Code System (ITI-96), Expand Value Set (ITI-97), Lookup Code (ITI-98), ' + 'Validate Code (ITI-99), Query Concept Map (ITI-100) and Translate Code (ITI-101). ' + 'Bases: /fhir, /orgs/{org} and /users/{user}. The global base shows public resources; ' + 'owner bases show resources visible to the caller. ' + f'[OCL API Swagger]({api_url}) describes the rest of OCL.') + security_definitions = self.get_security_definitions() + return openapi.Swagger( + info=info, paths=paths, security_definitions=security_definitions, + security=self.get_security_requirements(security_definitions), + _url=request.build_absolute_uri('/') if request else self.url, _prefix='', **dict(components)) diff --git a/core/fhir/tests_swagger.py b/core/fhir/tests_swagger.py new file mode 100644 index 00000000..bd8cd2e5 --- /dev/null +++ b/core/fhir/tests_swagger.py @@ -0,0 +1,251 @@ +"""The terminology schema describes the routed reads and keeps the other API operations unchanged.""" +import re +from pathlib import Path +from tempfile import TemporaryDirectory + +from django.conf import settings +from django.http import HttpResponse +from django.test import RequestFactory, SimpleTestCase, override_settings +from django.urls import resolve +from drf_yasg import openapi +from drf_yasg.generators import OpenAPISchemaGenerator +from rest_framework.request import Request +from rest_framework.test import APIClient, APITestCase +from whitenoise.middleware import WhiteNoiseMiddleware + +from core.common.swagger import OCLSwaggerEndpointEnumerator, OCLSwaggerSchemaGenerator +from core.fhir.base import FhirViewMixin, is_fhir_path +from core.fhir.search import PARAMETERS, STRING_FIELDS, STRING_LOOKUPS +from core.fhir.swagger import FhirSwaggerSchemaGenerator +from core.middlewares.middlewares import FhirMiddleware +from core.urls import api_info + +BASES = ('/fhir', '/orgs/{org}', '/users/{user}') +OPERATIONS = { + 'CodeSystem': ('/$lookup', '/{id}/$lookup', '/$validate-code', '/{id}/$validate-code'), + 'ValueSet': ('/$validate-code', '/{id}/$validate-code', '/$expand', '/{id}/$expand'), + 'ConceptMap': ('/$translate',), +} +MIDDLEWARE = ['core.middlewares.middlewares.FhirMiddleware'] +SERVICE_MIDDLEWARE = tuple(settings.MIDDLEWARE) + + +def expected_operations(): + """An independent pin of the documented methods and paths at all three bases.""" + expected = {} + for base in BASES: + for resource, operations in OPERATIONS.items(): + for suffix in ('', '/{id}'): + expected[('get', f'{base}/{resource}{suffix}')] = resource + for suffix in operations: + for method in ('get', 'post'): + expected[(method, f'{base}/{resource}{suffix}')] = resource + for suffix in ('/metadata', '/$versions'): + expected[('get', base + suffix)] = 'Capabilities' + return expected + + +def operations_in(schema): + """Index actual schema operations without counting path-level parameters.""" + return {(method, path): operation for path, item in schema['paths'].items() + for method, operation in item.items() if method in openapi.PathItem.OPERATION_NAMES} + + +class FhirSwaggerTest(SimpleTestCase): + """Schema generation is independent of resource data and serializer introspection.""" + + def setUp(self): + self.schema = FhirSwaggerSchemaGenerator(api_info).get_schema(public=True) + + def test_exact_operations_and_routed_methods(self): + actual, expected = operations_in(self.schema), expected_operations() + self.assertEqual(set(actual), set(expected)) + self.assertEqual(len(actual), 78) + identifiers = [] + for (method, path), operation in actual.items(): + self.assertFalse(path.endswith('/'), path) + self.assertEqual(operation['tags'], [expected[(method, path)]]) + self.assertTrue(operation['summary']) + self.assertNotIn('\n', operation['summary']) + self.assertTrue(operation['description']) + self.assertNotIn('security', operation) # the schema's: a token shows private content at owner bases + identifiers.append(operation['operationId']) + concrete = path.replace('{org}', 'Acme').replace('{user}', 'jo').replace('{id}', '123') + view_class = resolve(concrete).func.view_class + self.assertTrue(issubclass(view_class, FhirViewMixin), path) + self.assertIn(method, view_class.http_method_names, path) + self.assertTrue(callable(getattr(view_class, method, None)), path) + parameters = operation['parameters'] + self.assertEqual({p['name'] for p in parameters if p['in'] == 'path'}, set(re.findall(r'{(\w+)}', path))) + self.assertEqual(len({(p['in'], p['name']) for p in parameters}), len(parameters)) + self.assertIn('_format', [p['name'] for p in parameters]) + self.assertEqual(len(identifiers), len(set(identifiers))) + + def test_search_parameters_match_the_search_parser(self): + for base in BASES: + for resource in OPERATIONS: + parameters = self.schema['paths'][f'{base}/{resource}']['get']['parameters'] + query = {p['name']: p for p in parameters if p['in'] == 'query'} + names = {f'{name}:{modifier}' if modifier else name + for name in STRING_FIELDS for modifier in STRING_LOOKUPS} + self.assertEqual(set(query), set(PARAMETERS[resource]) | names | {'_count', 'page', '_format'}) + self.assertEqual(query['_count']['type'], 'integer') + self.assertEqual(query['_count']['maximum'], 100) + self.assertEqual(query['_format']['enum'], ['json', 'xml']) + + def test_operation_parameters_and_bodies(self): + expected = { + 'CodeSystem/$lookup': {'system', 'version', 'code', 'coding', 'displayLanguage'}, + 'CodeSystem/$validate-code': {'url', 'version', 'code', 'coding', 'display', 'displayLanguage'}, + 'ValueSet/$validate-code': { + 'url', 'valueSetVersion', 'system', 'systemVersion', 'code', 'coding', 'display', 'displayLanguage'}, + 'ValueSet/$expand': { + 'url', 'valueSetVersion', 'filter', 'displayLanguage', 'includeDesignations', 'activeOnly', + 'offset', 'count', 'includeDefinition', 'excludePostCoordinated', 'excludeNested', 'excludeNotForUI'}, + 'ConceptMap/$translate': { + 'url', 'conceptMapVersion', 'code', 'system', 'version', 'coding', 'source', 'target', + 'targetsystem', 'reverse'}, + } + for path, names in expected.items(): + for base in BASES: + for method in ('get', 'post'): + operation = self.schema['paths'][f'{base}/{path}'][method] + parameters = operation['parameters'] + query = {p['name']: p for p in parameters if p['in'] == 'query'} + self.assertEqual(set(query), names | {'_format'}) + self.assertTrue(all('type' in parameter for parameter in query.values())) + bodies = [p for p in parameters if p['in'] == 'body'] + self.assertEqual(len(bodies), int(method == 'post')) + if bodies: + self.assertEqual(operation['consumes'], ['application/fhir+json']) + self.assertEqual(bodies[0]['schema']['$ref'], '#/definitions/FhirParameters') + expand = self.schema['paths']['/fhir/ValueSet/$expand']['get']['parameters'] + count = next(p for p in expand if p['name'] == 'count') + self.assertEqual((count['type'], count['maximum']), ('integer', 1000)) + + def test_coding_says_which_inputs_it_replaces(self): + # a coding replaces these inputs (core.fhir.ops: coded_input, translate.read_request), and the coding's + # description names them; those it leaves unread say so ($lookup's system still names an instance: mismatch) + replaced = { + 'CodeSystem/$lookup': ({'code', 'system'}, {'code'}), + 'CodeSystem/$validate-code': ({'code', 'display'}, {'code', 'display'}), + 'ValueSet/$validate-code': ({'code', 'system', 'systemVersion', 'display'}, + {'code', 'system', 'systemVersion', 'display'}), + 'ConceptMap/$translate': ({'code', 'system', 'version'}, {'code', 'system', 'version'}), + } + for path, (names, unread) in replaced.items(): + query = {p['name']: p for p in self.schema['paths'][f'/fhir/{path}']['get']['parameters']} + for name in names: + self.assertIn(name, query['coding']['description'], (path, name)) + self.assertEqual('Not read when coding' in query[name]['description'], name in unread, (path, name)) + lookup = {p['name']: p for p in self.schema['paths']['/fhir/CodeSystem/$lookup']['get']['parameters']} + self.assertIn('At an instance it must name that code system', lookup['system']['description']) + + def test_tags_order_responses_and_security_definitions(self): + for tag in (*OPERATIONS, 'Capabilities'): + paths = [path for (_, path), operation in operations_in(self.schema).items() if operation['tags'] == [tag]] + order = [next(index for index, base in enumerate(BASES) if path.startswith(base + '/')) for path in paths] + self.assertEqual(order, sorted(order)) + self.assertEqual(set(self.schema['securityDefinitions']), {'Basic', 'Token'}) + self.assertEqual({name for requirement in self.schema['security'] for name in requirement}, {'Basic', 'Token'}) + for (_, path), operation in operations_in(self.schema).items(): + success = operation['responses']['200'] + self.assertTrue(success['schema']) + for code, response in operation['responses'].items(): + if code != '200': + self.assertEqual(response['schema']['properties']['resourceType']['enum'], ['OperationOutcome']) + if path.endswith('/metadata'): + self.assertIn('TerminologyCapabilities', success['description']) + + +@override_settings(FHIR_SUBDOMAIN='fhir', MIDDLEWARE=MIDDLEWARE, ALLOWED_HOSTS=['*']) +class FhirSwaggerHostTest(SimpleTestCase): + """The FHIR host serves its own documentation without opening other API paths.""" + + def test_schema_urls_use_the_request_host_and_forwarded_scheme(self): + for path in ('/swagger/?format=openapi', '/swagger.json', '/swagger.yaml'): + response = APIClient().get(path, HTTP_HOST='fhir.example.org:8443', HTTP_X_FORWARDED_PROTO='https') + self.assertEqual(response.status_code, 200, path) + self.assertEqual(set(operations_in(response.data)), set(expected_operations())) + self.assertEqual(response.data['host'], 'fhir.example.org:8443') + self.assertEqual(response.data['schemes'], ['https']) + self.assertEqual(response.data['basePath'], '/') + self.assertIn('FHIR terminology', response.data['info']['title']) + + def test_only_the_documentation_paths_open_on_the_fhir_host(self): + response = APIClient().get('/swagger/', HTTP_HOST='fhir.example.org', HTTP_ACCEPT='text/html') + self.assertEqual(response.status_code, 200) + middleware = FhirMiddleware(lambda request: HttpResponse('allowed')) + for path in ('/swagger/', '/swagger.json', '/swagger.yaml', '/version/'): + request = RequestFactory().get(path, HTTP_HOST='fhir.example.org') + self.assertEqual(middleware(request).content, b'allowed', path) + for path in ('/orgs/', '/redoc/', '/swagger', '/swagger/other', '/swagger.json/extra', '/swagger.yaml/'): + response = APIClient().get(path, HTTP_HOST='fhir.example.org') + self.assertEqual(response.status_code, 404, path) + self.assertEqual(response.json()['resourceType'], 'OperationOutcome', path) + + def test_the_ui_files_are_served_ahead_of_the_host_check(self): + """Static files are served before FhirMiddleware, so the Swagger UI's files load on the FHIR host.""" + static = 'whitenoise.middleware.WhiteNoiseMiddleware' + if static in SERVICE_MIDDLEWARE: + self.assertLess(SERVICE_MIDDLEWARE.index(static), + SERVICE_MIDDLEWARE.index('core.middlewares.middlewares.FhirMiddleware')) + with TemporaryDirectory() as directory: + target = Path(directory) / 'drf-yasg' / 'style.css' + target.parent.mkdir(parents=True) + target.write_text('body {}') + with self.settings(STATIC_ROOT=directory, STATIC_URL='/static/', WHITENOISE_AUTOREFRESH=True, + WHITENOISE_USE_FINDERS=False): + middleware = WhiteNoiseMiddleware(FhirMiddleware( + lambda request: HttpResponse('not static', status=404))) + response = middleware(RequestFactory().get('/static/drf-yasg/style.css', HTTP_HOST='fhir.example.org')) + self.assertEqual(response.status_code, 200) + response.close() + + +class OriginalSchemaGenerator(OpenAPISchemaGenerator): + """The API generator before FHIR curation, for a comparison of non-FHIR operations.""" + endpoint_enumerator_class = OCLSwaggerEndpointEnumerator + + +@override_settings(MIDDLEWARE=MIDDLEWARE, ALLOWED_HOSTS=['*']) +class ApiSwaggerTest(APITestCase): + """Compare the complete API schema using real resolver endpoints and the original inspector.""" + + def schema(self): + response = self.client.get('/swagger/?format=openapi', HTTP_HOST='api.example.org', + HTTP_X_FORWARDED_PROTO='https') + self.assertEqual(response.status_code, 200) + return response.data + + @override_settings(FHIR_SUBDOMAIN='fhir') + def test_separate_host_omits_fhir_and_links_to_its_swagger(self): + schema = self.schema() + self.assertFalse(any(is_fhir_path(path) for path in schema['paths'])) + self.assertFalse(any('CodeSystem' in path or 'ValueSet' in path or 'ConceptMap' in path + for path in schema['paths'])) + self.assertIn('https://fhir.example.org/swagger/', schema['info']['description']) + self.assertNotIn('fhir.example.org', api_info.description) + + @override_settings(FHIR_SUBDOMAIN=None) + def test_same_host_includes_only_curated_fhir(self): + schema = self.schema() + fhir = {key: operation for key, operation in operations_in(schema).items() if is_fhir_path(key[1])} + curated = operations_in(FhirSwaggerSchemaGenerator(api_info).get_schema(public=True)) + self.assertEqual(fhir, curated) + self.assertFalse(any('CodeSystem$' in path or 'ValueSet$' in path or 'ConceptMap$' in path + for path in schema['paths'])) + + def test_non_fhir_paths_and_operations_are_unchanged(self): + request = Request(RequestFactory().get('/swagger/', HTTP_HOST='api.example.org')) + original = OriginalSchemaGenerator(api_info) + endpoints = {path: endpoint for path, endpoint in original.get_endpoints(request).items() + if not issubclass(endpoint[0], FhirViewMixin) and not is_fhir_path(path)} + original_schema = original.get_schema(request, public=True) + expected = {path: item for path, item in original_schema.paths.items() if path in endpoints} + self.assertTrue(expected) + for subdomain in ('fhir', None): + with self.settings(FHIR_SUBDOMAIN=subdomain): + schema = OCLSwaggerSchemaGenerator(api_info).get_schema(request, public=True) + actual = {path: item for path, item in schema.paths.items() if not is_fhir_path(path)} + self.assertEqual(actual, dict(expected)) diff --git a/core/middlewares/middlewares.py b/core/middlewares/middlewares.py index 15eb94f7..635afae1 100644 --- a/core/middlewares/middlewares.py +++ b/core/middlewares/middlewares.py @@ -19,6 +19,7 @@ from core.common.utils import set_current_user, set_request_url from core.fhir.base import FHIR_JSON, is_fhir_path, is_outcome_response, outcome_response, response_json from core.fhir.outcomes import issue, operation_outcome, outcome_for_status +from core.fhir.swagger import on_fhir_host from core.services.analytics_event_emitter import AnalyticsEventEmitter from core.services.auth.core import AuthService @@ -298,14 +299,16 @@ class FhirMiddleware(BaseMiddleware): Keeps FHIR on the FHIR host when FHIR_SUBDOMAIN is set: FHIR resource paths 404 elsewhere, and other paths 404 on the FHIR host, each with an OperationOutcome. """ + documentation_paths = ('/swagger/', '/swagger.json', '/swagger.yaml') # the FHIR host's own Swagger def __call__(self, request): fhir_path = is_fhir_path(request.path) if settings.FHIR_SUBDOMAIN: - on_fhir_host = request.get_host().startswith(settings.FHIR_SUBDOMAIN + '.') - if on_fhir_host and not fhir_path and request.path.strip('/').split('/')[0] != 'version': + fhir_host = on_fhir_host(request) + if fhir_host and not fhir_path and request.path.strip('/').split('/')[0] != 'version' and \ + request.path not in self.documentation_paths: return self.not_found(request, 'This path is not part of the FHIR API.') - if not on_fhir_host and fhir_path: + if not fhir_host and fhir_path: return self.not_found(request, 'The FHIR API is served on its own host.') return self.get_response(request)