enable login

Author: michaelmattig

geoengine/auth.py

@@ -2,9 +2,10 @@
 Module for encapsulating Geo Engine authentication
 '''
 
-from typing import ClassVar, Dict, Optional
+from typing import ClassVar, Dict, Optional, Tuple
 from uuid import UUID
 
+import os
 import requests as req
 from requests.auth import AuthBase
 
@@ -35,14 +36,25 @@ class Session:
 
     session: ClassVar[req.Session] = None
 
-    def __init__(self, server_url: str) -> None:
+    def __init__(self, server_url: str, credentials: Tuple[str, str] = None) -> None:
         '''
         Initialize communication between this library and a Geo Engine instance
+
+        if credentials are provided, the session will be authenticated
+
+        optional arguments: (email, password) as tuple
+        optional environment variables: GEOENGINE_EMAIL, GEOENGINE_PASSWORD
         '''
 
-        # TODO: username and password for Pro version
+        session = None
 
-        session = req.post(f'{server_url}/anonymous').json()
+        if credentials is not None:
+            session = req.post(f'{server_url}/login', json={"email": credentials[0], "password": credentials[1]}).json()
+        elif "GEOENGINE_EMAIL" in os.environ and "GEOENGINE_PASSWORD" in os.environ:
+            session = req.post(f'{server_url}/login',
+                               json={"email": os.environ.get("GEOENGINE_EMAIL"), "password": os.environ.get("GEOENGINE_PASSWORD")}).json()
+        else:
+            session = req.post(f'{server_url}/anonymous').json()
 
         if 'error' in session:
             raise GeoEngineException(session)
@@ -87,6 +99,13 @@ def requests_bearer_auth(self) -> BearerAuth:
 
         return BearerAuth(self.__id)
 
+    def logout(self):
+        '''
+        Logout the current session
+        '''
+
+        req.post(f'{self.server_url}/logout', headers=self.auth_header)
+
 
 def get_session() -> Session:
     '''
@@ -101,19 +120,25 @@ def get_session() -> Session:
     return Session.session
 
 
-def initialize(server_url: str) -> None:
+def initialize(server_url: str, credentials: Tuple[str, str] = None) -> None:
     '''
     Initialize communication between this library and a Geo Engine instance
+
+    if credentials are provided, the session will be authenticated
+
+    optional arugments: (email, password) as tuple
+    optional environment variables: GEOENGINE_EMAIL, GEOENGINE_PASSWORD
     '''
 
-    Session.session = Session(server_url)
+    Session.session = Session(server_url, credentials)
 
 
-def reset() -> None:
+def reset(logout: bool = True) -> None:
     '''
     Resets the current session
     '''
 
-    # TODO: active logout for Pro version
+    if Session.session is not None and logout:
+        Session.session.logout()
 
     Session.session = None

tests/test_auth.py

@@ -3,6 +3,8 @@
 from datetime import datetime
 
 import unittest
+import json
+import os
 import requests_mock
 
 import geoengine as ge
@@ -13,7 +15,9 @@ class AuthTests(unittest.TestCase):
     '''Tests runner regarding Geo Engine authentication'''
 
     def setUp(self) -> None:
-        ge.reset()
+        assert "GEOENGINE_EMAIL" not in os.environ and "GEOENGINE_PASSWORD" not in os.environ, \
+            "Please unset GEOENGINE_EMAIL and GEOENGINE_PASSWORD"
+        ge.reset(False)
 
     def test_uninitialized(self):
         with self.assertRaises(ge.UninitializedException) as exception:
@@ -45,6 +49,53 @@ def test_initialize(self):
             self.assertEqual(type(ge.get_session()),
                              ge.Session)
 
