Add unit tests for transports

Author: kadler

tests/test_unit_transport_database.py

@@ -0,0 +1,54 @@
+from itoolkit import iToolKit
+from itoolkit.transport import DatabaseTransport
+
+
+def test_database_transport_callproc(database_callproc):
+    transport = DatabaseTransport(database_callproc)
+    tk = iToolKit()
+    transport.call(tk)
+
+    cursor = database_callproc.cursor()
+
+    cursor.callproc.assert_called_once()
+    cursor.__iter__.assert_called_once()
+
+
+def test_database_transport_execute(database_execute):
+    transport = DatabaseTransport(database_execute)
+    tk = iToolKit()
+    transport.call(tk)
+
+    cursor = database_execute.cursor()
+
+    cursor.execute.assert_called_once()
+    cursor.__iter__.assert_called_once()
+
+
+def test_database_transport_execute_schema(database_execute):
+    schema = 'MYSCHEMA'
+    transport = DatabaseTransport(database_execute, schema=schema)
+    tk = iToolKit()
+    transport.call(tk)
+
+    cursor = database_execute.cursor()
+
+    cursor.execute.assert_called_once()
+    cursor.__iter__.assert_called_once()
+
+    assert len(cursor.execute.call_args[0]) > 0
+    assert schema in cursor.execute.call_args[0][1]  # argument 0 is self
+
+
+def test_database_transport_callproc_schema(database_execute):
+    schema = 'MYSCHEMA'
+    transport = DatabaseTransport(database_execute, schema=schema)
+    tk = iToolKit()
+    transport.call(tk)
+
+    cursor = database_execute.cursor()
+
+    cursor.execute.assert_called_once()
+    cursor.__iter__.assert_called_once()
+
+    assert len(cursor.execute.call_args[0]) > 0
+    assert schema in cursor.execute.call_args[0][1]  # argument 0 is self

tests/test_unit_transport_http.py

@@ -0,0 +1,102 @@
+import sys
+
+from itoolkit import iToolKit
+from itoolkit.transport import HttpTransport
+
+if sys.version_info >= (3, 0):
+    if sys.version_info >= (3, 6):
+        from urllib.parse import parse_qs
+    else:
+        from urllib.parse import urlencode
+else:
+    from urllib import urlencode
+
+XMLIN = "<?xml version='1.0'?>\n<xmlservice></xmlservice>"
+
+
+def mock_http_urlopen(mocker):
+    mock_urlopen = mocker.patch('itoolkit.transport.http.urlopen')
+    mock_response = mocker.Mock()
+    mock_response.read.side_effect = XMLIN.encode('utf-8')
+    mock_urlopen.return_value = mock_response
+
+    return mock_urlopen
+
+
+def assert_urlopen_params_correct(mock_urlopen, url, uid, pwd, db2='*LOCAL',
+                                  ipc='*na', ctl='*here *cdata',
+                                  xmlout=str(HttpTransport.OUT_SIZE)):
+    # assert_called_once only available in Python 3.6+
+    if sys.version_info >= (3, 6):
+        mock_urlopen.assert_called_once()
+
+        args, kwargs = mock_urlopen.call_args
+
+        assert len(kwargs) == 0
+        assert len(args) == 2
+
+        assert args[0] == url
+
+        data = {key.decode('utf-8'): value[0].decode('utf-8') for (key, value)
+                in parse_qs(args[1]).items()}
+
+        exp_data = dict(
+            uid=uid,
+            pwd=pwd,
+            db2=db2,
+            ipc=ipc,
+            ctl=ctl,
+            xmlout=xmlout,
+            xmlin=XMLIN + "\n",
+        )
+        assert data == exp_data
+    else:
+        mock_urlopen.assert_called_once_with(url, urlencode({
+                'db2': db2,
+                'uid': uid,
+                'pwd': pwd,
+                'ipc': ipc,
+                'ctl': ctl,
+                'xmlin': XMLIN + "\n",
+                'xmlout': int(xmlout)
+        }).encode("utf-8"))
+
+
+def test_http_transport_minimal(mocker):
+    mock_urlopen = mock_http_urlopen(mocker)
+
+    url = 'http://example.com/cgi-bin/xmlcgi.pgm'
+    user = 'dummy'
+    password = 'passw0rd'
+
+    transport = HttpTransport(url, user, password)
+    tk = iToolKit()
+    transport.call(tk)
+
+    assert_urlopen_params_correct(
+        mock_urlopen,
+        url,
+        uid=user,
+        pwd=password,
+    )
+
+
+def test_http_transport_with_database(mocker):
+    mock_urlopen = mock_http_urlopen(mocker)
+
+    url = 'http://example.com/cgi-bin/xmlcgi.pgm'
+    user = 'dummy'
+    password = 'passw0rd'
+    database = 'MYDB'
+
+    transport = HttpTransport(url, user, password, database=database)
+    tk = iToolKit()
+    transport.call(tk)
+
+    assert_urlopen_params_correct(
+        mock_urlopen,
+        url,
+        uid=user,
+        pwd=password,
+        db2=database
+    )

tests/test_unit_transport_ssh.py

@@ -0,0 +1,29 @@
+from itoolkit import iToolKit
+from itoolkit.transport import SshTransport
+
+
+def mock_ssh(mocker):
+    client = mocker.MagicMock()
+
+    channels = (
+        mocker.MagicMock(),
+        mocker.MagicMock(closed=True),
+        mocker.MagicMock(closed=True),
+    )
+
+    client.channels = channels
+
+    client.exec_command.return_value = channels
+
+    return client
+
+
+def test_ssh_transport_minimal(mocker):
+    ssh_client = mock_ssh(mocker)
+
+    transport = SshTransport(ssh_client)
+    tk = iToolKit()
+    transport.call(tk)
+
+    command = "/QOpenSys/pkgs/bin/xmlservice-cli"
+    ssh_client.exec_command.assert_called_once_with(command)