Run tests with GitHub Actions.

.github/workflows/ci.yml

@@ -0,0 +1,30 @@
+name: ci
+on:
+  pull_request:
+  push:
+    branches: [main]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.9"]
+        vtiger-version: ["8.2.0"]
+    services:
+      vtiger:
+        image: vtigercrm/vtigercrm-${{ matrix.vtiger-version }}
+        env:
+          mysql_user: admin
+          mysql_pass: admin
+        ports:
+          - "80:80"
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+      - run: python -m unittest
+        env:
+          VTIGER_HOST: http://localhost
+          VTIGER_USER: admin
+          VTIGER_PASS: admin

.gitignore

@@ -0,0 +1 @@
+__pycache__

test.sh

@@ -0,0 +1,17 @@
+#!/bin/sh -e
+usage() {
+    echo "Usage: $0 [-h]" 1>&2;
+    echo "  -h Display this help message." 1>&2;
+    exit 1;
+}
+
+while getopts "hi" opt; do
+  case $opt in
+    *)
+      usage
+      ;;
+  esac
+done
+
+PYTHON=python3.9
+$PYTHON -m unittest

tests/__init__.py

@@ -0,0 +1 @@
+from . import test_vlamb

tests/test_vlamb.py

@@ -0,0 +1,30 @@
+import os
+import unittest
+
+from vlamb import Vtapi, VtapiError
+
+
+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_login_fail(self):
+        with Vtapi(os.environ['VTIGER_HOST']) as api, self.assertRaises(VtapiError) as fail:
+            api.login(os.environ['VTIGER_USER'], 'bogus')
+        self.assertEqual("INVALID_USER_CREDENTIALS: Invalid username or password", str(fail.exception))
+
+    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])

vlamb.py

@@ -0,0 +1,131 @@
+import hashlib
+import json
+import logging
+import urllib.parse
+import urllib.request
+
+_logger = logging.getLogger(__name__)
+
+
+class Vtapi:
+    def __init__(self, url):
+        self.session_name = None
+        self.url = url + '/webservice.php'
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        try:
+            self.logout()
+        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):
+        data = {
+            'operation': 'create',
+            'sessionName': self.session_name,
+            'elementType': module,
+            'element': json.dumps(values),
+        }
+        with self._urlopen(self.url, data=data) as response:
+            return self._result(response)
+
+    def download(self, id):
+        params = {
+            'operation': 'download',
+            'sessionName': self.session_name,
+            'id': id,
+        }
+        with self._urlopen(self.url, params=params) as response:
+            return self._result(response)
+
+    def listtypes(self):
+        params = {
+            'operation': 'listtypes',
+            'sessionName': self.session_name,
+        }
+        with self._urlopen(self.url, params=params) as response:
+            return self._result(response)
+
+    def login(self, username, accesskey):
+        token = self._getchallenge(username)
+        self.session_name = self._login(username, token, accesskey)
+
+    def logout(self):
+        if self.session_name:
+            data = {
+                'operation': 'logout',
+                'sessionName': self.session_name,
+            }
+            self._urlopen(self.url, data=data)
+            self.session_name = None
+
+    def query(self, query):
+        params = {
+            'operation': 'query',
+            'sessionName': self.session_name,
+            'query': query,
+        }
+        try:
+            with self._urlopen(self.url, params=params) as response:
+                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):
+        params = {
+            'operation': 'getchallenge',
+            'username': username,
+        }
+        with self._urlopen(self.url, params=params) as response:
+            result = self._result(response)
+        token = result['token']
+        return token
+
+    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(),
+        }
+        with self._urlopen(self.url, data=data) as response:
+            body = self._result(response)
+        return body['sessionName']
+
+    def _result(self, response):
+        body = json.loads(response.read().decode('utf-8'))
+        if not body['success']:
+            raise VtapiError(**body['error'], status=response.status)
+        return body['result']
+
+    def _urlopen(self, url, data=None, params=None):
+        if params:
+            url = url + '?' + urllib.parse.urlencode(params)
+        if data:
+            data = urllib.parse.urlencode(data).encode('utf-8')
+        return urllib.request.urlopen(url, data=data)
+
+
+
+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}")