+    def test_initialize_tuple(self):
+        with requests_mock.Mocker() as m:
+            m.post('http://mock-instance/login',
+                   additional_matcher=lambda request: request.text == json.dumps({"email": "[email protected]", "password": "secret123"}),
+                   json={
+                       "id": "e327d9c3-a4f3-4bd7-a5e1-30b26cae8064",
+                       "user": {
+                           "id": "328ca8d1-15d7-4f59-a989-5d5d72c98744",
+                       },
+                       "created": "2021-06-08T15:22:22.605891994Z",
+                       "validUntil": "2021-06-08T16:22:22.605892183Z",
+                       "project": None,
+                       "view": None
+                   })
+
+            ge.initialize("http://mock-instance", ("[email protected]", "secret123"))
+
+            self.assertEqual(type(ge.get_session()),
+                             ge.Session)
+
+    def test_initialize_env(self):
+        with requests_mock.Mocker() as m:
+            m.post('http://mock-instance/login',
+                   additional_matcher=lambda request: request.text == json.dumps({"email": "[email protected]", "password": "secret123"}),
+                   json={
+                       "id": "e327d9c3-a4f3-4bd7-a5e1-30b26cae8064",
+                       "user": {
+                           "id": "328ca8d1-15d7-4f59-a989-5d5d72c98744",
+                       },
+                       "created": "2021-06-08T15:22:22.605891994Z",
+                       "validUntil": "2021-06-08T16:22:22.605892183Z",
+                       "project": None,
+                       "view": None
+                   })
+
+            os.environ["GEOENGINE_EMAIL"] = "[email protected]"
+            os.environ["GEOENGINE_PASSWORD"] = "secret123"
+
+            try:
+                ge.initialize("http://mock-instance")
+            finally:
+                del os.environ["GEOENGINE_EMAIL"]
+                del os.environ["GEOENGINE_PASSWORD"]
+
+            self.assertEqual(type(ge.get_session()),
+                             ge.Session)
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_plot.py

@@ -14,7 +14,7 @@ class WmsTests(unittest.TestCase):
     '''Test runner for the plotting functionality'''
 
     def setUp(self) -> None:
-        ge.reset()
+        ge.reset(False)
 
     def test_ndvi_histogram(self):
         with requests_mock.Mocker() as m:

tests/test_provenance.py

@@ -11,7 +11,7 @@ class ProvenanceTests(unittest.TestCase):
     '''Test runner for provenance tests'''
 
     def setUp(self) -> None:
-        ge.reset()
+        ge.reset(False)
 
     def test_provenance_call(self):
         with requests_mock.Mocker() as m:

tests/test_upload.py

@@ -14,7 +14,7 @@ class UploadTests(unittest.TestCase):
     '''Test runner regarding upload functionality'''
 
     def setUp(self) -> None:
-        ge.reset()
+        ge.reset(False)
 
     def test_upload(self):
         with requests_mock.Mocker() as m:

tests/test_wcs.py

@@ -14,7 +14,7 @@ class WcsTests(unittest.TestCase):
     '''WCS test runner'''
 
     def setUp(self) -> None:
-        ge.reset()
+        ge.reset(False)
 
     def test_ndvi(self):
         with requests_mock.Mocker() as m, open("tests/responses/ndvi.tiff", "rb") as ndvi_tiff:

tests/test_wfs.py

@@ -18,7 +18,7 @@ class WfsTests(unittest.TestCase):
     '''WFS test runner'''
 
     def setUp(self) -> None:
-        ge.reset()
+        ge.reset(False)
 
     def test_geopandas(self):
         with requests_mock.Mocker() as m:

tests/test_wms.py

@@ -18,7 +18,7 @@ class WmsTests(unittest.TestCase):
     '''WMS test runner'''
 
     def setUp(self) -> None:
-        ge.reset()
+        ge.reset(False)
 
     @responses.activate
     @ImageTesting(['wms'], tolerance=0)

tests/test_workflow_storage.py

@@ -15,7 +15,7 @@ class WorkflowStorageTests(unittest.TestCase):
     '''Test methods for storing workflows as datasets'''
 
     def setUp(self) -> None:
-        ge.reset()
+        ge.reset(False)
 
     def test_storing_workflow(self):
         with requests_mock.Mocker() as m: