Reorg code and tests
DON'T use main.py
This commit is contained in:
@@ -1,3 +1,4 @@
|
||||
config.ini
|
||||
.venv/
|
||||
__pycache__/
|
||||
config-mira.ini
|
||||
|
||||
Generated
+1
@@ -3,6 +3,7 @@
|
||||
<component name="NewModuleRootManager" inherit-compiler-output="true">
|
||||
<exclude-output />
|
||||
<content url="file://$MODULE_DIR$">
|
||||
<sourceFolder url="file://$MODULE_DIR$/integration_tests" isTestSource="true" />
|
||||
<sourceFolder url="file://$MODULE_DIR$/tests" isTestSource="true" />
|
||||
<excludeFolder url="file://$MODULE_DIR$/.venv" />
|
||||
</content>
|
||||
|
||||
+20
@@ -0,0 +1,20 @@
|
||||
<component name="ProjectRunConfigurationManager">
|
||||
<configuration default="false" name="Integrationtests" type="tests" factoryName="Unittests">
|
||||
<module name="ActiveCollabAPI" />
|
||||
<option name="ENV_FILES" value="" />
|
||||
<option name="INTERPRETER_OPTIONS" value="" />
|
||||
<option name="PARENT_ENVS" value="true" />
|
||||
<option name="SDK_HOME" value="$PROJECT_DIR$/.venv/bin/python" />
|
||||
<option name="SDK_NAME" value="Python 3.10 (ActiveCollabAPI)" />
|
||||
<option name="WORKING_DIRECTORY" value="" />
|
||||
<option name="IS_MODULE_SDK" value="false" />
|
||||
<option name="ADD_CONTENT_ROOTS" value="true" />
|
||||
<option name="ADD_SOURCE_ROOTS" value="true" />
|
||||
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
|
||||
<option name="_new_pattern" value="""" />
|
||||
<option name="_new_additionalArguments" value=""discover integration_tests"" />
|
||||
<option name="_new_target" value="""" />
|
||||
<option name="_new_targetType" value=""PATH"" />
|
||||
<method v="2" />
|
||||
</configuration>
|
||||
</component>
|
||||
Generated
+1
-1
@@ -12,7 +12,7 @@
|
||||
<option name="ADD_SOURCE_ROOTS" value="true" />
|
||||
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
|
||||
<option name="_new_pattern" value="""" />
|
||||
<option name="_new_additionalArguments" value="""" />
|
||||
<option name="_new_additionalArguments" value=""discover tests"" />
|
||||
<option name="_new_target" value="""" />
|
||||
<option name="_new_targetType" value=""PATH"" />
|
||||
<method v="2" />
|
||||
|
||||
@@ -5,14 +5,12 @@ import json
|
||||
import requests
|
||||
from requests import Response
|
||||
|
||||
from ActiveCollabAPI import AC_LOGIN_BASE_URL, AC_USER_AGENT
|
||||
from ActiveCollabAPI import AC_USER_AGENT
|
||||
|
||||
|
||||
class ACAuthenticator:
|
||||
base_url = ""
|
||||
|
||||
user = None
|
||||
accounts = None
|
||||
base_url: str = ""
|
||||
|
||||
def __init__(self, base_url: str):
|
||||
self.base_url = base_url
|
||||
@@ -32,24 +30,3 @@ class ACAuthenticator:
|
||||
return requests.post(self.base_url,
|
||||
data=json.dumps(login_data),
|
||||
headers=self.headers())
|
||||
|
||||
def login_with_check(self, email: str, password: str):
|
||||
res = self.login(email, password)
|
||||
|
||||
if res.status_code != 200:
|
||||
raise Exception('Login failed!')
|
||||
|
||||
login_res = res.json()
|
||||
if login_res['is_ok'] != 1:
|
||||
raise Exception('Login failed! (2)')
|
||||
|
||||
self.user = login_res['user']
|
||||
self.accounts = login_res['accounts']
|
||||
return res
|
||||
|
||||
def find_account(self, value, key='user_display_name'):
|
||||
found = list(filter(lambda a: a[key] == value,
|
||||
self.accounts))
|
||||
if len(found) == 0:
|
||||
raise Exception("Account '%s=%s' not found!" % (key, value))
|
||||
return found[0]
|
||||
|
||||
@@ -1,39 +1,18 @@
|
||||
import json
|
||||
|
||||
import requests
|
||||
from requests import Response
|
||||
|
||||
from ActiveCollabAPI import AC_USER_AGENT, AC_API_VERSION, AC_API_CLIENT_NAME, AC_API_CLIENT_VENDOR
|
||||
|
||||
|
||||
class ACTokenAuthenticator:
|
||||
|
||||
account = None
|
||||
user = None
|
||||
base_url = ""
|
||||
token = None
|
||||
|
||||
def __init__(self, account, user):
|
||||
self.account = account
|
||||
self.user = user
|
||||
#
|
||||
self.base_url = self.account['url'] + '/api/v%s' % AC_API_VERSION
|
||||
def __init__(self, base_url: str):
|
||||
self.base_url = base_url
|
||||
|
||||
def issue_token_intent(self):
|
||||
# use intent to get a token
|
||||
data = {
|
||||
'intent': self.user['intent'],
|
||||
'client_name': AC_API_CLIENT_NAME,
|
||||
'client_vendor': AC_API_CLIENT_VENDOR
|
||||
}
|
||||
r = requests.post(self.base_url+ '/issue-token-intent',
|
||||
headers=self.headers(),
|
||||
data=json.dumps(data))
|
||||
if r.status_code != 200:
|
||||
raise Exception('Error in issuing token intent!')
|
||||
token_res = r.json()
|
||||
if token_res['is_ok'] != 1:
|
||||
raise Exception('Error in issuing token (2)')
|
||||
return token_res['token']
|
||||
|
||||
def headers(self):
|
||||
return {
|
||||
@@ -41,3 +20,14 @@ class ACTokenAuthenticator:
|
||||
'Accept': 'application/json',
|
||||
'User-Agent': AC_USER_AGENT
|
||||
}
|
||||
|
||||
def issue_token_intent(self, intent: str) -> Response:
|
||||
# use intent to get a token
|
||||
data = {
|
||||
'intent': intent,
|
||||
'client_name': AC_API_CLIENT_NAME,
|
||||
'client_vendor': AC_API_CLIENT_VENDOR
|
||||
}
|
||||
return requests.post(self.base_url+ '/issue-token-intent',
|
||||
headers=self.headers(),
|
||||
data=json.dumps(data))
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class AcAccount:
|
||||
name: int
|
||||
url: str
|
||||
display_name: str
|
||||
user_display_name: str
|
||||
position: int
|
||||
class_: str
|
||||
status: str
|
||||
@@ -0,0 +1,13 @@
|
||||
from dataclasses import dataclass
|
||||
|
||||
from ActiveCollabAPI.AcAccount import AcAccount
|
||||
from ActiveCollabAPI.AcToken import AcToken
|
||||
from ActiveCollabAPI.AcUser import AcUser
|
||||
|
||||
|
||||
@dataclass
|
||||
class AcSession:
|
||||
user: AcUser
|
||||
accounts: [AcAccount]
|
||||
cur_account: AcAccount
|
||||
token: AcToken
|
||||
@@ -0,0 +1,6 @@
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class AcToken:
|
||||
token: str
|
||||
@@ -0,0 +1,9 @@
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass
|
||||
class AcUser:
|
||||
avatar_url: str
|
||||
first_name: str
|
||||
last_name: str
|
||||
intent: str
|
||||
@@ -0,0 +1,62 @@
|
||||
from dataclasses import dataclass
|
||||
|
||||
from ActiveCollabAPI import AC_API_VERSION
|
||||
from ActiveCollabAPI.ACAuthenticator import ACAuthenticator
|
||||
from ActiveCollabAPI.ACTokenAuthenticator import ACTokenAuthenticator
|
||||
from ActiveCollabAPI.AcAccount import AcAccount
|
||||
from ActiveCollabAPI.AcSession import AcSession
|
||||
from ActiveCollabAPI.AcToken import AcToken
|
||||
from ActiveCollabAPI.AcUser import AcUser
|
||||
|
||||
|
||||
@dataclass
|
||||
class AcLoginResponse:
|
||||
user: AcUser
|
||||
accounts: list[AcAccount]
|
||||
|
||||
|
||||
class ActiveCollab:
|
||||
base_url: str = ""
|
||||
|
||||
def __init__(self, base_url: str):
|
||||
self.base_url = base_url
|
||||
|
||||
def login_to_first_account(self, email: str, password: str) -> AcSession:
|
||||
login_res = self.user_login(email, password)
|
||||
cur_account = self.select_first_account(login_res.accounts)
|
||||
token = self.create_token(cur_account, login_res.user)
|
||||
session = AcSession(login_res.user, login_res.accounts, cur_account, token)
|
||||
return session
|
||||
|
||||
def user_login(self, email: str, password: str) -> AcLoginResponse:
|
||||
auth = ACAuthenticator(self.base_url)
|
||||
res = auth.login(email, password)
|
||||
if res.status_code != 200:
|
||||
raise Exception('Login failed!')
|
||||
res_data = res.json()
|
||||
if res_data['is_ok'] != 1:
|
||||
raise Exception('Login failed! (2)')
|
||||
|
||||
def map_acccount(a) -> AcAccount:
|
||||
a["class_"] = a["class"]
|
||||
del a["class"]
|
||||
return AcAccount(**a)
|
||||
|
||||
accounts = list(map(lambda a: map_acccount(a), res_data['accounts']))
|
||||
return AcLoginResponse(AcUser(**res_data['user']), accounts)
|
||||
|
||||
def select_first_account(self, accounts: list[AcAccount]) -> AcAccount:
|
||||
return accounts[0]
|
||||
|
||||
def create_token(self, account: AcAccount, user: AcUser) -> AcToken:
|
||||
authenticator = ACTokenAuthenticator(account.url + '/api/v%s' % AC_API_VERSION)
|
||||
res = authenticator.issue_token_intent(user.intent)
|
||||
if res.status_code != 200:
|
||||
raise Exception('Request token failed!')
|
||||
res_data = res.json()
|
||||
if res_data['is_ok'] != 1:
|
||||
raise Exception('Request token failed! (2)')
|
||||
return AcToken(res_data["token"])
|
||||
|
||||
def get_projects(self):
|
||||
pass
|
||||
+1
-1
@@ -5,7 +5,7 @@ ENV DEBIAN_FRONTEND=noninteractive
|
||||
RUN apt-get clean \
|
||||
&& apt-get update \
|
||||
&& apt-get upgrade -y
|
||||
RUN apt-get install -y python3 python3-pip
|
||||
RUN apt-get install -y python3 python3-pip build-essential
|
||||
COPY requirements.txt /tmp/requirements.txt
|
||||
RUN pip3 install -r /tmp/requirements.txt
|
||||
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
import json
|
||||
from pprint import pprint
|
||||
from unittest import TestCase
|
||||
|
||||
from ActiveCollabAPI.ActiveCollab import ActiveCollab
|
||||
|
||||
with open("test-config.json", "r") as cfg:
|
||||
config = json.load(cfg)
|
||||
|
||||
|
||||
class Test_Integration_Login(TestCase):
|
||||
|
||||
def test_login_success(self):
|
||||
ac = ActiveCollab(config["base_url"])
|
||||
session = ac.login_to_first_account(config["username"], config["password"])
|
||||
pprint(session)
|
||||
|
||||
self.assertEqual(config["first_name"], session.user.first_name)
|
||||
self.assertEqual(config["last_name"], session.user.last_name)
|
||||
|
||||
self.assertEqual(1, len(session.accounts))
|
||||
self.assertGreater(len(session.token.token), 5)
|
||||
@@ -12,6 +12,8 @@ from ActiveCollabAPI.ACTokenAuthenticator import ACTokenAuthenticator
|
||||
# the only thing missing will be the response.body which is not logged.
|
||||
from http.client import HTTPConnection
|
||||
|
||||
from ActiveCollabAPI.ActiveCollab import ActiveCollab
|
||||
|
||||
HTTPConnection.debuglevel = 1
|
||||
|
||||
logging.basicConfig() # you need to initialize logging, otherwise you will not see anything from requests
|
||||
@@ -30,10 +32,10 @@ def main():
|
||||
account_name = config['LOGIN']['account_name']
|
||||
|
||||
# 1. you need to log in to get a session and a list of accounts
|
||||
auth = ACAuthenticator(AC_LOGIN_BASE_URL)
|
||||
auth.login_with_check(username, password)
|
||||
# use first account
|
||||
account = auth.find_account(account_name)
|
||||
ac = ActiveCollab(AC_LOGIN_BASE_URL)
|
||||
session = ac.login(username, password)
|
||||
account = ac.get_account(session) # default returning first account
|
||||
token = ac.get_token(account, session.user)
|
||||
|
||||
# 2. you need to get a token for this account
|
||||
token = ACTokenAuthenticator(account, auth.user).issue_token_intent()
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
{
|
||||
"base_url": "https://activecollab.com/api/v1/external/login",
|
||||
"username": "packend_ja.05@icloud.com",
|
||||
"password": "myqkyc-muntuh-haRti6",
|
||||
|
||||
"first_name": "packend_ja",
|
||||
"last_name": ".05@icloud.com"
|
||||
}
|
||||
@@ -7,14 +7,6 @@ from ActiveCollabAPI.ACAuthenticator import ACAuthenticator
|
||||
|
||||
|
||||
class TestACAuthenticator(TestCase):
|
||||
def test_headers(self):
|
||||
authenticator = ACAuthenticator(AC_LOGIN_BASE_URL)
|
||||
headers = authenticator.headers()
|
||||
self.assertIn('Content-Type', headers)
|
||||
self.assertEqual('application/json; charset=utf8', headers['Content-Type'])
|
||||
self.assertIn('Accept', headers)
|
||||
self.assertEqual('application/json', headers['Accept'])
|
||||
self.assertIn('User-Agent', headers)
|
||||
|
||||
def mock_login_success(self):
|
||||
return {
|
||||
@@ -37,10 +29,6 @@ class TestACAuthenticator(TestCase):
|
||||
"intent": "random-string"}
|
||||
}
|
||||
|
||||
def mock_login_failed(self):
|
||||
return {
|
||||
"is_ok": 3
|
||||
}
|
||||
|
||||
def test_login_success_gets_user_and_accounts_list(self):
|
||||
with patch("ActiveCollabAPI.ACAuthenticator.requests.post") as mock_post:
|
||||
@@ -50,7 +38,7 @@ class TestACAuthenticator(TestCase):
|
||||
email = "collab@example.com"
|
||||
password = "VeryS3cret!"
|
||||
authenticator = ACAuthenticator(AC_LOGIN_BASE_URL)
|
||||
response = authenticator.login_with_check(email, password)
|
||||
response = authenticator.login(email, password)
|
||||
|
||||
self.assertEqual(200, response.status_code)
|
||||
res_data = response.json()
|
||||
@@ -62,19 +50,13 @@ class TestACAuthenticator(TestCase):
|
||||
self.assertGreater(len(accounts), 0)
|
||||
self.assertEqual(12345678, accounts[0]["name"])
|
||||
|
||||
def test_login_failed_http401(self):
|
||||
with self.assertRaises(Exception) as exc:
|
||||
with patch("ActiveCollabAPI.ACAuthenticator.requests.post") as mock_post:
|
||||
mock_post.return_value.status_code = 401
|
||||
mock_post.return_value.json.return_value = self.mock_login_failed()
|
||||
|
||||
email = "collab@example.com"
|
||||
password = "VeryS3cret!"
|
||||
authenticator = ACAuthenticator(AC_LOGIN_BASE_URL)
|
||||
authenticator.login_with_check(email, password)
|
||||
def mock_login_failed(self):
|
||||
return {
|
||||
"is_ok": 0
|
||||
}
|
||||
|
||||
def test_login_failed_not_ok(self):
|
||||
with self.assertRaises(Exception) as exc:
|
||||
with patch("ActiveCollabAPI.ACAuthenticator.requests.post") as mock_post:
|
||||
mock_post.return_value.status_code = 200
|
||||
mock_post.return_value.json.return_value = self.mock_login_failed()
|
||||
@@ -82,4 +64,24 @@ class TestACAuthenticator(TestCase):
|
||||
email = "collab@example.com"
|
||||
password = "VeryS3cret!"
|
||||
authenticator = ACAuthenticator(AC_LOGIN_BASE_URL)
|
||||
authenticator.login_with_check(email, password)
|
||||
response = authenticator.login(email, password)
|
||||
|
||||
self.assertEqual(200, response.status_code)
|
||||
res_data = response.json()
|
||||
self.assertNotEqual(1, res_data['is_ok'])
|
||||
|
||||
|
||||
def mock_login_forbidden(self):
|
||||
return {
|
||||
}
|
||||
def test_login_forbidden(self):
|
||||
with patch("ActiveCollabAPI.ACAuthenticator.requests.post") as mock_post:
|
||||
mock_post.return_value.status_code = 401
|
||||
mock_post.return_value.json.return_value = self.mock_login_forbidden()
|
||||
|
||||
email = "collab@example.com"
|
||||
password = "VeryS3cret!"
|
||||
authenticator = ACAuthenticator(AC_LOGIN_BASE_URL)
|
||||
response = authenticator.login(email, password)
|
||||
|
||||
self.assertEqual(401, response.status_code)
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
from unittest import TestCase
|
||||
from unittest.mock import patch
|
||||
|
||||
from ActiveCollabAPI.ACTokenAuthenticator import ACTokenAuthenticator
|
||||
|
||||
|
||||
class TestACTokenAuthenticator(TestCase):
|
||||
|
||||
def mock_token_success(self):
|
||||
return {
|
||||
"is_ok": 1,
|
||||
"token": "abcdef"
|
||||
}
|
||||
|
||||
def test_issue_token_intent_success(self):
|
||||
with patch("ActiveCollabAPI.ACTokenAuthenticator.requests.post") as mock_post:
|
||||
mock_post.return_value.status_code = 200
|
||||
mock_post.return_value.json.return_value = self.mock_token_success()
|
||||
|
||||
auth = ACTokenAuthenticator("http://mock")
|
||||
res = auth.issue_token_intent("this-is-the-token-intent-string")
|
||||
|
||||
self.assertEqual(200, res.status_code)
|
||||
res_data = res.json()
|
||||
self.assertEqual(1, res_data["is_ok"])
|
||||
|
||||
def mock_token_not_ok(self):
|
||||
return {
|
||||
"is_ok": 0,
|
||||
"token": "abcdef"
|
||||
}
|
||||
|
||||
def test_issue_token_intent_not_ok(self):
|
||||
with patch("ActiveCollabAPI.ACTokenAuthenticator.requests.post") as mock_post:
|
||||
mock_post.return_value.status_code = 200
|
||||
mock_post.return_value.json.return_value = self.mock_token_not_ok()
|
||||
|
||||
auth = ACTokenAuthenticator("http://mock")
|
||||
res = auth.issue_token_intent("this-is-the-token-intent-string")
|
||||
|
||||
self.assertEqual(200, res.status_code)
|
||||
res_data = res.json()
|
||||
self.assertNotEqual(1, res_data["is_ok"])
|
||||
|
||||
def mock_token_forbidden(self):
|
||||
return {
|
||||
}
|
||||
|
||||
def test_issue_token_intent_forbidden(self):
|
||||
with patch("ActiveCollabAPI.ACTokenAuthenticator.requests.post") as mock_post:
|
||||
mock_post.return_value.status_code = 401
|
||||
mock_post.return_value.json.return_value = self.mock_token_forbidden()
|
||||
|
||||
auth = ACTokenAuthenticator("http://mock")
|
||||
res = auth.issue_token_intent("this-is-the-token-intent-string")
|
||||
|
||||
self.assertEqual(401, res.status_code)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user