Spec Coverage via. matrix-doc in pytest

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "matrix-doc"]
+	path = matrix-doc
+	url = https://github.com/matrix-org/matrix-doc.git

matrix_client/api.py

@@ -97,20 +97,15 @@ def validate_certificate(self, valid):
         self.validate_cert = valid
         return
 
-    def register(self, login_type, **kwargs):
+    def register(self, content={}, query_params={}):
         """Performs /register.
 
         Args:
-            login_type(str): The value for the 'type' key.
-            **kwargs: Additional key/values to add to the JSON submitted.
+            content(dict): The request payload. Should include "type" such as "m.login.password" for all non-guest registrations.
+            query_params(dict): The query params for the request. Specify "kind": "guest" to register a guest account.
         """
-        content = {
-            "type": login_type
-        }
-        for key in kwargs:
-            content[key] = kwargs[key]
 
-        return self._send("POST", "/register", content, api_path=MATRIX_V2_API_PATH)
+        return self._send("POST", "/register", content=content, query_params=query_params, api_path=MATRIX_V2_API_PATH)
 
     def login(self, login_type, **kwargs):
         """Perform /login.
@@ -504,7 +499,7 @@ def create_filter(self, user_id, filter_params):
                           api_path=MATRIX_V2_API_PATH)
 
     def _send(self, method, path, content=None, query_params={}, headers={},
-              api_path="/_matrix/client/api/v1"):
+              api_path=MATRIX_V2_API_PATH):
         method = method.upper()
         if method not in ["GET", "PUT", "DELETE", "POST"]:
             raise MatrixError("Unsupported HTTP method: %s" % method)

matrix_client/client.py

@@ -112,6 +112,23 @@ def set_sync_token(self, token):
     def set_user_id(self, user_id):
         self.user_id = user_id
 
+    def register_as_guest(self):
+        """ Register a guest account on this HS.
+        Note: HS must have guest registration enabled.
+        Returns:
+            str: Access Token
+        Raises:
+            MatrixRequestError
+        """
+        response = self.api.register(query_params={'kind': 'guest'})
+        self.user_id = response["user_id"]
+        self.token = response["access_token"]
+        self.hs = response["home_server"]
+        self.api.token = self.token
+        self.sync_filter = '{ "room": { "timeline" : { "limit" : 20 } } }'
+        self._sync()
+        return self.token
+
     def register_with_password(self, username, password, limit=1):
         """ Register for a new account on this HS.
 
@@ -127,7 +144,7 @@ def register_with_password(self, username, password, limit=1):
             MatrixRequestError
         """
         response = self.api.register(
-            "m.login.password", user=username, password=password
+            {'type': "m.login.password", 'user': username, 'password': password}
         )
         self.user_id = response["user_id"]
         self.token = response["access_token"]

test/api_test.py

@@ -1,7 +1,9 @@
-import responses
+import pytest
+responses = pytest.responses_with_api_guide
+if not responses:
+    import responses
 from matrix_client import client
 
-
 class TestTagsApi:
     cli = client.MatrixClient("http://example.com")
     user_id = "@user:matrix.org"
@@ -67,14 +69,15 @@ class TestUnbanApi:
     cli = client.MatrixClient("http://example.com")
     user_id = "@user:matrix.org"
     room_id = "#foo:matrix.org"
-    
+
     @responses.activate
     def test_unban(self):
         unban_url = "http://example.com" \
-                "/_matrix/client/api/v1/rooms/#foo:matrix.org/unban"
+                "/_matrix/client/r0/rooms/#foo:matrix.org/unban"
         body = '{"user_id": "'+ self.user_id + '"}'
         responses.add(responses.POST, unban_url, body=body)
         self.cli.api.unban_user(self.room_id, self.user_id)
         req = responses.calls[0].request
         assert req.url == unban_url
         assert req.method == 'POST'
+

test/conftest.py

@@ -0,0 +1 @@
+pytest_plugins = "matrix_spec_coverage_plugin"

test/matrix_spec_coverage.py

@@ -0,0 +1,94 @@
+import sys
+import re
+import yaml
+from responses import RequestsMock
+
+INTERPOLATIONS = [
+    ("%CLIENT_MAJOR_VERSION%", "r0")
+]
+
+def interpolate_str(s):
+    for interpolation in INTERPOLATIONS:
+        s = s.replace(interpolation[0], interpolation[1])
+    return s
+
+def endpoint_to_regex(s):
+    # TODO sub by with more specific REGEXes per type
+    # e.g. roomId, eventId, userId
+    return re.sub('\{[a-zA-Z]+\}', '[a-zA-Z!\.:-@#]+', s)
+
+MISSING_BASE_PATH = "Not a valid API Base Path: "
+MISSING_ENDPOINT = "Not a valid API Endpoint: "
+MISSING_METHOD = "Not a valid API Method: "
+
+class ApiGuide:
+    def __init__(self, hostname="http://example.com"):
+        self.hostname = hostname
+        self.endpoints = {}
+        self.called = []
+        self.missing = []
+        self.total_endpoints = 0
+
+    def setup_from_files(self, files):
+        for file in files:
+            with open(file) as rfile:
+                definitions = yaml.load(rfile)
+                base_path = definitions['basePath']
+                resolved_base_path = interpolate_str(base_path)
+                if resolved_base_path not in self.endpoints:
+                    self.endpoints[resolved_base_path] = {}
+                regex_paths = { endpoint_to_regex(k): v for k,v in definitions['paths'].items() }
+                self.endpoints[resolved_base_path].update(regex_paths)
+                endpoints_added = sum(len(v) for v in definitions['paths'].values())
+                self.total_endpoints += endpoints_added
+
+    def process_request(self, request):
+        full_path_url = request.url
+        method = request.method
+        body = request.body
+        for base_path in self.endpoints.keys():
+            if base_path in full_path_url:
+                path_url = full_path_url.replace(base_path, '')
+                path_url = path_url.replace(self.hostname, '')
+                break
+        else:
+            self.add_called_missing(MISSING_BASE_PATH, request)
+            return
+        endpoints = self.endpoints[base_path]
+        for endpoint in endpoints.keys():
+            if re.fullmatch(endpoint, path_url):
+                break
+        else:
+            self.add_called_missing(MISSING_ENDPOINT, request)
+            return
+        endpoint_def = endpoints[endpoint]
+        try:
+            endpoint_def[method.lower()]
+            self.add_called(base_path, endpoint, method, body)
+        except KeyError:
+            self.add_called_missing(MISSING_METHOD, request)
+
+
+    def add_called(self, base_path, endpoint, method, body):
+        self.called.append((base_path, endpoint, method, body))
+
+    def add_called_missing(self, error,request):
+        self.missing.append((error, request.url, request.method, request.body))
+
+    def print_summary(self):
+        print("Accessed: %i out of %i endpoints. %0.2f%% Coverage." %
+            (len(self.called), self.total_endpoints, len(self.called)*100 / self.total_endpoints)
+        )
+        if self.missing:
+            missing_summary = "\n".join(m[0] + ", ".join(m[1:-1]) for m in self.missing)
+            raise AssertionError("The following invalid API Requests were made:\n" +
+                missing_summary)
+
+class RequestsMockWithApiGuide(RequestsMock):
+    def __init__(self, api_guide, assert_all_requests_are_fired=True):
+        self.api_guide = api_guide
+        super().__init__(assert_all_requests_are_fired)
+
+    def _on_request(self, adapter, request, **kwargs):
+        self.api_guide.process_request(request)
+        return super()._on_request(adapter, request, **kwargs)

test/matrix_spec_coverage_plugin.py

@@ -0,0 +1,32 @@
+import _pytest
+import pytest
+from _pytest._pluggy import HookspecMarker
+from matrix_spec_coverage import ApiGuide, RequestsMockWithApiGuide
+
+hookspec = HookspecMarker("pytest")
+
+# We use this to print api_guide coverage stats
+# after pytest has finished running
+def pytest_terminal_summary(terminalreporter, exitstatus):
+    if pytest.responses_with_api_guide:
+        guide = pytest.responses_with_api_guide.api_guide
+        guide.print_summary()
+
+
+def build_api_guide():
+    import os
+    from glob import glob
+    DOC_FOLDER = "../matrix-doc/api/client-server/"
+    api_files = glob(os.path.join(DOC_FOLDER, '*.yaml'))
+    if not api_files:
+        return
+    guide = ApiGuide()
+    guide.setup_from_files(API_FILES)
+    return guide
+
+# Load api_guide stats into the pytest namespace so
+# that we can print a the stats on terminal summary
+@hookspec(historic=True)
+def pytest_namespace():
+    guide = build_api_guide()
+    return { 'responses_with_api_guide': RequestsMockWithApiGuide(guide) }