DB2

Details

ID	`DB2`
Provider	IBM
Category	Database
From Version	6.0.0
Docker Image	`demisto/ibm-db2:1.0.0.9067966`
Supported Modules	Agentix XSIAM

README

Integration to provide connectivity to IBM DB2 using the python ibm_db2 library.
This integration was integrated and tested with version 0.1 of DB2

Configure DB2 in Cortex

Parameter	Required
Database host	True
Port	False
Database Name	False
Username	True
Password	True
Connection Arguments (ex: arg1=val1&arg2=val2)	False
Use an SSL connection	False
Use Persistent Connection	False

Commands

You can execute these commands from the CLI, as part of an automation, or in a playbook.
After you successfully execute a command, a DBot message appears in the War Room with the command details.

db2-query

Running a DB2 query command

Base Command

db2-query

Input

Argument Name	Description	Required
query	The DB2 query to run.	Required
limit	The maximum number of results. Default is 50.	Optional
skip	The offset at which to start the result. Default is 0.	Optional
bind_variables_name	A comma separated list of names which will be replaced in query having ‘:<name>’.	Optional
bind_variables_values	A comma separated list of values corresponding to `bind_variables_name` or values in order of ‘?’ mark to be replaced in query.	Optional

Context Output

There is no context output for this command.

Command Example

!db2-query query="CREATE TABLE publishers(publisher_id INT GENERATED BY DEFAULT AS IDENTITY NOT NULL, name VARCHAR(255) NOT NULL, PRIMARY KEY(publisher_id))"

Human Readable Output

Configuration parameters

host — Database host (required)
port — Port
dbname — Database Name
credentials — Username (required)
connect_parameters — Connection Arguments (ex: arg1=val1&arg2=val2)
ssl_connect — Use an SSL connection
use_persistent — Use Persistent Connection

Commands (1)

db2-query
Running a DB2 query command

Source

DB2.py DB2_test.py DB2.yml DB2_description.md

Download History Raw

import ibm_db
import pytest
from CommonServerPython import *
from DB2 import Client, query_command
from ibm_db_dbi import Connection

ARGS1 = {
    "query": (
        "CREATE TABLE publishers( "
        "publisher_id INT GENERATED BY DEFAULT AS IDENTITY NOT NULL, "
        "name VARCHAR(255) NOT NULL, "
        "PRIMARY KEY(pubisher_id))"
    )
}
ARGS2 = {
    "query": "SELECT * FROM publishers",
    "limit": 1,
    "skip": 0,
}
ARGS3 = {
    "query": "INSERT INTO publishers VALUES(:id, :name)",
    "bind_variables_name": "id, name",
    "bind_variables_values": "2, larry",
}
ARGS4 = {"query": "SELECT * FROM publishers", "limit": 1, "skip": 0}
ARGS5 = {"query": "INSERT INTO publishers VALUES(?, ?)", "bind_variables_values": "1, john smith"}
ARGS6 = {"query": "SELECT * FROM publishers", "limit": 2, "skip": 0}

EMPTY_HEADERS = []
PUBLISHER_HEADERS = ["id", "name"]

RES1235 = []
RES4 = [{"id": 2, "name": "larry"}]
RES6 = [{"id": 2, "name": "larry"}, {"id": 1, "name": "john smith"}]

EXPECTED_OUTPUT1 = {
    "DB2(val.Query && val.Query === obj.Query": {
        "DB2": {
            "Result": [],
            "Query": (
                "CREATE TABLE publishers( "
                "publisher_id INT GENERATED BY DEFAULT AS IDENTITY NOT NULL, "
                "name VARCHAR(255) NOT NULL, "
                "PRIMARY KEY(pubisher_id))"
            ),
            "DbName": "sample",
        }
    }
}
EXPECTED_OUTPUT2 = {"Result": [], "Query": "SELECT * FROM publishers", "DbName": "sample"}
EXPECTED_OUTPUT3 = {"Result": [], "Query": "INSERT INTO publishers VALUES(:id, :name)", "DbName": "sample"}
EXPECTED_OUTPUT4 = {"Result": [{"id": "2", "name": "larry"}], "Query": "SELECT * FROM publishers", "DbName": "sample"}
EXPECTED_OUTPUT5 = {"Result": [], "Query": "INSERT INTO publishers VALUES(?, ?)", "DbName": "sample"}
EXPECTED_OUTPUT6 = {
    "Result": [{"id": "2", "name": "larry"}, {"id": "1", "name": "john smith"}],
    "Query": "SELECT * FROM publishers",
    "DbName": "sample",
}


# ====== Connection Fixture =========
@pytest.fixture
def client(mocker):
    """
    generate client object
    """
    mocker.patch.object(ibm_db, "get_db_info", return_value="test")
    mocker.patch.object(ibm_db, "connect", return_value=Connection)
    mocker.patch.object(ibm_db, "close", return_value=True)
    client = Client("localhost", "username", "password", "8080", "sample", False)
    return client


@pytest.mark.parametrize(
    "command, args, response, headers, expected_result",
    [
        (query_command, ARGS1, None, EMPTY_HEADERS, None),
        (query_command, ARGS2, None, EMPTY_HEADERS, None),
        (query_command, ARGS3, None, EMPTY_HEADERS, None),
        (query_command, ARGS5, None, EMPTY_HEADERS, None),
    ],
)
def test_create_insert_empty_table_query_commands(command, args, response, headers, expected_result, mocker, client):
    """
    Test create table command into empty table.
    """
    mocker.patch.object(client, "execute_query", side_effect=Exception("No results found"))
    mocker.patch.object(demisto, "error")
    mocker.patch.object(demisto, "info")
    with pytest.raises(DemistoException):
        command(client, args)


@pytest.mark.parametrize(
    "command, args, response, headers, expected_result",
    [
        (query_command, ARGS4, RES4, PUBLISHER_HEADERS, EXPECTED_OUTPUT4),
        (query_command, ARGS6, RES6, PUBLISHER_HEADERS, EXPECTED_OUTPUT6),
    ],
)
def test_fetch_query_commands(command, args, response, headers, expected_result, mocker, client):
    """
    Test create table command
    """
    mocker.patch.object(client, "execute_query", return_value=(response, headers))
    mocker.patch.object(demisto, "error")
    mocker.patch.object(demisto, "info")
    result = command(client, args)
    assert result.raw_response == response
    assert result.outputs == expected_result