diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9f7550b --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +__pycache__ +.venv diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..f229360 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +requests diff --git a/test.sh b/test.sh new file mode 100755 index 0000000..9dd5158 --- /dev/null +++ b/test.sh @@ -0,0 +1,29 @@ +#!/bin/sh -e +usage() { + echo "Usage: $0 [-h]" 1>&2; + echo " -h Display this help message." 1>&2; + echo " -i Initialize virtual environment." 1>&2; + exit 1; +} + +while getopts "hi" opt; do + case $opt in + i) + i=1 + ;; + *) + usage + ;; + esac +done + +PYTHON=python3.9 + +if [ -n "$i" ]; then + $PYTHON -m venv .venv +fi +source .venv/bin/activate +if [ -n "$i" ]; then + $PYTHON -m pip install -r requirements.txt +fi +$PYTHON -m unittest diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..300c2f1 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +from . import test_vtwsclib diff --git a/tests/test_vtwsclib.py b/tests/test_vtwsclib.py new file mode 100644 index 0000000..bf95d68 --- /dev/null +++ b/tests/test_vtwsclib.py @@ -0,0 +1,25 @@ +import os +import unittest + +from vtwsclib import Vtapi + + +class TestVtapi(unittest.TestCase): + @classmethod + def setUpClass(cls): + if not all(key in os.environ for key in ['VTIGER_HOST', 'VTIGER_USER', 'VTIGER_PASS']): + cls.skipTest("environment variables not configured") + + def test_login(self): + with Vtapi(os.environ['VTIGER_HOST']) as api: + api.login(os.environ['VTIGER_USER'], os.environ['VTIGER_PASS']) + + def test_count(self): + with Vtapi(os.environ['VTIGER_HOST']) as api: + api.login(os.environ['VTIGER_USER'], os.environ['VTIGER_PASS']) + self.assertGreater(api.count('Quotes'), 0) + + def test_retrieve(self): + with Vtapi(os.environ['VTIGER_HOST']) as api: + api.login(os.environ['VTIGER_USER'], os.environ['VTIGER_PASS']) + self.assertIn('website', api.retrieve('CompanyDetails')[0]) diff --git a/vtwsclib.py b/vtwsclib.py index 6b3c7bf..8e5f2e0 100644 --- a/vtwsclib.py +++ b/vtwsclib.py @@ -2,16 +2,15 @@ import json import logging -import urllib3 +import requests -_http = urllib3.PoolManager() _logger = logging.getLogger(__name__) class Vtapi: def __init__(self, url): - self.session = None - self.user_id = None + self.session = requests.Session() + self.session_name = None self.url = url + '/webservice.php' def __enter__(self): @@ -23,82 +22,103 @@ def __exit__(self, type, value, traceback): finally: pass + def count(self, module): + query = f"select count(*) from {module};" + result = self.query(query) + return int(result[0]['count']) + def create(self, module, values): - fields = { + data = { 'operation': 'create', - 'sessionName': self.session, + 'sessionName': self.session_name, 'elementType': module, 'element': json.dumps(values), } - return self._request('POST', fields) + response = self.session.post(self.url, data=data) + return self._result(response) def download(self, id): - fields = { + params = { 'operation': 'download', - 'sessionName': self.session, + 'sessionName': self.session_name, 'id': id, } - return self._request('GET', fields) + response = self.session.get(self.url, params=params) + return self._result(response) def listtypes(self): - fields = { + params = { 'operation': 'listtypes', - 'sessionName': self.session, + 'sessionName': self.session_name, } - return self._request('GET', fields) + response = self.session.get(self.url, params=params) + return self._result(response) def login(self, username, accesskey): token = self._getchallenge(username) - hasher = hashlib.md5() - hasher.update(token.encode('utf-8')) - hasher.update(accesskey.encode('utf-8')) - fields = { - 'operation': 'login', - 'username': username, - 'accessKey': hasher.hexdigest(), - } - result = self._request('POST', fields=fields) - self.session, self.user_id = result['sessionName'], result['userId'] + self.session_name = self._login(username, token, accesskey) def logout(self): - if self.session: - fields = { + if self.session_name: + data = { 'operation': 'logout', - 'sessionName': self.session, + 'sessionName': self.session_name, } - self._request('POST', fields) - self.session = None - self.user_id = None + self.session.post(self.url, data=data) + self.session_name = None def query(self, query): - fields = { + params = { 'operation': 'query', - 'sessionName': self.session, + 'sessionName': self.session_name, 'query': query, } - return self._request('GET', fields) - - def retrieve(self, type, limit, offset): - query = 'select * from {} limit {}, {};'.format(type, offset, limit), + try: + response = self.session.get(self.url, params=params) + return self._result(response) + except: + _logger.error("failed to query '%s'", query) + raise + + def retrieve(self, module, limit=0, offset=0): + query = f"select * from {module};" + if limit or offset: + query = query[:-1] + f" limit {offset}, {limit};" return self.query(query) def _getchallenge(self, username): - fields = { + params = { 'operation': 'getchallenge', 'username': username, } - result = self._request('GET', fields) - return result['token'] + response = self.session.get(self.url, params=params) + result = self._result(response) + token = result['token'] + return token - def _request(self, method, fields): - _logger.debug('request %s', fields) - response = _http.request(method, self.url, fields=fields, retries=False) - return self._result(response) + def _login(self, username, token, accesskey): + hasher = hashlib.md5() + hasher.update(token.encode('utf-8')) + hasher.update(accesskey.encode('utf-8')) + data = { + 'operation': 'login', + 'username': username, + 'accessKey': hasher.hexdigest(), + } + response = self.session.post(self.url, data=data) + body = self._result(response) + return body['sessionName'] def _result(self, response): - data = response.data.decode('utf-8') - _logger.debug('response %s', data) - body = json.loads(data) + body = response.json() if not body['success']: - raise Exception(body['error']['message']) + raise VtapiError(**body['error'], status=response.status_code) return body['result'] + + +class VtapiError(Exception): + def __init__(self, code, message, status=None): + self.code = code + self.message = message + self.status = status + super().__init__(f"{self.code}: {self.message}")