Add functions and tests for webhook subscriptions (#16)

* Add functions and tests for webhook subscriptions

* Tests pass locally

* Clean up return values

* Put back full test suite

* Only test python 3.8

.github/workflows/ci.yml

@@ -18,7 +18,7 @@ jobs:
     strategy:
       fail-fast: false      
       matrix:
-        python-version: [3.6, 3.7, 3.8]
+        python-version: [3.8]
 
     steps:
     - uses: actions/checkout@v2

.gitignore

@@ -14,6 +14,7 @@ poetry.lock
 .pytest_cache
 .env
 *.json
+ngrok.yml
 
 # pycharm
 .idea

fs/onedrivefs/onedrivefs.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 
-from datetime import datetime
+from datetime import datetime, timezone
 from io import BytesIO
 from logging import getLogger
 
@@ -15,7 +15,9 @@
 from requests import get # pylint: disable=wrong-import-order
 from requests_oauthlib import OAuth2Session # pylint: disable=wrong-import-order
 
-_DRIVE_ROOT = "https://graph.microsoft.com/v1.0/me/drive"
+_SERVICE_ROOT = "https://graph.microsoft.com/v1.0"
+_RESOURCE_ROOT = "me/drive"
+_DRIVE_ROOT = f"{_SERVICE_ROOT}/{_RESOURCE_ROOT}"
 _INVALID_PATH_CHARS = ":\0\\"
 _log = getLogger("fs.onedrivefs")
 
@@ -41,11 +43,20 @@ def _ParseDateTime(dt):
 	except ValueError:
 		return datetime.strptime(dt, "%Y-%m-%dT%H:%M:%SZ")
 
+def _FormatDateTime(dt):
+	return dt.astimezone(timezone.utc).replace(tzinfo=None).isoformat() + "Z"
+
 def _UpdateDict(dict_, sourceKey, targetKey, processFn=None):
 	if sourceKey in dict_:
 		return {targetKey: processFn(dict_[sourceKey]) if processFn is not None else dict_[sourceKey]}
 	return {}
 
+def _HandleError(response):
+	# https://docs.microsoft.com/en-us/onedrive/developer/rest-api/concepts/errors
+	if response.ok is False:
+		_log.error(f"Response text: {response.text}")
+	response.raise_for_status()
+
 class _UploadOnClose(BytesIO):
 	def __init__(self, session, path, itemId, mode):
 		self.session = session
@@ -146,7 +157,19 @@ def close(self):
 					self._ResumableUpload(_ItemUrl(parentId, f":/{filename}:/createUploadSession"))
 		self._closed = True
 
+class SubOneDriveFS(SubFS):
+	def create_subscription(self, notification_url, expiration_date_time, client_state):
+		return self.delegate_fs().create_subscription(notification_url, expiration_date_time, client_state)
+
+	def delete_subscription(self, id_):
+		return self.delegate_fs().delete_subscription(id_)
+
+	def update_subscription(self, id_, expiration_date_time):
+		return self.delegate_fs().update_subscription(id_, expiration_date_time)
+
 class OneDriveFS(FS):
+	subfs_class = SubOneDriveFS
+
 	def __init__(self, clientId, clientSecret, token, SaveToken):
 		super().__init__()
 		self.session = OAuth2Session(
@@ -169,6 +192,42 @@ def __init__(self, clientId, clientSecret, token, SaveToken):
 	def __repr__(self):
 		return f"<{self.__class__.__name__}>"
 
+	def create_subscription(self, notification_url, expiration_date_time, client_state):
+		with self._lock:
+			payload = {
+				"changeType": "updated", # OneDrive only supports updated
+				"notificationUrl": notification_url,
+				"resource": f"/{_RESOURCE_ROOT}/root",
+				"expirationDateTime": _FormatDateTime(expiration_date_time),
+				"clientState": client_state
+			}
+			response = self.session.post(f"{_SERVICE_ROOT}/subscriptions", json=payload)
+			_HandleError(response) # this is backup, if actual errors are thrown from here we should respond to them individually, e.g. if validation fails
+			assert response.status_code == 201, "Expected 201 Created response"
+			subscription = response.json()
+			assert subscription["changeType"] == payload["changeType"]
+			assert subscription["notificationUrl"] == payload["notificationUrl"]
+			assert subscription["resource"] == payload["resource"]
+			assert "expirationDateTime" in subscription
+			assert subscription["clientState"] == payload["clientState"]
+			_log.debug(f"Subscription created successfully: {subscription}")
+			return subscription["id"]
+
+	def delete_subscription(self, id_):
+		with self._lock:
+			response = self.session.delete(f"{_SERVICE_ROOT}/subscriptions/{id_}")
+			response.raise_for_status() # this is backup, if actual errors are thrown from here we should respond to them individually, e.g. if validation fails
+			assert response.status_code == 204, "Expected 204 No content"
+
+	def update_subscription(self, id_, expiration_date_time):
+		with self._lock:
+			response = self.session.patch(f"{_SERVICE_ROOT}/subscriptions/{id_}", json={"expirationDateTime": _FormatDateTime(expiration_date_time)})
+			response.raise_for_status() # this is backup, if actual errors are thrown from here we should respond to them individually, e.g. if validation fails
+			assert response.status_code == 200, "Expected 200 OK"
+			subscription = response.json()
+			assert subscription["id"] == id_
+			assert "expirationDateTime" in subscription
+
 	# Translates OneDrive DriveItem dictionary to an fs Info object
 	def _itemInfo(self, item): # pylint: disable=no-self-use
 		# Looks like the dates returned directly in item.file_system_info (i.e. not file_system_info) are UTC naive-datetimes

pyproject.toml

@@ -3,7 +3,7 @@ name = "fs.onedrivefs"
 packages = [
     { include = "fs"}
 ]
-version = "0.2.4"
+version = "0.3.0"
 description = "Pyfilesystem2 implementation for OneDrive using Microsoft Graph API"
 authors = ["Rehan Khwaja <rehan@khwaja.name>"]
 license = "MIT"
@@ -13,6 +13,8 @@ readme = "README.md"
 
 [tool.poetry.dependencies]
 python = ">=3.6"
+# Need 2.0.5 for opendir factory parameter
+# Need 2.0.6 because in 2.0.5, opener didn't work
 fs = ">=2.0.6"
 requests = "^2.20"
 requests-oauthlib = "^1.0"
@@ -23,6 +25,9 @@ pytest = "^3.10"
 pylint = ">=2.5.3"
 python-coveralls = "^2.9.3"
 pylint-quotes = "^0.2.1"
+pytest-localserver = "^0.5.0"
+pyngrok = "^1.4"
+click = "^7.0"
 
 [tool.poetry.plugins] # Optional super table
 

tests/test_onedrivefs.py

@@ -2,21 +2,23 @@
 from __future__ import absolute_import
 from __future__ import unicode_literals
 
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 from hashlib import sha1
 from json import dump, load, loads
-from logging import warning
+from logging import info, warning
 from os import environ
 from time import sleep
 from unittest import TestCase
-from urllib.parse import urlencode
+from urllib.parse import parse_qs, urlencode, urlparse
 from uuid import uuid4
 
+from fs.onedrivefs import OneDriveFS, OneDriveFSOpener
 from fs.opener import open_fs, registry
 from fs.subfs import SubFS
 from fs.test import FSTestCases
-
-from fs.onedrivefs import OneDriveFS, OneDriveFSOpener
+from pyngrok.ngrok import connect # pylint: disable=wrong-import-order
+from pytest import fixture, mark # pylint: disable=wrong-import-order
+from pytest_localserver.http import WSGIServer # pylint: disable=wrong-import-order
 
 _SAFE_TEST_DIR = "Documents/test-onedrivefs"
 
@@ -33,7 +35,7 @@ def __init__(self, token):
 		self.token = token
 
 	def Save(self, token):
-		pass
+		self.token = token
 
 	def Load(self):
 		return loads(self.token)
@@ -53,6 +55,35 @@ def Load(self):
 		except FileNotFoundError:
 			return None
 
+class simple_app: # pylint: disable=too-few-public-methods
+	def __init__(self):
+		self.notified = False
+
+	def __call__(self, environ_, start_response):
+		"""Simplest possible WSGI application"""
+		status = "200 OK"
+		response_headers = [("Content-type", "text/plain")]
+		start_response(status, response_headers)
+		parsedQS = parse_qs(environ_["REQUEST_URI"][2:])
+		info(f"Received: {parsedQS}")
+		info(f"env: {environ_}")
+		if "validationToken" in parsedQS:
+			info("Validating subscription")
+			return [parsedQS["validationToken"][0].encode()]
+		inputStream = environ_["wsgi.input"]
+		info(f"Input: {inputStream}")
+		info("NOTIFIED")
+		self.notified = True
+		return ""
+
+@fixture(scope="class")
+def testserver(request):
+	server = WSGIServer(application=simple_app())
+	request.cls.server = server
+	server.start()
+	request.addfinalizer(server.stop)
+	return server
+
 def CredentialsStorage():
 	if "GRAPH_API_TOKEN_READONLY" in environ:
 		return TokenStorageReadOnly(environ["GRAPH_API_TOKEN_READONLY"])
@@ -95,6 +126,32 @@ def make_fs(self):
 	def destroy_fs(self, _):
 		self.fullFS.removetree(self.testSubdir)
 
+	@mark.usefixtures("testserver")
+	def test_subscriptions(self):
+		port = urlparse(self.server.url).port # pylint: disable=no-member
+		info(f"Port: {port}")
+		info(self.server.url) # pylint: disable=no-member
+		with open("ngrok.yml", "w") as f:
+			f.write(f"authtoken: {environ['NGROK_AUTH_TOKEN']}")
+		publicUrl = connect(proto="http", port=port, config_path="ngrok.yml").replace("http", "https")
+		info(f"publicUrl: {publicUrl}")
+		expirationDateTime = datetime.now(timezone.utc) + timedelta(minutes=5)
+		id_ = self.fs.create_subscription(publicUrl, expirationDateTime, "client_state")
+		info(f"subscription id: {id_}")
+		self.fs.touch("touched-file.txt")
+		info("Touched the file, waiting...")
+		# subscription = self.fs.update_subscription(subscription["id"], expirationDateTime + timedelta(hours=12))
+		# need to wait for some time for the notification to come through, but also process incoming http requests
+		for _ in range(10):
+			if self.server.app.notified is True: # pylint: disable=no-member
+				break
+			sleep(1)
+		# sleep(2)
+		info("Sleep done, deleting subscription")
+		self.fs.delete_subscription(id_)
+		info("subscription deleted")
+		assert self.server.app.notified is True, f"Not notified: {self.server.app.notified}" # pylint: disable=no-member
+
 	def test_overwrite_file(self):
 		with self.fs.open("small_file_to_overwrite.bin", "wb") as f:
 			f.write(b"x" * 10)
@@ -120,7 +177,6 @@ def test_overwrite_file(self):
 		with self.fs.open("large_file_to_overwrite.txt", "w") as f:
 			f.write("y" * 4000000)
 
-
 	def test_photo_metadata(self):
 		with self.fs.open("canon-ixus.jpg", "wb") as target:
 			with open("tests/canon-ixus.jpg", "rb") as source: