swimlane: update client and config

automon/integrations/swimlaneWrapper/__init__.py

@@ -0,0 +1 @@
+from .client import SwimlaneClient, SwimlaneClientRest, SwimlaneConfig

automon/integrations/swimlaneWrapper/api/v1.py

@@ -0,0 +1,20 @@
+class Api(object):
+    api = f'api'
+
+
+class User(object):
+    user = f'{Api.api}/user'
+    login = f'{user}/login'
+
+
+class Auth(object):
+
+    def __init__(self, userId: str):
+        self.auth = f'{Api.api}/auth'
+
+        self.token = f'{self.auth}/token'
+        self.create = f'{self.token}/create'
+
+        self.user = f'{self.auth}/user'
+        self.userId = userId
+        self.token = f'{userId}/token'

automon/integrations/swimlaneWrapper/client.py

@@ -1,6 +1,57 @@
+from automon import log
+from automon.integrations.requestsWrapper import RequestsClient
+
+from .config import SwimlaneConfig
+from .api.v1 import Auth, User
+
+logger = log.logging.getLogger(__name__)
+logger.setLevel(log.DEBUG)
+
+
 class SwimlaneClient(object):
     pass
 
 
 class SwimlaneClientRest(object):
-    pass
+
+    def __init__(self):
+        self.config = SwimlaneConfig()
+
+        self.requests = RequestsClient()
+
+    async def is_ready(self):
+        if await self.config.is_ready():
+            return True
+
+    async def test_connection(self):
+        return
+
+    async def login(self):
+        """Login with username and password"""
+        url = f'{self.host}/{User.login}'
+
+        r = await self.requests.post(
+            url=url,
+            data=self.config.credentials,
+        )
+
+        return
+
+    async def auth_create_token(self):
+        """Creates a new access token for the user making the request"""
+        url = f'{self.host}/{Auth(userId=self.userId).create}'
+
+        r = await self.requests.post(
+            url=url,
+            headers=self.config.headers,
+        )
+
+        return
+
+    @property
+    def host(self):
+        return self.config.host
+
+    @property
+    def userId(self):
+        return self.config.userName

automon/integrations/swimlaneWrapper/config.py

@@ -1,2 +1,60 @@
+from automon import log
+from automon.helpers import environ
+
+logger = log.logging.getLogger(__name__)
+logger.setLevel(log.DEBUG)
+
+
 class SwimlaneConfig(object):
-    pass
+
+    def __init__(
+            self,
+            host: str = None,
+            userName: str = None,
+            password: str = None,
+            apiKey: str = None,
+            jwt_token: str = None,
+    ):
+        self.host = host or environ('SWIMLANE_HOST')
+        self.userName = userName or environ('SWIMLANE_USERID')
+        self.password = password or environ('SWIMLANE_PASSWORD')
+        self.apiKey = apiKey or environ('SWIMLANE_APIKEY')
+        self.jwt_token = jwt_token or environ('SWIMLANE_JWT_TOKEN')
+
+        creds = {'userName': 'USERNAME', 'password': 'PASSWORD'}
+
+    @property
+    def credentials(self):
+        return {
+            'userName': self.userName,
+            'password': self.password,
+        }
+
+    @property
+    def token(self):
+        return self.apiKey or self.jwt_token
+
+    @property
+    def headers(self):
+        if self.token:
+            return {
+                'Authorization': f'Bearer {self.token}'
+            }
+
+    async def is_ready(self) -> bool:
+        if self.host:
+            if self.userName and self.password:
+                return True
+            if self.apiKey:
+                return True
+            if self.jwt_token:
+                return True
+
+        logger.error(str(dict(
+            host=self.host,
+            userName=self.userName,
+            password=self.password,
+            apiKey=self.apiKey,
+            jwt_token=self.jwt_token,
+        )))
+        return False

automon/integrations/swimlaneWrapper/tests/test_auth.py

@@ -0,0 +1,19 @@
+import unittest
+import asyncio
+
+from automon.integrations.swimlaneWrapper.client import SwimlaneClientRest
+
+client = SwimlaneClientRest()
+client.config.userName = 'soc2'
+
+
+class MyTestCase(unittest.TestCase):
+    def test_something(self):
+        if asyncio.run(client.is_ready()):
+            self.assertTrue(
+                asyncio.run(client.login())
+            )
+
+
+if __name__ == '__main__':
+    unittest.main()